[INLONG-11863][SDK] Enhance Event Attribute Validation and Decoding Logic in SDK (#11864)

gosonzhang · gosonzhang · web-flow · commit 8a54dc480ce5 · 2025-05-15T10:38:24.000+08:00
Co-authored-by: gosonzhang &lt;gosonzhang@tencent.com&gt;
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/BaseMsgSenderFactory.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/BaseMsgSenderFactory.java
@@ -324,7 +324,7 @@ private boolean removeClusterIdSender(BaseSender msgSender, Map<String, BaseSend
     private int releaseAllGroupIdSenders(Map<String, BaseSender> senderMap) {
         int totalSenderCnt = 0;
         for (Map.Entry<String, BaseSender> entry : senderMap.entrySet()) {
-            if (entry == null || entry.getValue() == null) {
+            if (entry == null || entry.getKey() == null || entry.getValue() == null) {
                 continue;
             }
             try {
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/common/EventInfo.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/common/EventInfo.java
@@ -73,10 +73,12 @@ protected EventInfo(String groupId, String streamId, long dtMs, Long auditId, St
         // attrs
         if (attrs != null && !attrs.isEmpty()) {
             for (Map.Entry<String, String> entry : attrs.entrySet()) {
-                if (StringUtils.isBlank(entry.getKey())) {
+                if (entry == null
+                        || StringUtils.isBlank(entry.getKey())
+                        || entry.getValue() == null) {
                     continue;
                 }
-                innSetAttr(entry.getKey().trim(), entry.getValue());
+                innSetAttr(entry.getKey().trim(), entry.getValue().trim());
             }
         }
         if (auditId != null && auditId != -1L) {
@@ -129,20 +131,16 @@ protected void innSetAttr(String key, String value) throws ProxyEventException {
                     + AttributeConstants.KEY_VALUE_SEPARATOR + " or "
                     + AttributeConstants.KEY_VALUE_SEPARATOR + ")!");
         }
-        String valValue = value;
-        if (valValue != null) {
-            valValue = valValue.trim();
-            if (valValue.contains(AttributeConstants.SEPARATOR)
-                    || valValue.contains(AttributeConstants.KEY_VALUE_SEPARATOR)) {
-                if (exceptCnt.shouldPrint()) {
-                    logger.warn(String.format("Attribute value(%s) include reserved word(%s or %s)",
-                            valValue, AttributeConstants.KEY_VALUE_SEPARATOR, AttributeConstants.KEY_VALUE_SEPARATOR));
-                }
-                throw new ProxyEventException("Attribute value(" + valValue + ") include reserved word("
-                        + AttributeConstants.KEY_VALUE_SEPARATOR + " or "
-                        + AttributeConstants.KEY_VALUE_SEPARATOR + ")!");
+        if (value.contains(AttributeConstants.SEPARATOR)
+                || value.contains(AttributeConstants.KEY_VALUE_SEPARATOR)) {
+            if (exceptCnt.shouldPrint()) {
+                logger.warn(String.format("Attribute value(%s) include reserved word(%s or %s)",
+                        value, AttributeConstants.KEY_VALUE_SEPARATOR, AttributeConstants.KEY_VALUE_SEPARATOR));
             }
+            throw new ProxyEventException("Attribute value(" + value + ") include reserved word("
+                    + AttributeConstants.KEY_VALUE_SEPARATOR + " or "
+                    + AttributeConstants.KEY_VALUE_SEPARATOR + ")!");
         }
-        this.attrs.put(key, valValue);
+        this.attrs.put(key, value);
     }
 }
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/metric/MetaSyncInfo.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/metric/MetaSyncInfo.java
@@ -53,6 +53,11 @@ public void getAndResetValue(StringBuilder strBuff) {
             long curCnt = 0;
             strBuff.append("\"ms\":{\"errT\":{");
             for (Map.Entry<Integer, LongAdder> entry : syncErrInfo.entrySet()) {
+                if (entry == null
+                        || entry.getKey() == null
+                        || entry.getValue() == null) {
+                    continue;
+                }
                 if (curCnt++ > 0) {
                     strBuff.append(",");
                 }
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/metric/MetricDataHolder.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/metric/MetricDataHolder.java
@@ -421,6 +421,11 @@ public void getAndResetValue(StringBuilder strBuff) {
             metaSyncInfo.getAndResetValue(strBuff);
             strBuff.append(",\"tr\":[");
             for (Map.Entry<String, TrafficInfo> entry : trafficMap.entrySet()) {
+                if (entry == null
+                        || entry.getKey() == null
+                        || entry.getValue() == null) {
+                    continue;
+                }
                 if (count++ > 0) {
                     strBuff.append(",");
                 }
@@ -429,6 +434,11 @@ public void getAndResetValue(StringBuilder strBuff) {
             strBuff.append("],\"errs\":{");
             count = 0;
             for (Map.Entry<Integer, LongAdder> entry : errCodeMap.entrySet()) {
+                if (entry == null
+                        || entry.getKey() == null
+                        || entry.getValue() == null) {
+                    continue;
+                }
                 if (count++ > 0) {
                     strBuff.append(",");
                 }
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/metric/TimeCostInfo.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/metric/TimeCostInfo.java
@@ -78,6 +78,11 @@ public void getAndResetValue(StringBuilder strBuff) {
             long bucketCnt = 0;
             strBuff.append("\"").append(name).append("\":{\"bkts\":{");
             for (Map.Entry<String, LongAdder> entry : sendTimeBucketT.entrySet()) {
+                if (entry == null
+                        || entry.getKey() == null
+                        || entry.getValue() == null) {
+                    continue;
+                }
                 if (bucketCnt++ > 0) {
                     strBuff.append(",");
                 }
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/TcpClientMgr.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/TcpClientMgr.java
@@ -275,22 +275,22 @@ public boolean reportEvent(SendQos sendQos, TcpNettyClient client,
                     timerObj.newTimeout(new TimeoutTask(encObject.getMessageId()),
                             tcpConfig.getRequestTimeoutMs(), TimeUnit.MILLISECONDS));
             if (!client.write(clientTerm, encObject, procResult)) {
-                Timeout timeout = reqTimeouts.remove(encObject.getMessageId());
+                Timeout timeout = reqTimeouts.remove(newFuture.getMessageId());
                 if (timeout != null) {
                     timeout.cancel();
                 }
-                rmvMsgStubInfo(encObject.getMessageId());
+                rmvMsgStubInfo(newFuture.getMessageId());
             }
             return procResult.isSuccess();
         } else {
             // process sync report
             if (!client.write(clientTerm, encObject, procResult)) {
-                rmvMsgStubInfo(encObject.getMessageId());
+                rmvMsgStubInfo(newFuture.getMessageId());
                 return false;
             }
             boolean retValue = newFuture.get(procResult,
                     tcpConfig.getRequestTimeoutMs(), TimeUnit.MILLISECONDS);
-            if (rmvMsgStubInfo(encObject.getMessageId())) {
+            if (rmvMsgStubInfo(newFuture.getMessageId())) {
                 if (procResult.getErrCode() == ErrorCode.SEND_WAIT_TIMEOUT.getErrCode()) {
                     client.setBusy(clientTerm);
                 }
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/codec/DecodeObject.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/codec/DecodeObject.java
@@ -23,7 +23,6 @@
 import org.apache.inlong.sdk.dataproxy.common.ErrorCode;
 import org.apache.inlong.sdk.dataproxy.common.ProcessResult;
 
-import com.google.common.base.Splitter;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.HashMap;
@@ -36,10 +35,6 @@
  */
 public class DecodeObject {
 
-    private static final Splitter.MapSplitter MAP_SPLITTER =
-            Splitter.on(AttributeConstants.SEPARATOR).trimResults()
-                    .withKeyValueSeparator(AttributeConstants.KEY_VALUE_SEPARATOR);
-
     private final MsgType msgType;
     private int messageId;
     private String dpIp;
@@ -87,12 +82,30 @@ private void handleAttr(String attributes) {
             this.procResult = new ProcessResult(ErrorCode.OK);
             return;
         }
-        retAttr = new HashMap<>(MAP_SPLITTER.split(attributes));
+        // decode attribute string
+        this.retAttr = new HashMap<>();
+        String[] keyValSet = attributes.split(AttributeConstants.SEPARATOR);
+        for (String keyVal : keyValSet) {
+            if (StringUtils.isBlank(keyVal)) {
+                continue;
+            }
+            String[] keyValSplit = keyVal.split(AttributeConstants.KEY_VALUE_SEPARATOR);
+            if (keyValSplit.length == 1) {
+                if (StringUtils.isBlank(keyValSplit[0])) {
+                    continue;
+                }
+                retAttr.put(keyValSplit[0].trim(), "");
+            } else {
+                if (StringUtils.isBlank(keyValSplit[0]) || keyValSplit[1] == null) {
+                    continue;
+                }
+                retAttr.put(keyValSplit[0].trim(), keyValSplit[1].trim());
+            }
+        }
         if (retAttr.containsKey(AttributeConstants.MESSAGE_ID)) {
             this.messageId = Integer.parseInt(retAttr.get(AttributeConstants.MESSAGE_ID));
         }
         dpIp = retAttr.get(AttributeConstants.MESSAGE_DP_IP);
-
         String errCode = retAttr.get(AttributeConstants.MESSAGE_PROCESS_ERRCODE);
         // errCode is empty or equals 0 -> success
         if (StringUtils.isBlank(errCode) || Integer.parseInt(errCode) == 0) {
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/codec/EncodeObject.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/codec/EncodeObject.java
@@ -21,6 +21,7 @@
 import org.apache.inlong.common.msg.MsgType;
 
 import com.google.common.base.Joiner;
+import org.apache.commons.lang3.StringUtils;
 
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
@@ -86,10 +87,10 @@ public void setAttrInfo(int intMsgType, boolean isCompress, byte[] aesKey, Map<S
         this.aesKey = aesKey;
         if (tgtAttrs != null && !tgtAttrs.isEmpty()) {
             for (Map.Entry<String, String> entry : tgtAttrs.entrySet()) {
-                if (entry == null || entry.getKey() == null) {
+                if (entry == null || StringUtils.isBlank(entry.getKey()) || entry.getValue() == null) {
                     continue;
                 }
-                this.attrMap.put(entry.getKey(), entry.getValue());
+                this.attrMap.put(entry.getKey().trim(), entry.getValue().trim());
             }
             String preAttrStr = mapJoiner.join(this.attrMap);
             this.attrData = preAttrStr.getBytes(StandardCharsets.UTF_8);
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/sender/tcp/TcpEventInfo.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/sender/tcp/TcpEventInfo.java
@@ -79,9 +79,12 @@ public List<byte[]> getBodyList() {
 
     public void setAttr(String key, String value) throws ProxyEventException {
         if (StringUtils.isBlank(key)) {
-            throw new ProxyEventException("Key is blank!");
+            throw new ProxyEventException("Parameter key is blank!");
         }
-        innSetAttr(key.trim(), value);
+        if (value == null) {
+            throw new ProxyEventException("Parameter value is null!");
+        }
+        innSetAttr(key.trim(), value.trim());
     }
 
     @Override
diff --git a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/ProxyUtils.java b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/ProxyUtils.java
@@ -165,25 +165,25 @@ public static Map<String, String> getValidAttrs(Map<String, String> attrsMap) {
         if (attrsMap == null || attrsMap.isEmpty()) {
             return attrsMap;
         }
+        String tmpKey;
         String tmpValue;
         Map<String, String> validAttrsMap = new HashMap<>();
         for (Map.Entry<String, String> entry : attrsMap.entrySet()) {
-            if (StringUtils.isBlank(entry.getKey())
-                    || entry.getKey().contains(AttributeConstants.SEPARATOR)
-                    || entry.getKey().contains(AttributeConstants.KEY_VALUE_SEPARATOR)) {
+            if (entry == null
+                    || StringUtils.isBlank(entry.getKey())
+                    || entry.getValue() == null) {
                 continue;
             }
-            tmpValue = entry.getKey().trim();
-            if (ProxyUtils.SdkReservedWords.contains(tmpValue)) {
+            tmpKey = entry.getKey().trim();
+            tmpValue = entry.getValue().trim();
+            if (tmpKey.contains(AttributeConstants.SEPARATOR)
+                    || tmpKey.contains(AttributeConstants.KEY_VALUE_SEPARATOR)
+                    || ProxyUtils.SdkReservedWords.contains(tmpKey)
+                    || tmpValue.contains(AttributeConstants.SEPARATOR)
+                    || tmpValue.contains(AttributeConstants.KEY_VALUE_SEPARATOR)) {
                 continue;
             }
-            if (entry.getValue() != null) {
-                if (entry.getValue().contains(AttributeConstants.SEPARATOR)
-                        || entry.getValue().contains(AttributeConstants.KEY_VALUE_SEPARATOR)) {
-                    continue;
-                }
-            }
-            validAttrsMap.put(tmpValue, entry.getValue());
+            validAttrsMap.put(tmpKey, tmpValue);
         }
         return validAttrsMap;
     }
diff --git a/inlong-sdk/dataproxy-sdk/src/test/java/org/apache/inlong/sdk/dataproxy/ProxyUtilsTest.java b/inlong-sdk/dataproxy-sdk/src/test/java/org/apache/inlong/sdk/dataproxy/ProxyUtilsTest.java
@@ -17,11 +17,15 @@
 
 package org.apache.inlong.sdk.dataproxy;
 
+import org.apache.inlong.common.msg.AttributeConstants;
 import org.apache.inlong.sdk.dataproxy.utils.ProxyUtils;
 
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.HashMap;
+import java.util.Map;
+
 public class ProxyUtilsTest {
 
     @Test
@@ -30,4 +34,30 @@ public void getLocalIp() {
         Assert.assertNotNull(ip);
     }
 
+    @Test
+    public void testGetValidAttrs() {
+        Map<String, String> attrsMap = new HashMap<>();
+        attrsMap.put("first", "       ");
+        attrsMap.put("second", "");
+        attrsMap.put("third", " stst");
+        attrsMap.put("fourth", null);
+        attrsMap.put("fifth", " fifth ");
+        attrsMap.put("", "sixth");
+        attrsMap.put(null, "seventh");
+        attrsMap.put(null, "seventh");
+        attrsMap.put("eighth", "eig&hth");
+        attrsMap.put("ninth", "=ninth");
+        attrsMap.put(AttributeConstants.DATA_TIME, "tenth");
+        Map<String, String> tgtMap = ProxyUtils.getValidAttrs(attrsMap);
+        Assert.assertNotNull(tgtMap);
+        Assert.assertEquals(tgtMap.size(), 4);
+        Assert.assertNotNull(tgtMap.get("first"));
+        Assert.assertNotNull(tgtMap.get("second"));
+        Assert.assertNotNull(tgtMap.get("third"));
+        Assert.assertNull(tgtMap.get("fourth"));
+        Assert.assertNotNull(tgtMap.get("fifth"));
+        Assert.assertNull(tgtMap.get("eighth"));
+        Assert.assertNull(tgtMap.get("ninth"));
+        Assert.assertNull(tgtMap.get(AttributeConstants.DATA_TIME));
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -324,7 +324,7 @@ private boolean removeClusterIdSender(BaseSender msgSender, Map<String, BaseSend`
`324`	`324`	`private int releaseAllGroupIdSenders(Map<String, BaseSender> senderMap) {`
`325`	`325`	`int totalSenderCnt = 0;`
`326`	`326`	`for (Map.Entry<String, BaseSender> entry : senderMap.entrySet()) {`
`327`		`- if (entry == null \|\| entry.getValue() == null) {`
	`327`	`+ if (entry == null \|\| entry.getKey() == null \|\| entry.getValue() == null) {`
`328`	`328`	`continue;`
`329`	`329`	`}`
`330`	`330`	`try {`
Original file line number	Diff line number	Diff line change
`@@ -73,10 +73,12 @@ protected EventInfo(String groupId, String streamId, long dtMs, Long auditId, St`
`73`	`73`	`// attrs`
`74`	`74`	`if (attrs != null && !attrs.isEmpty()) {`
`75`	`75`	`for (Map.Entry<String, String> entry : attrs.entrySet()) {`
`76`		`- if (StringUtils.isBlank(entry.getKey())) {`
	`76`	`+ if (entry == null`
	`77`	`+ \|\| StringUtils.isBlank(entry.getKey())`
	`78`	`+ \|\| entry.getValue() == null) {`
`77`	`79`	`continue;`
`78`	`80`	`}`
`79`		`- innSetAttr(entry.getKey().trim(), entry.getValue());`
	`81`	`+ innSetAttr(entry.getKey().trim(), entry.getValue().trim());`
`80`	`82`	`}`
`81`	`83`	`}`
`82`	`84`	`if (auditId != null && auditId != -1L) {`
`@@ -129,20 +131,16 @@ protected void innSetAttr(String key, String value) throws ProxyEventException {`
`129`	`131`	`+ AttributeConstants.KEY_VALUE_SEPARATOR + " or "`
`130`	`132`	`+ AttributeConstants.KEY_VALUE_SEPARATOR + ")!");`
`131`	`133`	`}`
`132`		`- String valValue = value;`
`133`		`- if (valValue != null) {`
`134`		`- valValue = valValue.trim();`
`135`		`- if (valValue.contains(AttributeConstants.SEPARATOR)`
`136`		`- \|\| valValue.contains(AttributeConstants.KEY_VALUE_SEPARATOR)) {`
`137`		`- if (exceptCnt.shouldPrint()) {`
`138`		`- logger.warn(String.format("Attribute value(%s) include reserved word(%s or %s)",`
`139`		`- valValue, AttributeConstants.KEY_VALUE_SEPARATOR, AttributeConstants.KEY_VALUE_SEPARATOR));`
`140`		`- }`
`141`		`- throw new ProxyEventException("Attribute value(" + valValue + ") include reserved word("`
`142`		`- + AttributeConstants.KEY_VALUE_SEPARATOR + " or "`
`143`		`- + AttributeConstants.KEY_VALUE_SEPARATOR + ")!");`
	`134`	`+ if (value.contains(AttributeConstants.SEPARATOR)`
	`135`	`+ \|\| value.contains(AttributeConstants.KEY_VALUE_SEPARATOR)) {`
	`136`	`+ if (exceptCnt.shouldPrint()) {`
	`137`	`+ logger.warn(String.format("Attribute value(%s) include reserved word(%s or %s)",`
	`138`	`+ value, AttributeConstants.KEY_VALUE_SEPARATOR, AttributeConstants.KEY_VALUE_SEPARATOR));`
`144`	`139`	`}`
	`140`	`+ throw new ProxyEventException("Attribute value(" + value + ") include reserved word("`
	`141`	`+ + AttributeConstants.KEY_VALUE_SEPARATOR + " or "`
	`142`	`+ + AttributeConstants.KEY_VALUE_SEPARATOR + ")!");`
`145`	`143`	`}`
`146`		`- this.attrs.put(key, valValue);`
	`144`	`+ this.attrs.put(key, value);`
`147`	`145`	`}`
`148`	`146`	`}`
Original file line number	Diff line number	Diff line change
`@@ -79,9 +79,12 @@ public List<byte[]> getBodyList() {`
`79`	`79`
`80`	`80`	`public void setAttr(String key, String value) throws ProxyEventException {`
`81`	`81`	`if (StringUtils.isBlank(key)) {`
`82`		`- throw new ProxyEventException("Key is blank!");`
	`82`	`+ throw new ProxyEventException("Parameter key is blank!");`
`83`	`83`	`}`
`84`		`- innSetAttr(key.trim(), value);`
	`84`	`+ if (value == null) {`
	`85`	`+ throw new ProxyEventException("Parameter value is null!");`
	`86`	`+ }`
	`87`	`+ innSetAttr(key.trim(), value.trim());`
`85`	`88`	`}`
`86`	`89`
`87`	`90`	`@Override`