lnbest0707 commented on code in PR #279:
URL:
https://github.com/apache/flink-connector-kafka/pull/279#discussion_r3530968853
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -748,6 +796,19 @@ public void addReader(int subtaskId) {
// assign pending splits from the sub enumerator
clusterEnumeratorMap.forEach(
(cluster, subEnumerator) ->
subEnumerator.addReader(subtaskId));
+ if (!retainedSplitOffsetHandoffs.isEmpty()) {
+ pruneExpiredRetainedSplitOffsetHandoffs();
+ // A reader can join while a re-added cluster is waiting for
offset handoff. Send
+ // metadata first so the reader has reconciled its local retained
state before it
+ // answers the request. Restart the attempt so delayed responses
from the reader's
+ // previous attempt cannot count as its replacement report.
+ retainedSplitOffsetHandoffs
+ .values()
+ .forEach(handoff -> handoff.offsetsByReader.clear());
Review Comment:
Yes, a repeatedly re-registering reader restarts the handoff. This is
intentional: if a reader is unstable, we should not start the retained cluster
from an incomplete or stale offset snapshot. Once all readers stabilize, the
next handoff completes.
And normally, in such case, it usually means the slot itself is unhealthy
and would anyway block the process.
##########
docs/content/docs/connectors/datastream/dynamic-kafka.md:
##########
@@ -242,6 +242,8 @@ By default, metadata removal also removes that cluster's
split offsets from subs
To keep removed cluster offsets available for a later re-add or restore, set
`stream-metadata-removed-cluster-retention-ms` to a positive duration. For
example,
`604800000` retains removed cluster state for seven days before the source
stops checkpointing it.
+If the cluster is re-added, the source uses the retained offsets but computes
fresh reader
+assignments instead of reusing their previous owners.
Review Comment:
good point, will add
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -796,26 +857,303 @@ private void retainRemovedClusterEnumeratorStates(
}
long retainedUntilMs = System.currentTimeMillis() +
removedClusterStateRetentionMs;
- activeClusterEnumeratorStates.entrySet().stream()
- .filter(entry ->
!activeKafkaClusterIds.contains(entry.getKey()))
- .forEach(
- entry ->
- retainedClusterEnumeratorStates.put(
- entry.getKey(),
- new
DynamicKafkaSourceEnumState.RetainedClusterState(
- entry.getValue(),
retainedUntilMs)));
+ for (Entry<String, KafkaSourceEnumState> entry :
activeClusterEnumeratorStates.entrySet()) {
+ if (activeKafkaClusterIds.contains(entry.getKey())) {
+ continue;
+ }
+ retainedClusterEnumeratorStates.put(
+ entry.getKey(),
+ new DynamicKafkaSourceEnumState.RetainedClusterState(
+ entry.getValue(), retainedUntilMs));
+ retainedSplitOffsetHandoffs.remove(entry.getKey());
+ }
}
private void pruneExpiredRetainedClusterEnumeratorStates() {
if (removedClusterStateRetentionMs <= 0) {
retainedClusterEnumeratorStates.clear();
+ retainedSplitOffsetHandoffs.clear();
return;
}
long currentTimeMillis = System.currentTimeMillis();
retainedClusterEnumeratorStates
.entrySet()
.removeIf(entry -> entry.getValue().getRetainedUntilMs() <=
currentTimeMillis);
+
retainedSplitOffsetHandoffs.keySet().retainAll(retainedClusterEnumeratorStates.keySet());
+ }
+
+ private void pruneExpiredRetainedSplitOffsetHandoffs() {
+ long currentTimeMillis = System.currentTimeMillis();
+ retainedSplitOffsetHandoffs
+ .entrySet()
+ .removeIf(
+ entry -> {
+ RetainedSplitOffsetHandoff handoff =
entry.getValue();
+ if (!handoff.isExpired(currentTimeMillis)) {
+ return false;
+ }
+ logger.debug(
+ "Discarding timed out retained split
offset handoff for cluster {}: handoffId={}",
+ entry.getKey(),
+ handoff.handoffId);
+ handoff.offsetsByReader.clear();
+ return true;
+ });
+ }
+
+ private boolean isRetainedClusterReadyForAssignment(
+ String kafkaClusterId,
+ DynamicKafkaSourceEnumState.RetainedClusterState
retainedClusterState) {
+ Set<String> activeTopics =
+ latestClusterTopicsMap.getOrDefault(kafkaClusterId,
Collections.emptySet());
+ return
filterStateByTopics(retainedClusterState.getKafkaSourceEnumState(),
activeTopics)
+ .stream()
+ .noneMatch(
+ splitStatus ->
+
splitStatus.assignmentStatus().equals(AssignmentStatus.ASSIGNED));
+ }
+
+ private void startRetainedSplitOffsetHandoff(String kafkaClusterId) {
+ if (retainedSplitOffsetHandoffs.containsKey(kafkaClusterId)) {
+ return;
+ }
+
+ // Keep the attempt bounded without making a fast metadata refresh
interval shorter than a
+ // reader source-event round trip.
+ long handoffTimeoutMs =
+ Math.max(
+ kafkaMetadataServiceDiscoveryIntervalMs,
+ RETAINED_SPLIT_OFFSET_HANDOFF_MIN_TIMEOUT_MS);
+ long deadlineMs = System.currentTimeMillis() + handoffTimeoutMs;
+ RetainedSplitOffsetHandoff handoff =
+ new
RetainedSplitOffsetHandoff(++nextRetainedSplitOffsetHandoffId, deadlineMs);
+ retainedSplitOffsetHandoffs.put(kafkaClusterId, handoff);
+ scheduleRetainedSplitOffsetHandoffRetryIfNeeded();
+ for (int readerId : enumContext.registeredReaders().keySet()) {
+ sendRetainedSplitOffsetRequestToReader(kafkaClusterId, handoff,
readerId);
+ }
+ }
+
+ private void scheduleRetainedSplitOffsetHandoffRetryIfNeeded() {
+ if (kafkaMetadataServiceDiscoveryIntervalMs > 0
+ || retainedSplitOffsetHandoffRetryScheduled) {
+ return;
+ }
+
+ // One-time metadata discovery has no later refresh to discard an
expired handoff. Keep a
+ // single lightweight retry loop after the first handoff; it does not
fetch metadata.
+ retainedSplitOffsetHandoffRetryScheduled = true;
+ kafkaMetadataServiceDiscoveryContext.<Void>callAsync(
+ () -> null,
+ (ignored, t) -> {
+ if (t != null) {
+ throw new RuntimeException("Retained split offset
handoff retry failed", t);
+ }
+ pruneExpiredRetainedClusterEnumeratorStates();
+ pruneExpiredRetainedSplitOffsetHandoffs();
+ maybeStartReadyRetainedClusterEnumerators();
+ },
+ RETAINED_SPLIT_OFFSET_HANDOFF_MIN_TIMEOUT_MS,
+ RETAINED_SPLIT_OFFSET_HANDOFF_MIN_TIMEOUT_MS);
+ }
+
+ private void sendPendingRetainedSplitOffsetRequestsToReader(int readerId) {
+ retainedSplitOffsetHandoffs.forEach(
+ (kafkaClusterId, handoff) ->
+ sendRetainedSplitOffsetRequestToReader(kafkaClusterId,
handoff, readerId));
+ }
+
+ private void sendRetainedSplitOffsetRequestToReader(
+ String kafkaClusterId, RetainedSplitOffsetHandoff handoff, int
readerId) {
+ RequestRetainedSplitOffsetsEvent requestEvent =
+ new RequestRetainedSplitOffsetsEvent(handoff.handoffId,
kafkaClusterId);
+ logger.debug(
+ "Requesting retained split offsets from reader {}: {}",
readerId, requestEvent);
+ enumContext.sendEventToSourceReader(readerId, requestEvent);
+ }
+
+ private void handleRetainedSplitOffsetsEvent(
+ int subtaskId, RetainedSplitOffsetsEvent
retainedSplitOffsetsEvent) {
+ if (!enumContext.registeredReaders().containsKey(subtaskId)) {
+ logger.debug("Ignoring retained split offsets from unavailable
reader {}", subtaskId);
+ return;
+ }
+ pruneExpiredRetainedClusterEnumeratorStates();
+ String kafkaClusterId = retainedSplitOffsetsEvent.getKafkaClusterId();
+ RetainedSplitOffsetHandoff handoff =
retainedSplitOffsetHandoffs.get(kafkaClusterId);
+ if (handoff == null || handoff.handoffId !=
retainedSplitOffsetsEvent.getHandoffId()) {
+ logger.debug(
+ "Ignoring stale retained split offsets from reader {}: {}",
+ subtaskId,
+ retainedSplitOffsetsEvent);
+ return;
+ }
+ if (handoff.isExpired(System.currentTimeMillis())) {
+ logger.debug(
+ "Ignoring retained split offsets from timed out handoff
{}: {}",
+ handoff.handoffId,
+ retainedSplitOffsetsEvent);
+ clearRetainedSplitOffsetHandoff(kafkaClusterId, handoff);
+ return;
+ }
+
+ handoff.offsetsByReader.put(subtaskId,
retainedSplitOffsetsEvent.getRetainedSplitOffsets());
+ if (handoff.offsetsByReader.size() >=
enumContext.currentParallelism()) {
+ applyRetainedSplitOffsetHandoff(kafkaClusterId, handoff);
+ }
+ maybeStartReadyRetainedClusterEnumerators();
+ }
+
+ private void applyRetainedSplitOffsetHandoff(
+ String kafkaClusterId, RetainedSplitOffsetHandoff handoff) {
+ DynamicKafkaSourceEnumState.RetainedClusterState retainedClusterState =
+ retainedClusterEnumeratorStates.get(kafkaClusterId);
+ if (retainedClusterState == null) {
+ clearRetainedSplitOffsetHandoff(kafkaClusterId, handoff);
+ return;
+ }
+
+ KafkaSourceEnumState kafkaSourceEnumState =
retainedClusterState.getKafkaSourceEnumState();
+ Map<String, Long> retainedSplitOffsets = handoff.mergedOffsets();
+ Set<String> activeTopics =
+ latestClusterTopicsMap.getOrDefault(kafkaClusterId,
Collections.emptySet());
+ Set<SplitAndAssignmentStatus> updatedSplits = new HashSet<>();
+ for (SplitAndAssignmentStatus splitStatus :
kafkaSourceEnumState.splits()) {
+ if (!activeTopics.contains(splitStatus.split().getTopic())) {
+ updatedSplits.add(splitStatus);
+ continue;
+ }
+ if
(splitStatus.assignmentStatus().equals(AssignmentStatus.ASSIGNED)) {
+ Long retainedSplitOffset =
+ retainedSplitOffsets.get(
+ toDynamicSplitId(kafkaClusterId,
splitStatus.split()));
+ if (retainedSplitOffset == null) {
+ // No reader retains this offset anymore; let normal
discovery recreate it.
Review Comment:
Good catch! Yes, your analysis is correct, it would go back to `earliest` in
such case. And just to be clear, this branch would rarely be hit when -- topic
and cluster are retained in the enumerator checkpoint but missed in reader
checkpoint. This would only happen when the checkpoint is somehow corrupted or
in extreme time mismatch between TM/JM.
Falling completely back to `splitStatus.split()` is not safe because
enumerator state only contains the original assigned starting offset, not the
reader’s latest checkpointed progress. I’ll change this branch to keep the
partition as an uninitialized/MIGRATED unassigned split so
KafkaSourceEnumerator resolves it through the configured starting-offset
initializer.
Once with the above source side patch, it should not be conflict with the
sink behavior.
For the similar corner case, if reader has the retained offsets in
checkpoint but enumerator does not have it, we would treat then as "new"
cluster and use initializer to determine the starting point. That behavior
should be correct as we should already pass the TTL in such case.
--
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]
