You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2019/02/22 07:45:48 UTC
[flink] 17/17: [hotfix] [core] Subclass serializer cache in
PojoSerializer should be a Map
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 04ede2c66cb030a162436a1f6b1317fa4bdf34df
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Thu Feb 21 11:22:26 2019 +0800
[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.
---
.../apache/flink/api/java/typeutils/runtime/PojoSerializer.java | 8 ++++----
.../flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java | 2 +-
.../api/java/typeutils/runtime/PojoSerializerSnapshotData.java | 5 +++--
.../src/main/java/org/apache/flink/util/LinkedOptionalMap.java | 4 ++--
4 files changed, 10 insertions(+), 9 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
index fbaedc0..dcc0a72 100644
--- 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
@@ -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 final class PojoSerializer<T> extends TypeSerializer<T> {
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 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
return result;
}
- HashMap<Class<?>, TypeSerializer<?>> getSubclassSerializerCache() {
+ Map<Class<?>, TypeSerializer<?>> getSubclassSerializerCache() {
return subclassSerializerCache;
}
@@ -1110,7 +1110,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
TypeSerializer<?>[] registeredSubclassSerializers,
Field[] fields,
TypeSerializer<?>[] fieldSerializers,
- HashMap<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializerCache) {
+ Map<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializerCache) {
final LinkedHashMap<Class<?>, TypeSerializer<?>> subclassRegistry = new LinkedHashMap<>(registeredSubclassesToTags.size());
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java
index 7130134..5610536 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java
@@ -79,7 +79,7 @@ public class PojoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
Field[] fields,
TypeSerializer<?>[] fieldSerializers,
LinkedHashMap<Class<?>, TypeSerializer<?>> registeredSubclassSerializers,
- HashMap<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializers) {
+ Map<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializers) {
this.snapshotData = PojoSerializerSnapshotData.createFrom(
pojoClass,
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java
index e2efbdb..8999d77 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java
@@ -37,6 +37,7 @@ import java.io.IOException;
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 @@ final class PojoSerializerSnapshotData<T> {
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 @@ final class PojoSerializerSnapshotData<T> {
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<>(
diff --git a/flink-core/src/main/java/org/apache/flink/util/LinkedOptionalMap.java b/flink-core/src/main/java/org/apache/flink/util/LinkedOptionalMap.java
index a17dca4..db4318c 100644
--- a/flink-core/src/main/java/org/apache/flink/util/LinkedOptionalMap.java
+++ b/flink-core/src/main/java/org/apache/flink/util/LinkedOptionalMap.java
@@ -24,10 +24,10 @@ import org.apache.flink.annotation.VisibleForTesting;
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());