featzhang opened a new pull request, #27883: URL: https://github.com/apache/flink/pull/27883
## What is the purpose of the change? This PR fixes the flaky test `DataGeneratorSourceITCase#testGatedRateLimiter` by eliminating a race condition in the `FirstCheckpointFilter` inner class. The root cause is that `FirstCheckpointFilter` previously stopped collecting elements inside `snapshotState()`. However, the checkpoint barrier can arrive at `FirstCheckpointFilter` before all upstream elements (emitted in the same checkpoint cycle by `GatedRateLimiter`) have been processed, causing the filter to prematurely discard elements and making the assertion `assertThat(results).hasSize(capacityPerCheckpoint)` fail non-deterministically. Fixes: FLINK-39388 ## Brief change log - **`DataGeneratorSourceITCase`**: Refactored `FirstCheckpointFilter` to implement `CheckpointListener` in addition to `CheckpointedFunction`. The element cutoff logic is moved from `snapshotState()` to `notifyCheckpointComplete()`, ensuring that collection stops only after the first checkpoint has fully completed and all in-flight elements have been processed downstream. ## Verifying this change This change is a test-only fix. The existing assertion `assertThat(results).hasSize(capacityPerCheckpoint)` continues to validate the correctness of `GatedRateLimiter`. The fix eliminates the non-deterministic timing window that caused intermittent failures. - [x] This change is a test-only fix. - [x] The fix has been verified by code review of the race condition. ## Does this pull request potentially affect one of the following parts? - **Dependencies** (does it add or upgrade a dependency): No - **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 the Flink SQL / Table API**: No ## Documentation - Does this pull request introduce a new feature? **No** - If yes, how is the feature documented? N/A -- 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]
