fix flink-release 1.17 CoderTypeSerializer related bug

twitter-forks · Dec 1, 2023 · 8cbe2d7 · 8cbe2d7
1 parent 63fe1ba
commit 8cbe2d7
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 4 deletions.
diff --git a/...nk/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java b/...nk/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java
@@ -36,6 +36,8 @@
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * Flink {@link org.apache.flink.api.common.typeutils.TypeSerializer} for Beam {@link
@@ -230,14 +232,25 @@ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSeria
         return TypeSerializerSchemaCompatibility.incompatible();
       }
 
-      if (!this.serializer.pipelineOptions.toString().equals(coderTypeSerializer.pipelineOptions.toString())) {
+      if (this.serializer.coder.getCoderArguments().size() !=  coderTypeSerializer.coder.getCoderArguments().size()) {
+        return TypeSerializerSchemaCompatibility.incompatible();
+      }
+
+      if (!getCoderArguments(this.serializer.coder.getCoderArguments()).equals(getCoderArguments(coderTypeSerializer.coder.getCoderArguments()))) {
         return TypeSerializerSchemaCompatibility.incompatible();
       }
 
       return TypeSerializerSchemaCompatibility.compatibleAsIs();
     }
   }
 
+  private static List<String> getCoderArguments(List<? extends Coder<?>> coderArgList) {
+    return coderArgList.stream()
+        .map(coder -> coder.getEncodedTypeDescriptor().getType().getTypeName())
+        .sorted()
+        .collect(Collectors.toList());
+  }
+
   @Override
   public String toString() {
     return "CoderTypeSerializer{" + "coder=" + coder + '}';

diff --git a/...rc/test/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializerTest.java b/...rc/test/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializerTest.java
@@ -17,13 +17,11 @@
  */
 package org.apache.beam.runners.flink.translation.types;
 
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
-
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.Serializable;
+
 import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
 import org.apache.beam.sdk.coders.AtomicCoder;
 import org.apache.beam.sdk.coders.Coder;
@@ -33,6 +31,9 @@
 import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.junit.Test;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+
 /** Tests {@link CoderTypeSerializer}. */
 @SuppressWarnings({
   "rawtypes", // TODO(https://github.com/apache/beam/issues/20447)