[INLONG-10522][SDK] SortSDK support assgin subscription (apache#10523)

justinwwhuang · Jun 27, 2024 · f1679e3 · f1679e3
1 parent 396ae72
commit f1679e3
Show file tree

Hide file tree

Showing 7 changed files with 22 additions and 5 deletions.
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
@@ -35,6 +35,7 @@ public class SortClientConfig implements Serializable {
     private static final long serialVersionUID = -7531960714809683830L;
 
     private final String sortTaskId;
+    private final String subscription;
     private final String sortClusterName;
     private InLongTopicChangeListener assignmentsListener;
     private ReadCallback callback;
@@ -83,11 +84,22 @@ public SortClientConfig(
             InLongTopicChangeListener assignmentsListener,
             ConsumeStrategy consumeStrategy,
             String localIp) {
+        this(sortTaskId, sortClusterName, assignmentsListener, consumeStrategy, localIp, sortTaskId);
+    }
+
+    public SortClientConfig(
+            String sortTaskId,
+            String sortClusterName,
+            InLongTopicChangeListener assignmentsListener,
+            ConsumeStrategy consumeStrategy,
+            String localIp,
+            String subscription) {
         this.sortTaskId = sortTaskId;
         this.sortClusterName = sortClusterName;
         this.assignmentsListener = assignmentsListener;
         this.consumeStrategy = consumeStrategy;
         this.localIp = localIp;
+        this.subscription = subscription;
     }
 
     public boolean isStopConsume() {
@@ -102,6 +114,10 @@ public String getSortTaskId() {
         return sortTaskId;
     }
 
+    public String getSubscription() {
+        return subscription;
+    }
+
     public String getSortClusterName() {
         return sortClusterName;
     }

diff --git a/...t-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/kafka/KafkaMultiTopicsFetcher.java b/...t-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/kafka/KafkaMultiTopicsFetcher.java
@@ -106,7 +106,7 @@ public boolean init() {
     private KafkaConsumer<byte[], byte[]> createKafkaConsumer() {
         Properties properties = new Properties();
         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
-        properties.put(ConsumerConfig.GROUP_ID_CONFIG, context.getConfig().getSortTaskId());
+        properties.put(ConsumerConfig.GROUP_ID_CONFIG, context.getConfig().getSubscription());
         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                 ByteArrayDeserializer.class.getName());
         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,

diff --git a/...t-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/kafka/KafkaSingleTopicFetcher.java b/...t-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/kafka/KafkaSingleTopicFetcher.java
@@ -160,7 +160,7 @@ public List<InLongTopic> getTopics() {
     private void createKafkaConsumer(String bootstrapServers) {
         Properties properties = new Properties();
         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
-        properties.put(ConsumerConfig.GROUP_ID_CONFIG, context.getConfig().getSortTaskId());
+        properties.put(ConsumerConfig.GROUP_ID_CONFIG, context.getConfig().getSubscription());
         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                 ByteArrayDeserializer.class.getName());
         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,

diff --git a/...sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarMultiTopicsFetcher.java b/...sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarMultiTopicsFetcher.java
@@ -165,7 +165,7 @@ private Consumer<byte[]> createConsumer(Collection<InLongTopic> newTopics) {
                     .collect(Collectors.toList());
             Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
                     .topics(topicNames)
-                    .subscriptionName(context.getConfig().getSortTaskId())
+                    .subscriptionName(context.getConfig().getSubscription())
                     .subscriptionType(SubscriptionType.Shared)
                     .startMessageIdInclusive()
                     .subscriptionInitialPosition(position)

diff --git a/...sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarSingleTopicFetcher.java b/...sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarSingleTopicFetcher.java
@@ -151,7 +151,7 @@ private boolean createConsumer(PulsarClient client) {
 
             consumer = client.newConsumer(Schema.BYTES)
                     .topic(topic.getTopic())
-                    .subscriptionName(context.getConfig().getSortTaskId())
+                    .subscriptionName(context.getConfig().getSubscription())
                     .subscriptionType(SubscriptionType.Shared)
                     .startMessageIdInclusive()
                     .subscriptionInitialPosition(position)

diff --git a/...ort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/tube/TubeSingleTopicFetcher.java b/...ort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/tube/TubeSingleTopicFetcher.java
@@ -71,7 +71,7 @@ public boolean init() {
         TubeClientConfig tubeClientConfig = tubeConsumerCreator.getTubeClientConfig();
         try {
             ConsumerConfig consumerConfig = new ConsumerConfig(tubeClientConfig.getMasterInfo(),
-                    context.getConfig().getSortTaskId());
+                    context.getConfig().getSubscription());
 
             messageConsumer = tubeConsumerCreator.getMessageSessionFactory().createPullConsumer(consumerConfig);
             if (messageConsumer != null) {

diff --git a/...sdk/src/test/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImplTest.java b/...sdk/src/test/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImplTest.java
@@ -81,6 +81,7 @@ public void setUp() throws Exception {
 
         when(clientContext.getConfig()).thenReturn(sortClientConfig);
         when(sortClientConfig.getSortTaskId()).thenReturn("sortTaskId");
+        when(sortClientConfig.getSubscription()).thenReturn("sortTaskId");
 
     }