MartijnVisser opened a new pull request, #320:
URL: https://github.com/apache/flink-connector-kafka/pull/320
## What is the purpose of the change
`DynamicKafkaSourceReaderTest.testHandleSourceEventRetainsRemovedClusterOffsetsUntilExpired`
is
flaky (FLINK-40398). The reader stamps a removed cluster's retention
deadline from
`System.currentTimeMillis()` and prunes against it in `snapshotState`, and
the test gives it a one
second window, so the assertions have to win a race against the wall clock.
The window is not filled by scheduler jitter. `handleSourceEvents` stamps
the deadline and then,
in the same call, closes both `KafkaSourceReader`s and recreates the
surviving one. Timed locally
that call takes 25 ms on an idle machine but 571, 575 and 579 ms in three of
six runs under CPU
contention, so on a loaded runner it crosses the second and the split is
pruned before the
assertions read it. The `Time elapsed: 1.199 s` in the report fits that and
rules out a GC pause
longer than a second inside the method.
The retention logic is correct. What is missing is a way for a test to
control time, so this gives
the reader an injectable clock.
## Brief change log
- `DynamicKafkaSourceReader` takes a `Clock` through a new package-private
constructor and reads
it in the three places that used `System.currentTimeMillis()`. Both public
constructors pass
`SystemClock.getInstance()`, so production behaviour is unchanged, and
deadlines stay absolute
timestamps because they are written into checkpoints.
- The flaky test becomes four that drive a `ManualClock` and assert the
exact expiry boundary:
retained up to the millisecond before the deadline and dropped at it,
reactivated when the
cluster returns before the deadline, not reactivated after it, and a
restored split retained
until its own deadline. The last two are new coverage.
- The frozen ArchUnit store gains no violation. One existing entry moves to
the new constructor
signature, 59 entries before and after. The constructor is not annotated
`@VisibleForTesting`,
because that annotation is itself non public API and would add an entry.
## Verifying this change
This change added tests and can be verified as follows. All runs are local,
on 6b99bb38 and on
this branch.
| Tree | Test | Condition | Result |
|---|---|---|---|
| base | old test | `Thread.sleep(1_100)` after the removal event | red 3/3,
`NoSuchElementException: No value present` at
`DynamicKafkaSourceReaderTest.java:294`, the same frame as CI |
| base | old test | CPU contention, no sleep | 0/30, the condition never
occurred locally |
| head | new test 1 | the same `Thread.sleep(1_100)` | green |
| head | the three clock reads reverted | none | red 3/4, so the tests pin
the wiring and not only the timing |
| head | `DynamicKafkaSourceReaderTest` | none, and under CPU contention |
green 18/18, and 20/20 over 20 sequential runs |
| head | the four new tests | `@RepeatedTest(500)` | green 2000/2000 in 13.3
s |
The unforced loop reproduces nothing locally, as expected: this test failed
twice in roughly 180
CI executions since 2026-08-16, which would need about 300 local runs of a
class that starts two
Kafka brokers. The forced condition carries the evidence.
Also run: `DynamicKafkaSourceReaderIdlenessTest`,
`DynamicKafkaSourceReaderPauseResumeTest`,
`*ArchitectureTest` with a clean `archunit-violations` afterwards,
`spotless:check`,
`checkstyle:check` and the module's unit test phase (441/441).
`testActiveSplitCountMetricTracksMetadataRemoval` keeps the real clock on
purpose: it only asserts
that retained state is present, over a 60 s window, never a deadline.
`DynamicKafkaSourceEnumerator` reads the wall clock the same way and
`DynamicKafkaSourceEnumeratorTest.testEnumeratorStateRetainsRemovedClusterUntilExpired`
has the
same one second window. It has not been seen failing, so it gets its own
ticket rather than being
pulled into this one.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**. `flink-core`
is already a
`provided` dependency and `Clock`, `SystemClock` and `ManualClock` are
`@PublicEvolving` in
its main jar.
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: **no**
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing,
Kubernetes/Yarn, ZooKeeper: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no**
- If yes, how is the feature documented? **not applicable**
## AI Disclosure
- [x] This pull request was created with the assistance of an AI tool.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
