[hotfix] [core] Subclass serializer cache in PojoSerializer should be…

… a Map

In this case, there is no specific need to let that cache be
specifically a HashMap. Changing this to the generic Map does not break
contracts with how the PojoSerializer works. It also doesn't break
compatibility w.r.t. Java serialization of the PojoSerializer.

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java

@@ -92,7 +92,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
 	/**
 	 * Cache of non-registered subclasses to their serializers, created on-the-fly.
 	 */
-	private transient HashMap<Class<?>, TypeSerializer<?>> subclassSerializerCache;
+	private transient Map<Class<?>, TypeSerializer<?>> subclassSerializerCache;
 
 	// --------------------------------------------------------------------------------------------
 
@@ -145,7 +145,7 @@ public PojoSerializer(
 			TypeSerializer<Object>[] fieldSerializers,
 			LinkedHashMap<Class<?>, Integer> registeredClasses,
 			TypeSerializer<?>[] registeredSerializers,
-			HashMap<Class<?>, TypeSerializer<?>> subclassSerializerCache,
+			Map<Class<?>, TypeSerializer<?>> subclassSerializerCache,
 			ExecutionConfig executionConfig) {
 
 		this.clazz = checkNotNull(clazz);
@@ -987,7 +987,7 @@ LinkedHashMap<Class<?>, TypeSerializer<?>> getBundledSubclassSerializerRegistry(
 		return result;
 	}
 
-	HashMap<Class<?>, TypeSerializer<?>> getSubclassSerializerCache() {
+	Map<Class<?>, TypeSerializer<?>> getSubclassSerializerCache() {
 		return subclassSerializerCache;
 	}
 
@@ -1110,7 +1110,7 @@ private static <T> PojoSerializerSnapshot<T> buildSnapshot(
 			TypeSerializer<?>[] registeredSubclassSerializers,
 			Field[] fields,
 			TypeSerializer<?>[] fieldSerializers,
-			HashMap<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializerCache) {
+			Map<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializerCache) {
 
 		final LinkedHashMap<Class<?>, TypeSerializer<?>> subclassRegistry = new LinkedHashMap<>(registeredSubclassesToTags.size());
 

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java

@@ -79,7 +79,7 @@ public PojoSerializerSnapshot() {}
 			Field[] fields,
 			TypeSerializer<?>[] fieldSerializers,
 			LinkedHashMap<Class<?>, TypeSerializer<?>> registeredSubclassSerializers,
-			HashMap<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializers) {
+			Map<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializers) {
 
 		this.snapshotData = PojoSerializerSnapshotData.createFrom(
 			pojoClass,

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java

@@ -37,6 +37,7 @@
 import java.lang.reflect.Field;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
+import java.util.Map;
 
 import static org.apache.flink.util.LinkedOptionalMap.optionalMapOf;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -86,7 +87,7 @@ static <T> PojoSerializerSnapshotData<T> createFrom(
 			Field[] fields,
 			TypeSerializer<?>[] fieldSerializers,
 			LinkedHashMap<Class<?>, TypeSerializer<?>> registeredSubclassSerializers,
-			HashMap<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializers) {
+			Map<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializers) {
 
 		final LinkedOptionalMap<Field, TypeSerializerSnapshot<?>> fieldSerializerSnapshots = new LinkedOptionalMap<>(fields.length);
 
@@ -99,7 +100,7 @@ static <T> PojoSerializerSnapshotData<T> createFrom(
 		LinkedHashMap<Class<?>, TypeSerializerSnapshot<?>> registeredSubclassSerializerSnapshots = new LinkedHashMap<>(registeredSubclassSerializers.size());
 		registeredSubclassSerializers.forEach((k, v) -> registeredSubclassSerializerSnapshots.put(k, TypeSerializerUtils.snapshotBackwardsCompatible(v)));
 
-		HashMap<Class<?>, TypeSerializerSnapshot<?>> nonRegisteredSubclassSerializerSnapshots = new HashMap<>(nonRegisteredSubclassSerializers.size());
+		Map<Class<?>, TypeSerializerSnapshot<?>> nonRegisteredSubclassSerializerSnapshots = new HashMap<>(nonRegisteredSubclassSerializers.size());
 		nonRegisteredSubclassSerializers.forEach((k, v) -> nonRegisteredSubclassSerializerSnapshots.put(k, TypeSerializerUtils.snapshotBackwardsCompatible(v)));
 
 		return new PojoSerializerSnapshotData<>(

flink-core/src/main/java/org/apache/flink/util/LinkedOptionalMap.java

@@ -24,10 +24,10 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
-import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
+import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
@@ -59,7 +59,7 @@ public final class LinkedOptionalMap<K, V> {
 	 * @param <V>           value type
 	 * @return an {@code LinkedOptionalMap} with optional named keys, and optional values.
 	 */
-	public static <K, V> LinkedOptionalMap<K, V> optionalMapOf(HashMap<K, V> sourceMap, Function<K, String> keyNameGetter) {
+	public static <K, V> LinkedOptionalMap<K, V> optionalMapOf(Map<K, V> sourceMap, Function<K, String> keyNameGetter) {
 
 		LinkedHashMap<String, KeyValue<K, V>> underlyingMap = new LinkedHashMap<>(sourceMap.size());