[GitHub] [inlong] luchunliang commented on a diff in pull request #5093: [INLONG-5092][SDK] Change Kafka default partition assignment strategy to RangeAssignor

[GitBox <gi...@apache.org>]

luchunliang commented on code in PR #5093:
URL: https://github.com/apache/inlong/pull/5093#discussion_r922946908


##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarClusterProducer.java:
##########
@@ -282,6 +281,27 @@ public boolean send(DispatchProfile event) {
         }
     }
 
+    /**
+     * getProducerTopic
+     * @param event
+     * @return

Review Comment:
   add description



##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/kafka/AckOffsetOnRebalance.java:
##########
@@ -19,43 +19,43 @@
 
 package org.apache.inlong.sdk.sort.impl.kafka;
 
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+
 public class AckOffsetOnRebalance implements ConsumerRebalanceListener {
 
     private final Logger logger = LoggerFactory.getLogger(AckOffsetOnRebalance.class);
+    private final String clusterId;
     private final KafkaConsumer<byte[], byte[]> consumer;
     private final ConcurrentHashMap<TopicPartition, OffsetAndMetadata> commitOffsetMap;
 
-    public AckOffsetOnRebalance(KafkaConsumer<byte[], byte[]> consumer,
+    public AckOffsetOnRebalance(String clusterId, KafkaConsumer<byte[], byte[]> consumer,
             ConcurrentHashMap<TopicPartition, OffsetAndMetadata> commitOffsetMap) {
+        this.clusterId = clusterId;
         this.consumer = consumer;
         this.commitOffsetMap = commitOffsetMap;
     }
 
     @Override
     public void onPartitionsRevoked(Collection<TopicPartition> collection) {
         logger.debug("*- in ralance:onPartitionsRevoked");

Review Comment:
   fixed



##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/kafka/AckOffsetOnRebalance.java:
##########
@@ -19,43 +19,43 @@
 
 package org.apache.inlong.sdk.sort.impl.kafka;
 
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+
 public class AckOffsetOnRebalance implements ConsumerRebalanceListener {
 
     private final Logger logger = LoggerFactory.getLogger(AckOffsetOnRebalance.class);
+    private final String clusterId;
     private final KafkaConsumer<byte[], byte[]> consumer;
     private final ConcurrentHashMap<TopicPartition, OffsetAndMetadata> commitOffsetMap;
 
-    public AckOffsetOnRebalance(KafkaConsumer<byte[], byte[]> consumer,
+    public AckOffsetOnRebalance(String clusterId, KafkaConsumer<byte[], byte[]> consumer,
             ConcurrentHashMap<TopicPartition, OffsetAndMetadata> commitOffsetMap) {
+        this.clusterId = clusterId;
         this.consumer = consumer;
         this.commitOffsetMap = commitOffsetMap;
     }
 
     @Override
     public void onPartitionsRevoked(Collection<TopicPartition> collection) {
         logger.debug("*- in ralance:onPartitionsRevoked");
-        while (!commitOffsetMap.isEmpty()) {
-            consumer.commitSync(commitOffsetMap);
-        }
+        collection.forEach((v) -> {
+            logger.info("clusterId:{},onPartitionsRevoked:{}", clusterId, v.toString());
+        });
     }
 
     @Override
     public void onPartitionsAssigned(Collection<TopicPartition> collection) {
         logger.debug("*- in ralance:onPartitionsAssigned  ");

Review Comment:
   fixed



-- 
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: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org