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 aeb0392 [INLONG-2009][InLong-DataProxy] Topic obtained through
"openapi/dataproxy/getConfig" is wrong bug inlong-manager (#2011)
aeb0392 is described below
commit aeb03925779ba6e55e890f83673fb75dee63e830
Author: baomingyu <[email protected]>
AuthorDate: Thu Dec 16 16:07:28 2021 +0800
[INLONG-2009][InLong-DataProxy] Topic obtained through
"openapi/dataproxy/getConfig" is wrong bug inlong-manager (#2011)
---
.../dataproxy/source/ServerMessageHandler.java | 32 ++++++++++++++++++----
1 file changed, 27 insertions(+), 5 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 e6eee0e..8091a07 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;
+ String streamId = null;
if (null != groupId) {
String from = commonAttrMap.get(AttributeConstants.FROM);
if ("dc".equals(from)) {
@@ -241,7 +241,7 @@ public class ServerMessageHandler extends
SimpleChannelHandler {
}
}
- String value = configManager.getTopicProperties().get(groupId);
+ String value = getTopic(groupId);
if (StringUtils.isNotEmpty(value)) {
topicInfo.set(value.trim());
}
@@ -277,7 +277,7 @@ public class ServerMessageHandler extends
SimpleChannelHandler {
message.setGroupId(groupId);
message.setStreamId(streamId);
- String value =
configManager.getTopicProperties().get(groupId);
+ String value = getTopic(groupId, streamId);
if (StringUtils.isNotEmpty(value)) {
topicInfo.set(value.trim());
}
@@ -652,10 +652,32 @@ public class ServerMessageHandler extends
SimpleChannelHandler {
}
/**
+ * get topic
+ */
+ private String getTopic(String groupId) {
+ return getTopic(groupId, null);
+ }
+
+ /**
+ * get topic
+ */
+ private String getTopic(String groupId, String streamId) {
+ String topic = null;
+ if (StringUtils.isNotEmpty(groupId)) {
+ if (StringUtils.isNotEmpty(streamId)) {
+ topic = configManager.getTopicProperties().get(groupId + "/" +
streamId);
+ }
+ if (StringUtils.isEmpty(topic)) {
+ topic = configManager.getTopicProperties().get(groupId);
+ }
+ }
+ logger.debug("Get topic by groupId = {} , streamId = {}", groupId,
streamId);
+ return topic;
+ }
+
+ /**
* addMetric
*
- * @param currentRecord
- * @param topic
* @param result
* @param size
*/
