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

…urce
apache · Nov 25, 2024 · 3e25d55 · 3e25d55
1 parent 3645d99
commit 3e25d55
Showing 1 changed file with 3 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
@@ -69,6 +69,8 @@ public class FlinkTubeMQConsumer<T> extends RichParallelSourceFunction<T>
     private static final Logger LOG = LoggerFactory.getLogger(FlinkTubeMQConsumer.class);
     private static final String TUBE_OFFSET_STATE = "tube-offset-state";
 
+    private static final String UNDERSCORE = "_";
+
     /**
      * The address of TubeMQ master, format eg: 127.0.0.1:8715,127.0.0.2:8715.
      */
@@ -222,7 +224,7 @@ public void open(Configuration parameters) throws Exception {
         messagePullConsumer.subscribe(topic, streamIdSet);
         String jobId = getRuntimeContext().getJobId().toString();
         String attemptNumber = String.valueOf(getRuntimeContext().getAttemptNumber());
-        String startSessionKey = sessionKey.concat("_").concat(jobId).concat("_").concat(attemptNumber);
+        String startSessionKey = sessionKey.concat(UNDERSCORE).concat(jobId).concat(UNDERSCORE).concat(attemptNumber);
         LOG.info("start to init tube mq consumer, session key={}", startSessionKey);
         messagePullConsumer.completeSubscribe(startSessionKey, numTasks, true, currentOffsets);