[incubator-inlong] branch master updated: [INLONG-2014][InLong-DataProxy] inlong-dataproxy could not identify the groupId and topic format of topics.properties (#2016)

[gosonzhang]

This is an automated email from the ASF dual-hosted git repository.

gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new f5bec84  [INLONG-2014][InLong-DataProxy] inlong-dataproxy could not 
identify the groupId and topic format of topics.properties (#2016)
f5bec84 is described below

commit f5bec842383f65df6817dbb6c40068b3e2f7b9ee
Author: baomingyu <baomingy...@163.com>
AuthorDate: Fri Dec 17 09:30:52 2021 +0800

    [INLONG-2014][InLong-DataProxy] inlong-dataproxy could not identify the 
groupId and topic format of topics.properties (#2016)
---
 .../java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git 
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
 
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
index 8091a07..3e05d92 100644
--- 
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
+++ 
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
@@ -227,7 +227,7 @@ public class ServerMessageHandler extends 
SimpleChannelHandler {
     private void checkGroupIdInfo(ProxyMessage message, Map<String, String> 
commonAttrMap,
         Map<String, String> attrMap, AtomicReference<String> topicInfo) {
         String groupId = message.getGroupId();
-        String streamId = null;
+        String streamId = message.getStreamId();
         if (null != groupId) {
             String from = commonAttrMap.get(AttributeConstants.FROM);
             if ("dc".equals(from)) {
@@ -241,7 +241,7 @@ public class ServerMessageHandler extends 
SimpleChannelHandler {
                 }
             }
 
-            String value = getTopic(groupId);
+            String value = getTopic(groupId, streamId);
             if (StringUtils.isNotEmpty(value)) {
                 topicInfo.set(value.trim());
             }