You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "Anurag15v (via GitHub)" <gi...@apache.org> on 2026/10/03 15:10:12 UTC

[PR] [flink] Add Flink connector test suite ITCase for FlussSource [fluss]

Anurag15v opened a new pull request, #4537:
URL: https://github.com/apache/fluss/pull/4537

   **Generative AI disclosure:**
   - [x] Yes — Generated-by: Claude Code (Claude Opus 5.5) following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
   
   ### Purpose
   
   Linked issue: close #280
   
   The Flink connector lacked integration tests for restoring the source from savepoint state. The existing `FlinkTableSourceFailOverITCase#testRestore` only covers a partitioned log table via SQL with parallelism 1. Following the suggestion in the issue, this PR runs Flink's standard connector testing framework (`SourceTestSuiteBase` from `flink-connector-test-utils`) against `FlussSource`.
   
   ### Brief change log
   
   The 8 test cases themselves come from Flink's `SourceTestSuiteBase` (single split, multiple splits, savepoint restore, scale up, scale down, metrics, idle reader, TaskManager failover). This PR only adds the Fluss-side adapter and test classes — no production code changes, 7 new test files:
   
   - `FlussSourceExternalContext` (`fluss-flink-common`, test): the `DataStreamSourceExternalContext` for Fluss.
     - Each context creates its own partitioned log table with `bucket.num = 1`.
     - Each `createSourceSplitDataWriter()` call creates a **new partition**, and its writer stamps that partition value on every row, so 1 framework split = 1 partition = 1 bucket = 1 Fluss `LogSplit`. This keeps the number of Fluss splits equal to what the framework requests (required e.g. by `testIdleReader`) and also exercises restoring partition assignment state.
     - Bounded tests use `setStoppingOffsets(OffsetsInitializer.latest())`; unbounded tests use the default.
   - `FlussSourceTestSuiteITCase` (`fluss-flink-common`, test): abstract base wiring `MiniClusterTestEnvironment` (`@TestEnv`), the context factory (`@TestContext`), `EXACTLY_ONCE` semantics (`@TestSemantics`), and the Fluss cluster via `FlussClusterExtension`.
     - `@TestSemantics` uses the deprecated `org.apache.flink.streaming.api.CheckpointingMode` on purpose: it is the only type accepted by the framework in Flink 1.18/1.19, and Flink 1.20+ still accepts it via a Fluss-side adapter and test classes — no production code changes, 7 new test files:
   
   - `FlussSourceExternalContext` (`fluss-flink-common`, test): the `DataStreamSourceExternalContext` for Fluss.
     - Each context creates its own partitioned log table with `bucket.num = 1`.
     - Each `createSourceSplitDataWriter()` call creates a **new partition**, and its writer stamps that partition value on every row, so 1 framework split = 1 partition = 1 bucket = 1 Fluss `LogSplit`. This keeps the number of Fluss splits equal to what the framework requests (required e.g. by `testIdleReader`) and also exercises restoring partition assignment state.
     - Bounded tests use `setStoppingOffsets(OffsetsInitializer.latest())`; unbounded tests use the default.
   - `FlussSourceTestSuiteITCase` (`fluss-flink-common`, test): abstract base wiring `MiniClusterTestEnvironment` (`@TestEnv`), the context factory (`@TestContext`), `EXACTLY_ONCE` semantics (`@TestSemantics`), and the Fluss cluster via `FlussClusterExtension`.
     - `@TestSemantics` uses the deprecated `org.apache.flink.streaming.api.CheckpointingMode` on purpose: it is the only type accepted by the framework in Flink 1.18/1.19, and Flink 1.20+ still accepts it via a fallback, so one shared field works for all versions.
   - Version subclasses, following `fluss-flink-common/README.md`:
     - `Flink120FlussSourceTestSuiteITCase` runs the full suite.
     - `Flink118/119/22/23FlussSourceTestSuiteITCase` override `testSavepoint` only to mark it `@MultiVersionTest` as the representative multi-version test.
   
   ### Tests
   
   - `Flink120FlussSourceTestSuiteITCase`: 8/8 tests pass.
   - `Flink118FlussSourceTestSuiteITCase`, `Flink119FlussSourceTestSuiteITCase`, `Flink22FlussSourceTestSuiteITCase`, `Flink23FlussSourceTestSuiteITCase`: `testSavepoint` passes; the other 7 tests are disabled by the multi-version filter as intended.
   - Checkstyle, Spotless and RAT pass.
   
   ### API and Format
   
   No. Test-only change.
   
   ### Documentation
   
   No.
   
   🤖 AI-assisted changes - reviewed by human developer


-- 
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@fluss.apache.org

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

Re: [PR] [flink] Add Flink connector test suite ITCase for FlussSource [fluss]

Posted by "Anurag15v (via GitHub)" <gi...@apache.org>.
Anurag15v commented on PR #4537:
URL: https://github.com/apache/fluss/pull/4537#issuecomment-5980693592

   Hi @loserwang1024 
   Can you review this PR ?


-- 
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@fluss.apache.org

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