Admaing opened a new pull request, #337: URL: https://github.com/apache/flink-connector-kafka/pull/337
## What is the purpose of the change Reader-side follow-up to [FLINK-41088](https://issues.apache.org/jira/browse/FLINK-41088) for [FLINK-41100](https://issues.apache.org/jira/browse/FLINK-41100). `DynamicKafkaSourceSplit#splitId()` formatted the split id as `<cluster>-<topic>-<partition>`, so cluster `a` with topic `b-c` and cluster `a-b` with topic `c` both got `a-b-c-0`. Flink keys the split-local output, and with it the split-local watermark generator, by that id: the sub-readers of `DynamicKafkaSourceReader` emitted into the same output for both partitions, finishing one of them released the output the other one still wrote to (so that partition stopped holding back the combined watermark), watermark alignment paused and resumed both partitions together, and `GlobalSplitOwnerAssigner` conflated their owners. The split ids used towards Flink are now unique per physical partition. ## Brief change log - `DynamicKafkaSourceSplit#toSplitId(clusterId, partitionId)` is the single formatter for dynamic split ids. The cluster id is length prefixed (`<len>-<clusterId>-<partitionId>`), which makes the encoding injective over `(cluster id, partition id)` pairs even though topic names may contain `-`. - `DynamicKafkaSourceSplit#splitId()`, `KafkaPartitionSplitReaderWrapper` (`nextSplit()` and `finishedSplits()`) and `DynamicKafkaSourceEnumerator` all use that helper, so the ids the records carry match the ids the splits are registered under. - `DynamicKafkaSourceEnumerator#SPLIT_ORDER` drops the cluster id tiebreaker that FLINK-41088 added as a workaround for the colliding ids. - `GlobalSplitOwnerAssigner` needs no change: it keys its maps by split id, which is now unique per partition. A test pins that down. - `DynamicKafkaSourceSplitSerializer` is untouched: it stores the cluster id plus the `KafkaPartitionSplit`, never the formatted id, so its version, wire format and existing checkpoint/savepoint state are unaffected. ## Verifying this change - `DynamicKafkaSourceSplitTest`: cluster `a`/topic `b-c` and cluster `a-b`/topic `c` produce different ids; ids are unique per partition and stable. - `DynamicKafkaSourceReaderSplitOutputTest`: reader-level and mock based. A `ReaderOutput` that models the source operator's split-local outputs and a `WatermarkOutputMultiplexer` is fed by a `DynamicKafkaSourceReader` with two fake sub-readers; the test asserts that the partitions get distinct split-local outputs, that finishing one does not release the output the other still writes to, that the split-local watermarks stay per partition, and that the reading partition keeps holding back the combined watermark. - `KafkaPartitionSplitReaderWrapperTest`: the ids the wrapper emits for records and for finished splits equal the dynamic split ids. - `GlobalSplitOwnerAssignerTest`, `ReaderRecoveryGateTest` and `DynamicKafkaSourceEnumeratorRecoveryTest`: the tests that asserted the colliding format as a precondition now assert distinct ids (renamed from `...CollidingSplitIds...` to `...AmbiguousClusterAndTopicIds...`). Verified red-green: with the old `clusterId + "-" + partitionId` formatting these tests fail (for example `Expecting actual: "a-b-c-0" not to be equal to: "a-b-c-0"` and `Expected not same: ...RecordingSplitOutput@...`), and they pass with the change. ``` ./mvnw -pl flink-connector-kafka -Dtest='DynamicKafkaSourceSplitTest,KafkaPartitionSplitReaderWrapperTest,DynamicKafkaSourceReaderSplitOutputTest,GlobalSplitOwnerAssignerTest,ReaderRecoveryGateTest,DynamicKafkaSourceEnumeratorRecoveryTest,DynamicKafkaSourceReaderIdlenessTest,DynamicKafkaSourceReaderPauseResumeTest,DynamicKafkaSourceSplitSerializerTest,DynamicKafkaSourceEnumStateSerializerTest,DynamicKafkaSourceBuilderTest,KafkaClusterMetricGroupTest' -Dsurefire.failIfNoSpecifiedTests=false -Dspotless.check.skip=true -Dcheckstyle.skip=true -Drat.skip=true test ./mvnw -pl flink-connector-kafka spotless:check checkstyle:check ``` Testcontainer based tests (`DynamicKafkaSourceITTest`, `DynamicKafkaSourceClusterMetadataITCase`, `DynamicKafkaSourceEnumeratorTest`, `DynamicKafkaSourceReaderTest`) were not run locally. ## AI assistance disclosure Generative AI tooling was used for this change, as disclosed here and in the commit message. Generated-by: DeepSeek Harness (deepseek-v4.1-flash) -- 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]
