[INLONG-9783][Sort] Add compatibility processing of tid to streamId c…

…hanges in the message deserialization base class (#9785)
apache · Mar 7, 2024 · 03933e4 · 03933e4
1 parent 7c4adbf
commit 03933e4
Show file tree

Hide file tree

Showing 8 changed files with 42 additions and 13 deletions.
diff --git a/...ava/org/apache/inlong/sort/protocol/deserialization/InLongMsgCsv2DeserializationInfo.java b/...ava/org/apache/inlong/sort/protocol/deserialization/InLongMsgCsv2DeserializationInfo.java
@@ -17,7 +17,9 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonAlias;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -28,6 +30,7 @@
 /**
  * It represents CSV2 format of InLongMsg(m=9).
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class InLongMsgCsv2DeserializationInfo extends InLongMsgDeserializationInfo {
 
  private static final long serialVersionUID = 2188769102604850019L;
@@ -39,14 +42,14 @@ public class InLongMsgCsv2DeserializationInfo extends InLongMsgDeserializationIn
  private final Character escapeChar;
 
  public InLongMsgCsv2DeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter) {
  this(streamId, delimiter, null);
  }
 
  @JsonCreator
  public InLongMsgCsv2DeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter,
  @JsonProperty("escape_char") @Nullable Character escapeChar) {
  super(streamId);

diff --git a/...java/org/apache/inlong/sort/protocol/deserialization/InLongMsgCsvDeserializationInfo.java b/...java/org/apache/inlong/sort/protocol/deserialization/InLongMsgCsvDeserializationInfo.java
@@ -17,7 +17,9 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonAlias;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude.Include;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -29,6 +31,7 @@
 /**
  * It represents CSV format of InLongMsg(m=0).
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class InLongMsgCsvDeserializationInfo extends InLongMsgDeserializationInfo {
 
  private static final long serialVersionUID = 1499370571949888870L;
@@ -43,21 +46,21 @@ public class InLongMsgCsvDeserializationInfo extends InLongMsgDeserializationInf
  private final boolean deleteHeadDelimiter;
 
  public InLongMsgCsvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter) {
  this(streamId, delimiter, null, false);
  }
 
  public InLongMsgCsvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter,
  @JsonProperty("delete_head_delimiter") boolean deleteHeadDelimiter) {
  this(streamId, delimiter, null, deleteHeadDelimiter);
  }
 
  @JsonCreator
  public InLongMsgCsvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter,
  @JsonProperty("escape_char") @Nullable Character escapeChar,
  @JsonProperty("delete_head_delimiter") boolean deleteHeadDelimiter) {

diff --git a/...in/java/org/apache/inlong/sort/protocol/deserialization/InLongMsgDeserializationInfo.java b/...in/java/org/apache/inlong/sort/protocol/deserialization/InLongMsgDeserializationInfo.java
@@ -17,25 +17,33 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonAlias;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * InLongMsgDeserializationInfo.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public abstract class InLongMsgDeserializationInfo implements DeserializationInfo {
 
  private static final long serialVersionUID = 3707412713264864315L;
 
  private final String streamId;
 
- public InLongMsgDeserializationInfo(@JsonProperty("streamId") String streamId) {
+ public InLongMsgDeserializationInfo(@JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId) {
  this.streamId = checkNotNull(streamId);
  }
 
  @JsonProperty("streamId")
+ @JsonAlias(value = {"tid"})
  public String getStreamId() {
  return streamId;
  }
+
+ public String getTid() {
+ return streamId;
+ }
 }
diff --git a/.../java/org/apache/inlong/sort/protocol/deserialization/InLongMsgKvDeserializationInfo.java b/.../java/org/apache/inlong/sort/protocol/deserialization/InLongMsgKvDeserializationInfo.java
@@ -17,7 +17,9 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonAlias;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude.Include;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -29,6 +31,7 @@
 /**
  * It represents KV format of InLongMsg(m=5).
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class InLongMsgKvDeserializationInfo extends InLongMsgDeserializationInfo {
 
  private static final long serialVersionUID = 8431516458466278968L;
@@ -46,15 +49,15 @@ public class InLongMsgKvDeserializationInfo extends InLongMsgDeserializationInfo
  private final Character lineDelimiter;
 
  public InLongMsgKvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("entry_delimiter") char entryDelimiter,
  @JsonProperty("kv_delimiter") char kvDelimiter) {
  this(streamId, entryDelimiter, kvDelimiter, null, null);
  }
 
  @JsonCreator
  public InLongMsgKvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("entry_delimiter") char entryDelimiter,
  @JsonProperty("kv_delimiter") char kvDelimiter,
  @JsonProperty("escape_char") @Nullable Character escapeChar,

diff --git a/.../org/apache/inlong/sort/protocol/deserialization/InLongMsgTlogCsvDeserializationInfo.java b/.../org/apache/inlong/sort/protocol/deserialization/InLongMsgTlogCsvDeserializationInfo.java
@@ -17,7 +17,9 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonAlias;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude.Include;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -29,6 +31,7 @@
 /**
  * It represents TLog CSV format of InLongMsg(m=10).
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class InLongMsgTlogCsvDeserializationInfo extends InLongMsgDeserializationInfo {
 
  private static final long serialVersionUID = -6585242216925992303L;
@@ -40,14 +43,14 @@ public class InLongMsgTlogCsvDeserializationInfo extends InLongMsgDeserializatio
  private final Character escapeChar;
 
  public InLongMsgTlogCsvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter) {
  this(streamId, delimiter, null);
  }
 
  @JsonCreator
  public InLongMsgTlogCsvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter,
  @JsonProperty("escape_char") @Nullable Character escapeChar) {
  super(streamId);

diff --git a/...a/org/apache/inlong/sort/protocol/deserialization/InLongMsgTlogKvDeserializationInfo.java b/...a/org/apache/inlong/sort/protocol/deserialization/InLongMsgTlogKvDeserializationInfo.java
@@ -17,7 +17,9 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonAlias;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude.Include;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -29,6 +31,7 @@
 /**
  * It represents TLog KV format of InLongMsg(m=15).
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class InLongMsgTlogKvDeserializationInfo extends InLongMsgDeserializationInfo {
 
  private static final long serialVersionUID = 3299931901024581425L;
@@ -44,7 +47,7 @@ public class InLongMsgTlogKvDeserializationInfo extends InLongMsgDeserialization
  private final Character escapeChar;
 
  public InLongMsgTlogKvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter,
  @JsonProperty("entry_delimiter") char entryDelimiter,
  @JsonProperty("kv_delimiter") char kvDelimiter) {
@@ -53,7 +56,7 @@ public InLongMsgTlogKvDeserializationInfo(
 
  @JsonCreator
  public InLongMsgTlogKvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("delimiter") char delimiter,
  @JsonProperty("entry_delimiter") char entryDelimiter,
  @JsonProperty("kv_delimiter") char kvDelimiter,

diff --git a/...rc/main/java/org/apache/inlong/sort/protocol/deserialization/JsonDeserializationInfo.java b/...rc/main/java/org/apache/inlong/sort/protocol/deserialization/JsonDeserializationInfo.java
@@ -17,9 +17,12 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
 /**
  * Json deserialization info
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class JsonDeserializationInfo implements DeserializationInfo {
 
  private static final long serialVersionUID = -5344203248610337314L;

diff --git a/.../src/main/java/org/apache/inlong/sort/protocol/deserialization/KvDeserializationInfo.java b/.../src/main/java/org/apache/inlong/sort/protocol/deserialization/KvDeserializationInfo.java
@@ -18,7 +18,9 @@
 package org.apache.inlong.sort.protocol.deserialization;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonAlias;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -29,6 +31,7 @@
 /**
  * Kv deserialization info
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public class KvDeserializationInfo extends InLongMsgDeserializationInfo {
 
  private static final long serialVersionUID = -3182881360079888043L;
@@ -58,7 +61,7 @@ public KvDeserializationInfo(
 
  @JsonCreator
  public KvDeserializationInfo(
- @JsonProperty("streamId") String streamId,
+ @JsonProperty("streamId") @JsonAlias(value = {"tid"}) String streamId,
  @JsonProperty("entry_splitter") char entrySplitter,
  @JsonProperty("kv_splitter") char kvSplitter,
  @JsonProperty("escape_char") @Nullable Character escapeChar) {