[INLONG-11537][Sort] Optimize the session key generation of TubeMQ So…

…urce
vernedeng · Nov 25, 2024 · 3645d99 · 3645d99
1 parent 82128a3
commit 3645d99
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...rt-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java b/...rt-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java
@@ -221,7 +221,10 @@ public void open(Configuration parameters) throws Exception {
         messagePullConsumer = messageSessionFactory.createPullConsumer(consumerConfig);
         messagePullConsumer.subscribe(topic, streamIdSet);
         String jobId = getRuntimeContext().getJobId().toString();
-        messagePullConsumer.completeSubscribe(sessionKey.concat(jobId), numTasks, true, currentOffsets);
+        String attemptNumber = String.valueOf(getRuntimeContext().getAttemptNumber());
+        String startSessionKey = sessionKey.concat("_").concat(jobId).concat("_").concat(attemptNumber);
+        LOG.info("start to init tube mq consumer, session key={}", startSessionKey);
+        messagePullConsumer.completeSubscribe(startSessionKey, numTasks, true, currentOffsets);
 
         running = true;
     }