[FLINK-26835][serialization] Fix concurrent modification exception

silenceland · Apr 7, 2022 · 3ec3868 · 3ec3868
1 parent 1465633
commit 3ec3868
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 16 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
@@ -1040,7 +1040,7 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
         }
 
         cl = Thread.currentThread().getContextClassLoader();
-        subclassSerializerCache = new HashMap<Class<?>, TypeSerializer<?>>();
+        subclassSerializerCache = new HashMap<>();
     }
 
     // --------------------------------------------------------------------------------------------

diff --git a/...e/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeSerializerFactory.java b/...e/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeSerializerFactory.java
@@ -36,8 +36,6 @@ public final class RuntimeSerializerFactory<T>
 
     private TypeSerializer<T> serializer;
 
-    private boolean firstSerializer = true;
-
     private Class<T> clazz;
 
     // Because we read the class from the TaskConfig and instantiate ourselves
@@ -62,7 +60,6 @@ public void writeParametersToConfig(Configuration config) {
         }
     }
 
-    @SuppressWarnings("unchecked")
     @Override
     public void readParametersFromConfig(Configuration config, ClassLoader cl)
             throws ClassNotFoundException {
@@ -71,12 +68,8 @@ public void readParametersFromConfig(Configuration config, ClassLoader cl)
         }
 
         try {
-            this.clazz =
-                    (Class<T>) InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY_CLASS, cl);
-            this.serializer =
-                    (TypeSerializer<T>)
-                            InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY_SER, cl);
-            firstSerializer = true;
+            this.clazz = InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY_CLASS, cl);
+            this.serializer = InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY_SER, cl);
         } catch (ClassNotFoundException e) {
             throw e;
         } catch (Exception e) {
@@ -87,12 +80,7 @@ public void readParametersFromConfig(Configuration config, ClassLoader cl)
     @Override
     public TypeSerializer<T> getSerializer() {
         if (this.serializer != null) {
-            if (firstSerializer) {
-                firstSerializer = false;
-                return this.serializer;
-            } else {
-                return this.serializer.duplicate();
-            }
+            return this.serializer.duplicate();
         } else {
             throw new RuntimeException(
                     "SerializerFactory has not been initialized from configuration.");