This is an automated email from the ASF dual-hosted git repository. vernedeng pushed a commit to branch branch-1.8 in repository https://gitbox.apache.org/repos/asf/inlong.git
commit c67e21832c954c867366f300e421d32b903fcbbb Author: fuweng11 <[email protected]> AuthorDate: Thu Jul 13 09:44:19 2023 +0800 [INLONG-8504][Manager] Modify nodeIp in dataproxy message to clientIp (#8519) --- .../java/org/apache/inlong/manager/pojo/consume/BriefMQMessage.java | 4 ++-- .../apache/inlong/manager/service/message/DeserializeOperator.java | 2 +- .../inlong/manager/service/message/InlongMsgDeserializeOperator.java | 2 +- .../inlong/manager/service/message/PbMsgDeserializeOperator.java | 2 +- .../inlong/manager/service/message/RawMsgDeserializeOperator.java | 2 +- .../org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java | 4 ++-- 6 files changed, 8 insertions(+), 8 deletions(-) diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/BriefMQMessage.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/BriefMQMessage.java index 149d0a2353..a09c3c0b0c 100644 --- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/BriefMQMessage.java +++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/BriefMQMessage.java @@ -46,8 +46,8 @@ public class BriefMQMessage { @ApiModelProperty(value = "Date") private Long dt; - @ApiModelProperty(value = "Node ip") - private String nodeIp; + @ApiModelProperty(value = "Client ip") + private String clientIp; @ApiModelProperty(value = "Message body") private String body; diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/DeserializeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/DeserializeOperator.java index 876c828f12..290ac1ca15 100644 --- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/DeserializeOperator.java +++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/DeserializeOperator.java @@ -30,7 +30,7 @@ import java.util.Map; public interface DeserializeOperator { String COMPRESS_TYPE_KEY = "compressType"; - String NODE_IP = "NodeIP"; + String CLIENT_IP = "clientIp"; String MSG_TIME_KEY = "msgTime"; char INLONGMSG_ATTR_ENTRY_DELIMITER = '&'; char INLONGMSG_ATTR_KV_DELIMITER = '='; diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/InlongMsgDeserializeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/InlongMsgDeserializeOperator.java index c72c6ea4d7..9ae409425b 100644 --- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/InlongMsgDeserializeOperator.java +++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/InlongMsgDeserializeOperator.java @@ -72,7 +72,7 @@ public class InlongMsgDeserializeOperator implements DeserializeOperator { continue; } BriefMQMessage inLongMessage = - new BriefMQMessage(index, groupId, streamId, msgTime, attributes.get(NODE_IP), + new BriefMQMessage(index, groupId, streamId, msgTime, attributes.get(CLIENT_IP), new String(bodyBytes, Charset.forName(streamInfo.getDataEncoding()))); messageList.add(inLongMessage); } diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/PbMsgDeserializeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/PbMsgDeserializeOperator.java index 0ae13afb42..8a9320237d 100644 --- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/PbMsgDeserializeOperator.java +++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/PbMsgDeserializeOperator.java @@ -80,7 +80,7 @@ public class PbMsgDeserializeOperator implements DeserializeOperator { } BriefMQMessage briefMQMessage = new BriefMQMessage(index, headers.get(AttributeConstants.GROUP_ID), headers.get(AttributeConstants.STREAM_ID), messageObj.getMsgTime(), - headers.get(NODE_IP), + headers.get(CLIENT_IP), new String(messageObj.getBody().toByteArray(), Charset.forName(streamInfo.getDataEncoding()))); briefMQMessages.add(briefMQMessage); } diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/RawMsgDeserializeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/RawMsgDeserializeOperator.java index e039959323..de4b4a628b 100644 --- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/RawMsgDeserializeOperator.java +++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/RawMsgDeserializeOperator.java @@ -46,7 +46,7 @@ public class RawMsgDeserializeOperator implements DeserializeOperator { String streamId = headers.get(AttributeConstants.STREAM_ID); long msgTime = Long.parseLong(headers.getOrDefault(MSG_TIME_KEY, "0")); return Collections - .singletonList(new BriefMQMessage(null, groupId, streamId, msgTime, headers.get(NODE_IP), + .singletonList(new BriefMQMessage(null, groupId, streamId, msgTime, headers.get(CLIENT_IP), new String(msgBytes, Charset.forName(streamInfo.getDataEncoding())))); } diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java index 7af7c06363..ace88b88e3 100644 --- a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java +++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java @@ -55,7 +55,7 @@ public class MessageDeserializer implements Deserializer { private static final String INLONGMSG_ATTR_GROUP_ID = "groupId"; private static final String INLONGMSG_ATTR_TIME_T = "t"; private static final String INLONGMSG_ATTR_TIME_DT = "dt"; - private static final String INLONGMSG_ATTR_NODE_IP = "NodeIP"; + private static final String INLONGMSG_ATTR_CLIENT_IP = "clientIp"; private static final char INLONGMSG_ATTR_ENTRY_DELIMITER = '&'; private static final char INLONGMSG_ATTR_KV_DELIMITER = '='; private static final String DEFAULT_IP = "127.0.0.1"; @@ -194,7 +194,7 @@ public class MessageDeserializer implements Deserializer { INLONGMSG_ATTR_TIME_T + " or " + INLONGMSG_ATTR_TIME_DT)); } - String srcIp = Optional.ofNullable(attributes.get(INLONGMSG_ATTR_NODE_IP)) + String srcIp = Optional.ofNullable(attributes.get(INLONGMSG_ATTR_CLIENT_IP)) .orElse(DEFAULT_IP); Iterator<byte[]> iterator = inLongMsg.getIterator(attr);
