[PR] [FLINK-40398][connector/kafka] Inject a clock for retention expiry [flink-connector-kafka]

["MartijnVisser (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40398][connector/kafka] Inject a clock for retention expiry [flink-connector-kafka]

["spuru9 (via GitHub)" <gi...@apache.org>]

spuru9 commented on code in PR #320:
URL: https://github.com/apache/flink-connector-kafka/pull/320#discussion_r4136384215


##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -123,7 +128,32 @@ public DynamicKafkaSourceReader(
             KafkaRecordDeserializationSchema<T> deserializationSchema,
             Properties properties,
             OffsetsInitializer startingOffsetsInitializer) {
+        this(
+                readerContext,
+                deserializationSchema,
+                properties,
+                startingOffsetsInitializer,
+                SystemClock.getInstance());
+    }
+
+    /**
+     * Creates a reader that reads the current time through the given {@link Clock}. Removed cluster
+     * retention deadlines are absolute timestamps that are written into checkpoints, so the clock
+     * must report absolute time; production always passes {@link SystemClock#getInstance()}. Tests
+     * pass a manual clock to drive retention expiry instead of racing the wall clock.
+     *
+     * <p>Package-private rather than annotated {@code @VisibleForTesting}: the annotation is itself
+     * non-public Flink API and would add a frozen ArchUnit violation, which {@code
+     * archunit.properties} asks contributors to avoid.
+     */

Review Comment:
   ```suggestion
       /**
        * Creates a reader that reads the current time through the given {@link Clock}. Removed cluster
        * retention deadlines are absolute timestamps that are written into checkpoints, so the clock
        * must report absolute time; production always passes {@link SystemClock#getInstance()}.
        */
   ```
   ArchUnit paragraph explained in the PR, not needed in the code,



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org