[INLONG-5623][SDK] Support multi-topic fetcher for Kafka

inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/MultiTopicsFetcher.java

@@ -19,6 +19,7 @@
 
 import org.apache.inlong.sdk.sort.entity.InLongTopic;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -65,4 +66,20 @@ public MultiTopicsFetcher(
         this.executor = Executors.newSingleThreadScheduledExecutor();
     }
 
+    protected boolean needUpdate(Collection<InLongTopic> newTopics) {
+        if (newTopics.size() != onlineTopics.size()) {
+            return true;
+        }
+        // all topic should share the same properties in one task
+        if (Objects.equals(newTopics.stream().findFirst(), onlineTopics.values().stream().findFirst())) {
+            return true;
+        }
+        for (InLongTopic topic : newTopics) {
+            if (!onlineTopics.containsKey(topic.getTopic())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
 }

inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/kafka/AckOffsetOnRebalance.java

@@ -26,20 +26,32 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
 
 public class AckOffsetOnRebalance implements ConsumerRebalanceListener {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AckOffsetOnRebalance.class);
     private final String clusterId;
     private final Seeker seeker;
     private final ConcurrentHashMap<TopicPartition, OffsetAndMetadata> commitOffsetMap;
+    private final ConcurrentHashMap<TopicPartition, ConcurrentSkipListMap<Long, Boolean>> ackOffsetMap;
 
     public AckOffsetOnRebalance(String clusterId, Seeker seeker,
                                 ConcurrentHashMap<TopicPartition, OffsetAndMetadata> commitOffsetMap) {
+        this(clusterId, seeker, commitOffsetMap, null);
+    }
+
+    public AckOffsetOnRebalance(
+            String clusterId,
+            Seeker seeker,
+            ConcurrentHashMap<TopicPartition, OffsetAndMetadata> commitOffsetMap,
+            ConcurrentHashMap<TopicPartition, ConcurrentSkipListMap<Long, Boolean>> ackOffsetMap) {
         this.clusterId = clusterId;
         this.seeker = seeker;
         this.commitOffsetMap = commitOffsetMap;
+        this.ackOffsetMap = ackOffsetMap;
     }
 
     @Override
@@ -48,6 +60,30 @@ public void onPartitionsRevoked(Collection<TopicPartition> collection) {
         collection.forEach((v) -> {
             LOGGER.info("clusterId:{},onPartitionsRevoked:{}", clusterId, v.toString());
         });
+        if (Objects.nonNull(ackOffsetMap) && Objects.nonNull(commitOffsetMap)) {
+            ackRevokedPartitions(collection);
+        }
+    }
+
+    private void ackRevokedPartitions(Collection<TopicPartition> collection) {
+        collection.forEach(tp -> {
+            if (!ackOffsetMap.containsKey(tp)) {
+                return;
+            }
+            ConcurrentSkipListMap<Long, Boolean> tpOffsetMap = ackOffsetMap.remove(tp);
+            long commitOffset = -1;
+            for (Long ackOffset : tpOffsetMap.keySet()) {
+                if (!tpOffsetMap.get(ackOffset)) {
+                    break;
+                }
+                commitOffset = ackOffset;
+            }
+            // the first haven't ack, do nothing
+            if (commitOffset == -1) {
+                return;
+            }
+            commitOffsetMap.put(tp, new OffsetAndMetadata(commitOffset));
+        });
     }
 
     @Override