[INLONG-9077][Sort] Fix TubeMQ connector fail to subscribe streamId

vernedeng · Oct 19, 2023 · abf44d9 · abf44d9
1 parent f0908ee
commit abf44d9
Show file tree

Hide file tree

Showing 4 changed files with 6 additions and 6 deletions.
diff --git a/...-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/TubeMqProvider.java b/...-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/TubeMqProvider.java
@@ -53,8 +53,8 @@ public ExtractNode createExtractNode(StreamNode streamNodeInfo) {
                 source.getMasterRpc(),
                 source.getTopic(),
                 source.getSerializationType(),
-                source.getGroupId(),
+                source.getConsumeGroup(),
                 source.getSessionKey(),
-                source.getTid());
+                source.getStreamId());
     }
 }
diff --git a/...manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/tubemq/TubeMQSource.java b/...manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/tubemq/TubeMQSource.java
@@ -52,7 +52,7 @@ public class TubeMQSource extends StreamSource {
     private String topic;
 
     @ApiModelProperty("Group of the TubeMQ")
-    private String groupId;
+    private String consumeGroup;
 
     @ApiModelProperty("Session key of the TubeMQ")
     private String sessionKey;
@@ -61,7 +61,7 @@ public class TubeMQSource extends StreamSource {
      * The TubeMQ consumers use this tid set to filter records reading from server.
      */
     @ApiModelProperty("Tid of the TubeMQ")
-    private TreeSet<String> tid;
+    private TreeSet<String> streamId;
 
     public TubeMQSource() {
         this.setSourceType(SourceType.TUBEMQ);

diff --git a/...e/src/main/java/org/apache/inlong/manager/service/source/tubemq/TubeMQSourceOperator.java b/...e/src/main/java/org/apache/inlong/manager/service/source/tubemq/TubeMQSourceOperator.java
@@ -107,7 +107,7 @@ public Map<String, List<StreamSource>> getSourcesMap(InlongGroupInfo groupInfo,
             String streamId = streamInfo.getInlongStreamId();
             tubeMQSource.setSourceName(streamId);
             tubeMQSource.setTopic(groupInfo.getMqResource());
-            tubeMQSource.setGroupId(streamId);
+            tubeMQSource.setConsumeGroup(streamId);
             tubeMQSource.setMasterRpc(masterRpc);
             tubeMQSource.setIgnoreParseError(streamInfo.getIgnoreParseError());
 

diff --git a/...-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/TubeMQExtractNode.java b/...-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/TubeMQExtractNode.java
@@ -85,7 +85,7 @@ public TubeMQExtractNode(
             @Nonnull @JsonProperty("format") String format,
             @Nonnull @JsonProperty("consumeGroup") String consumeGroup,
             @JsonProperty("sessionKey") String sessionKey,
-            @JsonProperty("tid") TreeSet<String> streamId) {
+            @JsonProperty("streamId") TreeSet<String> streamId) {
         super(id, name, fields, waterMarkField, properties);
         this.masterRpc = Preconditions.checkNotNull(masterRpc, "TubeMQ masterRpc is null");
         this.topic = Preconditions.checkNotNull(topic, "TubeMQ topic is null");