MartijnVisser commented on PR #323: URL: https://github.com/apache/flink-connector-kafka/pull/323#issuecomment-5742888979
The test behind the blocker, using the helpers this PR adds to `DynamicKafkaSourceEnumeratorRecoveryTest`. Restored state has two splits on `cluster-0`, parallelism 2: ```java enumerator.start(); // all readers register before the first discovery completes; reader 0 reports both splits registerReaderAndReportRestoredSplits(context, enumerator, 0, splits); registerReaderAndReportRestoredSplits(context, enumerator, 1, Collections.emptyList()); assertThat(context.getSplitsAssignmentSequence()).isEmpty(); // a checkpoint completes in this window: reader state is empty, the sub-enumerator state // still marks both splits assigned. Reader 0 fails, restores from it and re-registers empty. context.unregisterReader(0); enumerator.addSplitsBack(Collections.emptyList(), 0); registerReaderAndReportRestoredSplits(context, enumerator, 0, Collections.emptyList()); context.runNextOneTimeCallable(); assertExactCheckpointOffsets(context, 0, splits); ``` Passes on main. On this branch the last assertion sees no assignment at all, and nothing rediscovers the two partitions because the sub-enumerator still holds them as assigned. -- 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]
