Dale Lane created FLINK-40620:
---------------------------------
Summary: SourceTopicIntegrityTest.testTopicIntegritySuccess flaky
- asserts exact record counts against at-least-once sink
Key: FLINK-40620
URL: https://issues.apache.org/jira/browse/FLINK-40620
Project: Flink
Issue Type: Bug
Components: Connectors / Kafka
Reporter: Dale Lane
{{SourceTopicIntegrityTest.testTopicIntegritySuccess}} fails intermittently in
CI with
{{java.util.concurrent.TimeoutException: Timeout waiting for 100 records in
topic SourceTopicIntegrityTest_sink-topic after PT2M}}
The failure is not the test running out of time. The polling log shows the sink
topic held *200* records throughout, while the test waited for {*}100{*}:
{{Found 200 records in topic SourceTopicIntegrityTest_sink-topic (expected: 100)
Found 200 records in topic SourceTopicIntegrityTest_sink-topic (expected: 100)
...}}
The wait is satisfied only when the count is _equal_ to the expected value, so
once the topic holds more records than expected the condition can never become
true. The two-minute timeout is then guaranteed from the first poll onwards,
however long the job is given.
The job under test builds its sink without specifying a delivery guarantee, so
it takes the {{KafkaSinkBuilder}} default of {{{}DeliveryGuarantee.NONE{}}}.
Under that guarantee a task restart legitimately re-emits records that were
already written (the 200 records I've seen on my unrelated pull request is
twice the 100 records that the test produces).
The only assertions are these two exact-count waits. It is asserting a
no-duplicates property that the sink it configures does not provide. Any event
that causes a restart (which is happening intermittently on the Github CI
workers) turns this into a failure.
I suggest splitting this into a exactly-once test (that enables exactly once
and avoids the above issue) and a at-least once test (that leaves the
at-least-once default config, but sets the assertions appropriately) so we have
coverage of both paths, plus stable tests.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
