Savonitar commented on code in PR #279:
URL:
https://github.com/apache/flink-connector-kafka/pull/279#discussion_r3573331486
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -833,7 +944,11 @@ 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:
1. Is my understanding correct that a reader drops its retained copy only
when the same split is freshly assigned to that same reader? So after a re-add
where ownership moved (e.g. rescale), a checkpoint can contain both the
retained copy on reader#1 and the active copy on reader#2 until the retention
expires?
2. If the job restores from such checkpoint, both readers report the split
here, and the retained copy takes this active branch because the cluster is no
longer in `retainedClusterEnumeratorStates`.
Wouldn't putIfAbsent then hit the duplicate and throw
`IllegalStateException`??
Wouldn't the job be stuck in a restart loop?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -208,19 +213,38 @@ private InputStatus consolidateInputStatus(
@Override
public void addSplits(List<DynamicKafkaSourceSplit> splits) {
logger.info("Adding splits to reader {}: {}",
readerContext.getIndexOfSubtask(), splits);
+ List<DynamicKafkaSourceSplit> activeSplits = new ArrayList<>();
+ long currentTimeMillis = System.currentTimeMillis();
for (DynamicKafkaSourceSplit split : splits) {
+ if (split.isRetained()) {
+ if (removedClusterStateRetentionMs > 0
+ && split.isRetained(currentTimeMillis)
+ && !activeSplitIds.contains(split.splitId())) {
+ retainedSplits.add(split);
+ }
+ continue;
+ }
+
+ retainedSplits.removeIf(
Review Comment:
if the fresh assignment of this split will be assigned to a different
reader, what will remove the retained copy from this reader before its
retention expires? if nothing, is the recovery path prepared for the same split
id appearing retained in one reader's checkpoint and active in another?
--
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]
