This is an automated email from the ASF dual-hosted git repository.
baomingyu 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 2f462677d5 [INLONG-9879][Sort] Sort inlongmsg-rowdata-kv format parse
head has error (#9880)
2f462677d5 is described below
commit 2f462677d56202fd0e8088304d8d1f825fef161c
Author: baomingyu <[email protected]>
AuthorDate: Thu Mar 28 16:15:09 2024 +0800
[INLONG-9879][Sort] Sort inlongmsg-rowdata-kv format parse head has error
(#9880)
---
.../inlong/sort/formats/inlongmsg/InLongMsgUtils.java | 2 ++
.../formats/inlongmsgbinlog/InLongMsgBinlogUtils.java | 7 ++++++-
.../sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java | 4 ++++
.../sort/formats/inlongmsgkv/InLongMsgKvUtils.java | 3 +++
.../inlong/sort/formats/inlongmsg/InLongMsgUtils.java | 2 +-
.../formats/inlongmsgbinlog/InLongMsgBinlogUtils.java | 16 ++++++++++------
.../sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java | 18 +++++++++---------
.../sort/formats/inlongmsgkv/InLongMsgKvUtils.java | 17 ++++++++---------
8 files changed, 43 insertions(+), 26 deletions(-)
diff --git
a/inlong-sort/sort-formats/format-row/format-inlongmsg-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
b/inlong-sort/sort-formats/format-row/format-inlongmsg-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
index 131c8099ec..39a05b6986 100644
---
a/inlong-sort/sort-formats/format-row/format-inlongmsg-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
+++
b/inlong-sort/sort-formats/format-row/format-inlongmsg-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
@@ -63,6 +63,8 @@ public class InLongMsgUtils {
// keys in attributes
public static final String INLONGMSG_ATTR_STREAM_ID = "streamId";
+ @Deprecated
+ public static final String INLONGMSG_ATTR_INTERFACE_NAME = "iname";
@Deprecated
public static final String INLONGMSG_ATTR_TID = "tid";
public static final String INLONGMSG_ATTR_TIME_T = "t";
diff --git
a/inlong-sort/sort-formats/format-row/format-inlongmsg-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
b/inlong-sort/sort-formats/format-row/format-inlongmsg-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
index e9bb0a5a5d..b2a6a26d17 100644
---
a/inlong-sort/sort-formats/format-row/format-inlongmsg-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
+++
b/inlong-sort/sort-formats/format-row/format-inlongmsg-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
@@ -46,6 +46,7 @@ import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_AT
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.DEFAULT_TIME_FIELD_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.FORMAT_ATTRIBUTES_FIELD_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.FORMAT_TIME_FIELD_NAME;
+import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.getPredefinedFields;
@@ -117,10 +118,14 @@ public class InLongMsgBinlogUtils {
streamId = attributes.get(INLONGMSG_ATTR_STREAM_ID);
} else if (attributes.containsKey(INLONGMSG_ATTR_TID)) {
streamId = attributes.get(INLONGMSG_ATTR_TID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
+ streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_NAME);
} else {
throw new IllegalArgumentException(
"Could not find " + INLONGMSG_ATTR_STREAM_ID
- + " or " + INLONGMSG_ATTR_TID + " in attributes!");
+ + " or " + INLONGMSG_ATTR_TID
+ + " or " + INLONGMSG_ATTR_INTERFACE_NAME
+ + " in attributes!");
}
// Extracts time from the attributes
diff --git
a/inlong-sort/sort-formats/format-row/format-inlongmsg-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
b/inlong-sort/sort-formats/format-row/format-inlongmsg-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
index cad6ae9dc0..fdccdc2b61 100644
---
a/inlong-sort/sort-formats/format-row/format-inlongmsg-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
+++
b/inlong-sort/sort-formats/format-row/format-inlongmsg-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
@@ -35,6 +35,7 @@ import java.util.Map;
import java.util.stream.Collectors;
import static
org.apache.inlong.sort.formats.base.TableFormatUtils.deserializeBasicField;
+import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_DT;
@@ -65,10 +66,13 @@ public class InLongMsgCsvUtils {
streamId = attributes.get(INLONGMSG_ATTR_STREAM_ID);
} else if (attributes.containsKey(INLONGMSG_ATTR_TID)) {
streamId = attributes.get(INLONGMSG_ATTR_TID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
+ streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_NAME);
} else {
throw new IllegalArgumentException(
"Could not find " + INLONGMSG_ATTR_STREAM_ID
+ " or " + INLONGMSG_ATTR_TID
+ + " or " + INLONGMSG_ATTR_INTERFACE_NAME
+ " in attributes!");
}
diff --git
a/inlong-sort/sort-formats/format-row/format-inlongmsg-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
b/inlong-sort/sort-formats/format-row/format-inlongmsg-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
index b25115fdfe..7fb7966992 100644
---
a/inlong-sort/sort-formats/format-row/format-inlongmsg-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
+++
b/inlong-sort/sort-formats/format-row/format-inlongmsg-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
@@ -32,6 +32,7 @@ import java.util.Map;
import java.util.stream.Collectors;
import static
org.apache.inlong.sort.formats.base.TableFormatUtils.deserializeBasicField;
+import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_DT;
@@ -55,6 +56,8 @@ public class InLongMsgKvUtils {
String streamId;
if (attributes.containsKey(INLONGMSG_ATTR_STREAM_ID)) {
streamId = attributes.get(INLONGMSG_ATTR_STREAM_ID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
+ streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_NAME);
} else if (attributes.containsKey(INLONGMSG_ATTR_TID)) {
streamId = attributes.get(INLONGMSG_ATTR_TID);
} else {
diff --git
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
index 1fd82e0d82..fae71b09f8 100644
---
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
+++
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/InLongMsgUtils.java
@@ -68,7 +68,7 @@ public class InLongMsgUtils {
public static final String INLONGMSG_ATTR_INTERFACE_ID = "id";
@Deprecated
public static final String INLONGMSG_ATTR_INTERFACE_TID = "tid";
- public static final String INLONGMSG_ATTR_STREAMID = "streamId";
+ public static final String INLONGMSG_ATTR_STREAM_ID = "streamId";
public static final String INLONGMSG_ATTR_TIME_T = "t";
public static final String INLONGMSG_ATTR_TIME_DT = "dt";
public static final String INLONGMSG_ATTR_ADD_COLUMN_PREFIX = "__addcol";
diff --git
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
index 6357482946..06d7174bb0 100644
---
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
+++
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-binlog/src/main/java/org/apache/inlong/sort/formats/inlongmsgbinlog/InLongMsgBinlogUtils.java
@@ -46,6 +46,7 @@ import static
org.apache.inlong.sort.formats.base.TableFormatUtils.deserializeBa
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_TID;
+import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.getPredefinedFields;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.parseAttr;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.parseEpochTime;
@@ -73,17 +74,20 @@ public class InLongMsgBinlogUtils {
// Extracts interface from the attributes.
String streamId;
- if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
+ if (attributes.containsKey(INLONGMSG_ATTR_STREAM_ID)) {
+ streamId = attributes.get(INLONGMSG_ATTR_STREAM_ID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_TID)) {
+ streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_TID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_NAME);
} else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_ID)) {
streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_ID);
- } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_TID)) {
- streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_TID);
} else {
throw new IllegalArgumentException(
- "Could not find " + INLONGMSG_ATTR_INTERFACE_NAME +
- " or " + INLONGMSG_ATTR_INTERFACE_ID +
- " or " + INLONGMSG_ATTR_INTERFACE_TID + " in
attributes!");
+ "Could not find " + INLONGMSG_ATTR_STREAM_ID
+ + " or " + INLONGMSG_ATTR_INTERFACE_TID
+ + " or " + INLONGMSG_ATTR_INTERFACE_NAME
+ + " or " + INLONGMSG_ATTR_INTERFACE_ID + " in
attributes!");
}
// Extracts time from the attributes
diff --git
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
index 24883a65bd..cd737eb1b1 100644
---
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
+++
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-csv/src/main/java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvUtils.java
@@ -40,7 +40,7 @@ import static
org.apache.inlong.sort.formats.base.TableFormatUtils.deserializeBa
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_TID;
-import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAMID;
+import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_DT;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_T;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.getPredefinedFields;
@@ -65,20 +65,20 @@ public class InLongMsgCsvUtils {
// Extracts interface from the attributes.
String streamId;
- if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
+ if (attributes.containsKey(INLONGMSG_ATTR_STREAM_ID)) {
+ streamId = attributes.get(INLONGMSG_ATTR_STREAM_ID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_TID)) {
+ streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_TID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_NAME);
} else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_ID)) {
streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_ID);
- } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_TID)) {
- streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_TID);
- } else if (attributes.containsKey(INLONGMSG_ATTR_STREAMID)) {
- streamId = attributes.get(INLONGMSG_ATTR_STREAMID);
} else {
throw new IllegalArgumentException(
- "Could not find " + INLONGMSG_ATTR_INTERFACE_NAME +
- " or " + INLONGMSG_ATTR_INTERFACE_ID +
+ "Could not find " + INLONGMSG_ATTR_STREAM_ID +
" or " + INLONGMSG_ATTR_INTERFACE_TID +
- " or " + INLONGMSG_ATTR_STREAMID + " in
attributes!");
+ " or " + INLONGMSG_ATTR_INTERFACE_NAME +
+ " or " + INLONGMSG_ATTR_INTERFACE_ID + " in
attributes!");
}
// Extracts time from the attributes
diff --git
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
index c6f1eaf771..dd9cf95512 100644
---
a/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
+++
b/inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvUtils.java
@@ -35,7 +35,7 @@ import java.util.stream.Collectors;
import static
org.apache.inlong.sort.formats.base.TableFormatUtils.deserializeBasicField;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_NAME;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_INTERFACE_TID;
-import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAMID;
+import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_STREAM_ID;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_DT;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.INLONGMSG_ATTR_TIME_T;
import static
org.apache.inlong.sort.formats.inlongmsg.InLongMsgUtils.getPredefinedFields;
@@ -53,18 +53,17 @@ public class InLongMsgKvUtils {
Map<String, String> attributes = parseAttr(attr);
String streamId;
- if (attributes.containsKey(INLONGMSG_ATTR_STREAMID)) {
- streamId = attributes.get(INLONGMSG_ATTR_STREAMID);
- }
- if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
- streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_NAME);
+ if (attributes.containsKey(INLONGMSG_ATTR_STREAM_ID)) {
+ streamId = attributes.get(INLONGMSG_ATTR_STREAM_ID);
} else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_TID)) {
streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_TID);
+ } else if (attributes.containsKey(INLONGMSG_ATTR_INTERFACE_NAME)) {
+ streamId = attributes.get(INLONGMSG_ATTR_INTERFACE_NAME);
} else {
throw new IllegalArgumentException(
- "Could not find " + INLONGMSG_ATTR_STREAMID
- + " or " + INLONGMSG_ATTR_INTERFACE_NAME
- + " or " + INLONGMSG_ATTR_INTERFACE_TID + " in
attributes!");
+ "Could not find " + INLONGMSG_ATTR_STREAM_ID
+ + " or " + INLONGMSG_ATTR_INTERFACE_TID
+ + " or " + INLONGMSG_ATTR_INTERFACE_NAME + " in
attributes!");
}
Timestamp time;
