This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 23490b26e5 [INLONG-8504][Manager] Modify nodeIp in dataproxy message
to clientIp (#8519)
23490b26e5 is described below
commit 23490b26e5964062d711b1a65f9a2285de7d74fb
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);
