Savonitar commented on code in PR #279:
URL:
https://github.com/apache/flink-connector-kafka/pull/279#discussion_r3881327825
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -833,13 +944,33 @@ private void reassignReportedSplits() {
new TreeMap<>(pendingReportedSplitsByReader).entrySet()) {
int readerId = readerSplits.getKey();
for (DynamicKafkaSourceSplit split : readerSplits.getValue()) {
- if (isSplitActive(split)) {
+ boolean retainedClusterWaitingForHandoff =
+ split.isRetained()
+ && retainedClusterEnumeratorStates.containsKey(
+ split.getKafkaClusterId());
+ if (!retainedClusterWaitingForHandoff && isSplitActive(split))
{
Review Comment:
I have a question.
Say we have 1 partition, 2 subtasks, `GLOBAL` enumerator mode, 1 cluster.
1. The cluster is removed while reader 1 owns the partition at offset 100,
so reader 1 (and only reader 1) keeps a retained copy of the split in its
state.
2. The cluster is re-added, the offset handoff completes, ownership is
reassigned from scratch, and the round-robin in `GlobalSplitOwnerAssigner`
assigns the fresh split to reader 0 `(floorMod(0, 2) = 0). `
3. Reader 0 consumes on from 100, say to 150. Since
`retainedSplits.removeIf` only runs on the reader that receives the fresh
assignment, reader 1 still carries its **retained** copy(100) checkpointed.
4. Now reader 1 alone fails over (e,g. task failure). On recovery it reports
its restored state, which is just the retained copy at registration.
5. Reader 0 was never unregistered, so `allReadersRegistered()` is satisfied
and `reassignReportedSplits` runs with only reader 1's batch, then
`clearRetention()` called, and goes through `addSplitsBack` (which also removes
reader 0's entry from the sub-enumerator's `assignedSplits`) . And then
assigned as a second active copy at offset 100 while reader 0 keeps consuming
at 150.
Is that correct, or am I missing a reconciliation/protection step somewhere?
If it is correct, then in my understanding EOS sink will commit duplicates
and violate EOS guarantee.
--
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]
