You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/02/20 09:04:26 UTC
[flink] branch master updated: [FLINK-9803] Drop canEqual() from
TypeSerializer
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 09bb7bb [FLINK-9803] Drop canEqual() from TypeSerializer
09bb7bb is described below
commit 09bb7bbc0f2535dab90c59a3362dfe53a70055ef
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Fri Feb 8 15:10:38 2019 +0100
[FLINK-9803] Drop canEqual() from TypeSerializer
This change removes TypeSerializer.canEqual() because it is not useful
on that class hierachy. Serializers can only equals() on an exact
match, not with different serializers up and down the hierarchy.
This adds a serialVersionUID to InstantSerializer, to ensure that it
doesn't change from removing the canEqual() method.
Plus we also remove canEqual() from PojoField, where it is not needed.
---
.../connectors/kafka/FlinkKafkaProducer011.java | 10 ----------
.../streaming/connectors/kafka/FlinkKafkaProducer.java | 10 ----------
.../api/java/typeutils/runtime/WritableSerializer.java | 7 +------
.../flink/api/common/typeutils/CompositeSerializer.java | 15 +++------------
.../flink/api/common/typeutils/TypeDeserializer.java | 8 --------
.../api/common/typeutils/TypeDeserializerAdapter.java | 4 ----
.../flink/api/common/typeutils/TypeSerializer.java | 8 --------
.../common/typeutils/UnloadableDummyTypeSerializer.java | 5 -----
.../api/common/typeutils/base/BigDecSerializer.java | 5 -----
.../api/common/typeutils/base/BigIntSerializer.java | 5 -----
.../api/common/typeutils/base/BooleanSerializer.java | 5 -----
.../common/typeutils/base/BooleanValueSerializer.java | 5 -----
.../flink/api/common/typeutils/base/ByteSerializer.java | 5 -----
.../api/common/typeutils/base/ByteValueSerializer.java | 5 -----
.../flink/api/common/typeutils/base/CharSerializer.java | 5 -----
.../api/common/typeutils/base/CharValueSerializer.java | 5 -----
.../flink/api/common/typeutils/base/DateSerializer.java | 5 -----
.../api/common/typeutils/base/DoubleSerializer.java | 5 -----
.../common/typeutils/base/DoubleValueSerializer.java | 5 -----
.../flink/api/common/typeutils/base/EnumSerializer.java | 7 +------
.../api/common/typeutils/base/FloatSerializer.java | 5 -----
.../api/common/typeutils/base/FloatValueSerializer.java | 5 -----
.../common/typeutils/base/GenericArraySerializer.java | 8 +-------
.../api/common/typeutils/base/InstantSerializer.java | 8 +++-----
.../flink/api/common/typeutils/base/IntSerializer.java | 5 -----
.../api/common/typeutils/base/IntValueSerializer.java | 5 -----
.../flink/api/common/typeutils/base/ListSerializer.java | 5 -----
.../flink/api/common/typeutils/base/LongSerializer.java | 5 -----
.../api/common/typeutils/base/LongValueSerializer.java | 5 -----
.../flink/api/common/typeutils/base/MapSerializer.java | 5 -----
.../api/common/typeutils/base/NullValueSerializer.java | 5 -----
.../api/common/typeutils/base/ShortSerializer.java | 5 -----
.../api/common/typeutils/base/ShortValueSerializer.java | 5 -----
.../api/common/typeutils/base/SqlDateSerializer.java | 5 -----
.../api/common/typeutils/base/SqlTimeSerializer.java | 5 -----
.../common/typeutils/base/SqlTimestampSerializer.java | 5 -----
.../api/common/typeutils/base/StringSerializer.java | 5 -----
.../common/typeutils/base/StringValueSerializer.java | 5 -----
.../common/typeutils/base/TypeSerializerSingleton.java | 8 +-------
.../flink/api/common/typeutils/base/VoidSerializer.java | 5 -----
.../base/array/BooleanPrimitiveArraySerializer.java | 5 -----
.../base/array/BytePrimitiveArraySerializer.java | 5 -----
.../base/array/CharPrimitiveArraySerializer.java | 5 -----
.../base/array/DoublePrimitiveArraySerializer.java | 5 -----
.../base/array/FloatPrimitiveArraySerializer.java | 5 -----
.../base/array/IntPrimitiveArraySerializer.java | 5 -----
.../base/array/LongPrimitiveArraySerializer.java | 5 -----
.../base/array/ShortPrimitiveArraySerializer.java | 5 -----
.../typeutils/base/array/StringArraySerializer.java | 5 -----
.../org/apache/flink/api/java/typeutils/PojoField.java | 6 +-----
.../java/typeutils/runtime/CopyableValueSerializer.java | 11 ++---------
.../api/java/typeutils/runtime/EitherSerializer.java | 8 +-------
.../api/java/typeutils/runtime/NullableSerializer.java | 6 ------
.../api/java/typeutils/runtime/PojoSerializer.java | 8 +-------
.../flink/api/java/typeutils/runtime/RowSerializer.java | 7 +------
.../api/java/typeutils/runtime/Tuple0Serializer.java | 11 -----------
.../api/java/typeutils/runtime/TupleSerializerBase.java | 8 +-------
.../api/java/typeutils/runtime/ValueSerializer.java | 7 +------
.../api/java/typeutils/runtime/kryo/KryoSerializer.java | 8 +-------
.../typeutils/CompositeTypeSerializerSnapshotTest.java | 17 ++---------------
.../api/common/typeutils/TypeInformationTestBase.java | 1 -
.../typeutils/TypeSerializerSerializationUtilTest.java | 5 -----
.../common/typeutils/TypeSerializerSnapshotTest.java | 5 -----
.../StatefulComplexPayloadSerializer.java | 5 -----
.../flink/formats/avro/typeutils/AvroSerializer.java | 5 -----
.../flink/api/java/io/CollectionInputFormatTest.java | 7 +------
.../main/java/org/apache/flink/cep/nfa/DeweyNumber.java | 5 -----
.../src/main/java/org/apache/flink/cep/nfa/NFA.java | 5 -----
.../org/apache/flink/cep/nfa/NFAStateSerializer.java | 5 -----
.../java/org/apache/flink/cep/nfa/SharedBuffer.java | 5 -----
.../org/apache/flink/cep/nfa/sharedbuffer/EventId.java | 5 -----
.../org/apache/flink/cep/nfa/sharedbuffer/Lockable.java | 5 -----
.../org/apache/flink/cep/nfa/sharedbuffer/NodeId.java | 5 -----
.../flink/cep/nfa/sharedbuffer/SharedBufferEdge.java | 5 -----
.../flink/cep/nfa/sharedbuffer/SharedBufferNode.java | 5 -----
.../drivers/transform/LongValueWithProperHashCode.java | 5 -----
.../types/valuearray/ByteValueArraySerializer.java | 5 -----
.../types/valuearray/CharValueArraySerializer.java | 5 -----
.../types/valuearray/DoubleValueArraySerializer.java | 5 -----
.../types/valuearray/FloatValueArraySerializer.java | 5 -----
.../graph/types/valuearray/IntValueArraySerializer.java | 5 -----
.../types/valuearray/LongValueArraySerializer.java | 5 -----
.../types/valuearray/NullValueArraySerializer.java | 5 -----
.../types/valuearray/ShortValueArraySerializer.java | 5 -----
.../types/valuearray/StringValueArraySerializer.java | 5 -----
.../queryablestate/client/VoidNamespaceSerializer.java | 5 -----
.../apache/flink/runtime/state/ArrayListSerializer.java | 5 -----
.../org/apache/flink/runtime/state/JavaSerializer.java | 5 -----
.../flink/runtime/state/VoidNamespaceSerializer.java | 5 -----
.../operators/testutils/types/IntListSerializer.java | 11 -----------
.../operators/testutils/types/IntPairSerializer.java | 11 -----------
.../operators/testutils/types/StringPairSerializer.java | 11 -----------
.../apache/flink/runtime/query/KvStateRegistryTest.java | 5 -----
.../runtime/state/InternalPriorityQueueTestBase.java | 5 -----
.../flink/runtime/state/OperatorStateBackendTest.java | 9 ---------
.../runtime/state/StateBackendMigrationTestBase.java | 5 -----
.../runtime/state/StateSnapshotTransformerTest.java | 7 -------
.../runtime/state/heap/TestDuplicateSerializer.java | 5 -----
.../runtime/testutils/recordutils/RecordSerializer.java | 10 ----------
.../runtime/testutils/statemigration/TestType.java | 5 -----
.../flink/api/scala/typeutils/EitherSerializer.scala | 5 -----
.../flink/api/scala/typeutils/EnumValueSerializer.scala | 6 +-----
.../flink/api/scala/typeutils/NothingSerializer.scala | 6 +-----
.../flink/api/scala/typeutils/OptionSerializer.scala | 6 +-----
.../api/scala/typeutils/TraversableSerializer.scala | 6 +-----
.../flink/api/scala/typeutils/TrySerializer.scala | 6 +-----
.../flink/api/scala/typeutils/UnitSerializer.scala | 4 ----
.../streaming/api/datastream/CoGroupedStreams.java | 7 +------
.../api/functions/sink/TwoPhaseCommitSinkFunction.java | 5 -----
.../operators/InternalTimersSnapshotReaderWriters.java | 5 -----
.../flink/streaming/api/operators/TimerSerializer.java | 5 -----
.../api/operators/co/IntervalJoinOperator.java | 5 -----
.../streaming/api/windowing/windows/GlobalWindow.java | 5 -----
.../streaming/api/windowing/windows/TimeWindow.java | 5 -----
.../runtime/streamrecord/StreamElementSerializer.java | 7 +------
.../flink/table/dataview/ListViewSerializer.scala | 4 +---
.../apache/flink/table/dataview/MapViewSerializer.scala | 4 +---
.../flink/table/runtime/types/CRowSerializer.scala | 11 ++---------
.../jar/CheckpointingCustomKvStateProgram.java | 5 -----
.../TypeSerializerSnapshotMigrationITCase.java | 5 -----
120 files changed, 33 insertions(+), 686 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
index de2f01d..54734e9 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
@@ -1228,11 +1228,6 @@ public class FlinkKafkaProducer011<IN>
target.writeShort(source.readShort());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof TransactionStateSerializer;
- }
-
// ------------------------------------------------------------------------
@Override
@@ -1327,11 +1322,6 @@ public class FlinkKafkaProducer011<IN>
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof ContextStateSerializer;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
index 6a1e1c3..e822142 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -1234,11 +1234,6 @@ public class FlinkKafkaProducer<IN>
target.writeShort(source.readShort());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof FlinkKafkaProducer.TransactionStateSerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
@@ -1334,11 +1329,6 @@ public class FlinkKafkaProducer<IN>
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof FlinkKafkaProducer.ContextStateSerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java b/flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java
index cb6c254..d6d8e27 100644
--- a/flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java
+++ b/flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java
@@ -146,17 +146,12 @@ public final class WritableSerializer<T extends Writable> extends TypeSerializer
if (obj instanceof WritableSerializer) {
WritableSerializer<?> other = (WritableSerializer<?>) obj;
- return other.canEqual(this) && typeClass == other.typeClass;
+ return typeClass == other.typeClass;
} else {
return false;
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof WritableSerializer;
- }
-
// --------------------------------------------------------------------------------------------
// Serializer configuration snapshotting & compatibility
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializer.java
index 8bc26a5..9e337c3 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializer.java
@@ -190,18 +190,9 @@ public abstract class CompositeSerializer<T> extends TypeSerializer<T> {
@SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
@Override
public boolean equals(Object obj) {
- if (canEqual(obj)) {
- CompositeSerializer<?> other = (CompositeSerializer<?>) obj;
- return precomputed.immutable == other.precomputed.immutable
- && Arrays.equals(fieldSerializers, other.fieldSerializers);
- }
- return false;
- }
-
- @Override
- public boolean canEqual(Object obj) {
- // as this is an abstract class, we allow equality only between instances of the same class
- return obj != null && getClass().equals(obj.getClass());
+ CompositeSerializer<?> other = (CompositeSerializer<?>) obj;
+ return precomputed.immutable == other.precomputed.immutable
+ && Arrays.equals(fieldSerializers, other.fieldSerializers);
}
@Override
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializer.java
index 2ec064a..de60ebc 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializer.java
@@ -74,14 +74,6 @@ public interface TypeDeserializer<T> {
*/
int getLength();
- /**
- * Returns true if the given object can be equaled with this object. If not, it returns false.
- *
- * @param obj Object which wants to take part in the equality relation
- * @return true if obj can be equaled with this, otherwise false
- */
- boolean canEqual(Object obj);
-
boolean equals(Object obj);
int hashCode();
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializerAdapter.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializerAdapter.java
index 72b2514..7b8e9b5 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializerAdapter.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializerAdapter.java
@@ -88,10 +88,6 @@ public final class TypeDeserializerAdapter<T> extends TypeSerializer<T> implemen
return (deserializer != null) ? deserializer.equals(obj) : serializer.equals(obj);
}
- public boolean canEqual(Object obj) {
- return (deserializer != null) ? deserializer.canEqual(obj) : serializer.canEqual(obj);
- }
-
public int hashCode() {
return (deserializer != null) ? deserializer.hashCode() : serializer.hashCode();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
index ddb0b87..80b28ae 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
@@ -177,14 +177,6 @@ public abstract class TypeSerializer<T> implements Serializable {
public abstract boolean equals(Object obj);
- /**
- * Returns true if the given object can be equaled with this object. If not, it returns false.
- *
- * @param obj Object which wants to take part in the equality relation
- * @return true if obj can be equaled with this, otherwise false
- */
- public abstract boolean canEqual(Object obj);
-
public abstract int hashCode();
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/UnloadableDummyTypeSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/UnloadableDummyTypeSerializer.java
index fe55be7..6660ad8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/UnloadableDummyTypeSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/UnloadableDummyTypeSerializer.java
@@ -119,11 +119,6 @@ public class UnloadableDummyTypeSerializer<T> extends TypeSerializer<T> {
}
@Override
- public boolean canEqual(Object obj) {
- return false;
- }
-
- @Override
public boolean equals(Object o) {
if (this == o) {
return true;
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigDecSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigDecSerializer.java
index 7e4e97f..52efb35 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigDecSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigDecSerializer.java
@@ -110,11 +110,6 @@ public final class BigDecSerializer extends TypeSerializerSingleton<BigDecimal>
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof BigDecSerializer;
- }
-
// --------------------------------------------------------------------------------------------
// Static Helpers for BigInteger Serialization
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigIntSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigIntSerializer.java
index ef4358e..87dff94 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigIntSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigIntSerializer.java
@@ -82,11 +82,6 @@ public final class BigIntSerializer extends TypeSerializerSingleton<BigInteger>
copyBigInteger(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof BigIntSerializer;
- }
-
// --------------------------------------------------------------------------------------------
// Static Helpers for BigInteger Serialization
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanSerializer.java
index 84056e5..0479a3e 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanSerializer.java
@@ -85,11 +85,6 @@ public final class BooleanSerializer extends TypeSerializerSingleton<Boolean> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof BooleanSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Boolean> snapshotConfiguration() {
return new BooleanSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanValueSerializer.java
index 3e497ce..0206c6e 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanValueSerializer.java
@@ -84,11 +84,6 @@ public final class BooleanValueSerializer extends TypeSerializerSingleton<Boolea
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof BooleanValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<BooleanValue> snapshotConfiguration() {
return new BooleanValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteSerializer.java
index 5fdd9b6..4acb482 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteSerializer.java
@@ -85,11 +85,6 @@ public final class ByteSerializer extends TypeSerializerSingleton<Byte> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof ByteSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Byte> snapshotConfiguration() {
return new ByteSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteValueSerializer.java
index 6f0cb76..b5055a6 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteValueSerializer.java
@@ -82,11 +82,6 @@ public final class ByteValueSerializer extends TypeSerializerSingleton<ByteValue
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof ByteValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<ByteValue> snapshotConfiguration() {
return new ByteValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharSerializer.java
index cee0d0d..8bf4f16 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharSerializer.java
@@ -85,11 +85,6 @@ public final class CharSerializer extends TypeSerializerSingleton<Character> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof CharSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Character> snapshotConfiguration() {
return new CharSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharValueSerializer.java
index aa580fe..af9a180 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharValueSerializer.java
@@ -82,11 +82,6 @@ public class CharValueSerializer extends TypeSerializerSingleton<CharValue> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof CharValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<CharValue> snapshotConfiguration() {
return new CharValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DateSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DateSerializer.java
index 0675e22..f95ee48 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DateSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DateSerializer.java
@@ -100,11 +100,6 @@ public final class DateSerializer extends TypeSerializerSingleton<Date> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof DateSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Date> snapshotConfiguration() {
return new DateSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleSerializer.java
index cd4854b..997442c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleSerializer.java
@@ -85,11 +85,6 @@ public final class DoubleSerializer extends TypeSerializerSingleton<Double> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof DoubleSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Double> snapshotConfiguration() {
return new DoubleSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleValueSerializer.java
index c0c6a27..d7bca29 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleValueSerializer.java
@@ -82,11 +82,6 @@ public final class DoubleValueSerializer extends TypeSerializerSingleton<DoubleV
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof DoubleValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<DoubleValue> snapshotConfiguration() {
return new DoubleValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index dcd2673..b8142a0 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -134,18 +134,13 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
if(obj instanceof EnumSerializer) {
EnumSerializer<?> other = (EnumSerializer<?>) obj;
- return other.canEqual(this) && other.enumClass == this.enumClass;
+ return other.enumClass == this.enumClass;
} else {
return false;
}
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof EnumSerializer;
- }
-
- @Override
public int hashCode() {
return enumClass.hashCode();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatSerializer.java
index 6da6f47..23439f7 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatSerializer.java
@@ -85,11 +85,6 @@ public final class FloatSerializer extends TypeSerializerSingleton<Float> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof FloatSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Float> snapshotConfiguration() {
return new FloatSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatValueSerializer.java
index 1bfc87d..a6b62b1 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatValueSerializer.java
@@ -82,11 +82,6 @@ public class FloatValueSerializer extends TypeSerializerSingleton<FloatValue> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof FloatValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<FloatValue> snapshotConfiguration() {
return new FloatValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java
index a4949fb..8e10f7a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java
@@ -183,8 +183,7 @@ public final class GenericArraySerializer<C> extends TypeSerializer<C[]> {
if (obj instanceof GenericArraySerializer) {
GenericArraySerializer<?> other = (GenericArraySerializer<?>)obj;
- return other.canEqual(this) &&
- componentClass == other.componentClass &&
+ return componentClass == other.componentClass &&
componentSerializer.equals(other.componentSerializer);
} else {
return false;
@@ -192,11 +191,6 @@ public final class GenericArraySerializer<C> extends TypeSerializer<C[]> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof GenericArraySerializer;
- }
-
- @Override
public String toString() {
return "Serializer " + componentClass.getName() + "[]";
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/InstantSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/InstantSerializer.java
index 17f1b7a..11773b9 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/InstantSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/InstantSerializer.java
@@ -32,6 +32,9 @@ import java.time.Instant;
*/
@Internal
public final class InstantSerializer extends TypeSerializerSingleton<Instant> {
+
+ private static final long serialVersionUID = -4131715684999061277L;
+
static final int SECONDS_BYTES = Long.BYTES;
static final int NANOS_BYTES = Integer.BYTES;
@@ -101,11 +104,6 @@ public final class InstantSerializer extends TypeSerializerSingleton<Instant> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof InstantSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Instant> snapshotConfiguration() {
return new InstantSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntSerializer.java
index 6d0d407..a5edb28 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntSerializer.java
@@ -85,11 +85,6 @@ public final class IntSerializer extends TypeSerializerSingleton<Integer> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof IntSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Integer> snapshotConfiguration() {
return new IntSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntValueSerializer.java
index 846df31..291c1ab 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntValueSerializer.java
@@ -82,11 +82,6 @@ public final class IntValueSerializer extends TypeSerializerSingleton<IntValue>
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof IntValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<IntValue> snapshotConfiguration() {
return new IntValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
index 5178031..fc5176c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
@@ -160,11 +160,6 @@ public final class ListSerializer<T> extends TypeSerializer<List<T>> {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public int hashCode() {
return elementSerializer.hashCode();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongSerializer.java
index 24cf7c7..bf3d33a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongSerializer.java
@@ -85,11 +85,6 @@ public final class LongSerializer extends TypeSerializerSingleton<Long> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof LongSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Long> snapshotConfiguration() {
return new LongSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongValueSerializer.java
index 9f4b5d8..7b72a3d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/LongValueSerializer.java
@@ -82,11 +82,6 @@ public final class LongValueSerializer extends TypeSerializerSingleton<LongValue
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof LongValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<LongValue> snapshotConfiguration() {
return new LongValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/MapSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/MapSerializer.java
index bedaf69..c0596eb 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/MapSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/MapSerializer.java
@@ -187,11 +187,6 @@ public final class MapSerializer<K, V> extends TypeSerializer<Map<K, V>> {
}
@Override
- public boolean canEqual(Object obj) {
- return (obj != null && obj.getClass() == getClass());
- }
-
- @Override
public int hashCode() {
return keySerializer.hashCode() * 31 + valueSerializer.hashCode();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/NullValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/NullValueSerializer.java
index 48c0b57..bbc8954 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/NullValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/NullValueSerializer.java
@@ -78,11 +78,6 @@ public final class NullValueSerializer extends TypeSerializerSingleton<NullValue
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof NullValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<NullValue> snapshotConfiguration() {
return new NullValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortSerializer.java
index fa6189d..5b76342 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortSerializer.java
@@ -85,11 +85,6 @@ public final class ShortSerializer extends TypeSerializerSingleton<Short> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof ShortSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Short> snapshotConfiguration() {
return new ShortSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortValueSerializer.java
index e667faa4..fa1318f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ShortValueSerializer.java
@@ -82,11 +82,6 @@ public final class ShortValueSerializer extends TypeSerializerSingleton<ShortVal
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof ShortValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<ShortValue> snapshotConfiguration() {
return new ShortValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlDateSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlDateSerializer.java
index ab96dd9..b101de6 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlDateSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlDateSerializer.java
@@ -101,11 +101,6 @@ public final class SqlDateSerializer extends TypeSerializerSingleton<Date> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof SqlDateSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Date> snapshotConfiguration() {
return new SqlDateSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimeSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimeSerializer.java
index b5bf7b5..bd7d8c0 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimeSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimeSerializer.java
@@ -99,11 +99,6 @@ public final class SqlTimeSerializer extends TypeSerializerSingleton<Time> {
target.writeLong(source.readLong());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof SqlTimeSerializer;
- }
-
// --------------------------------------------------------------------------------------------
// Serializer configuration snapshotting
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimestampSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimestampSerializer.java
index 5d7760c..a5e1f83 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimestampSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/SqlTimestampSerializer.java
@@ -109,11 +109,6 @@ public final class SqlTimestampSerializer extends TypeSerializerSingleton<Timest
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof SqlTimestampSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Timestamp> snapshotConfiguration() {
return new SqlTimestampSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringSerializer.java
index dc67a9c..50aa042 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringSerializer.java
@@ -86,11 +86,6 @@ public final class StringSerializer extends TypeSerializerSingleton<String> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof StringSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<String> snapshotConfiguration() {
return new StringSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringValueSerializer.java
index 498b0f0..9b1546f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/StringValueSerializer.java
@@ -107,11 +107,6 @@ public final class StringValueSerializer extends TypeSerializerSingleton<StringV
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof StringValueSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<StringValue> snapshotConfiguration() {
return new StringValueSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/TypeSerializerSingleton.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/TypeSerializerSingleton.java
index 437aa14..5eb3a00 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/TypeSerializerSingleton.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/TypeSerializerSingleton.java
@@ -43,13 +43,7 @@ public abstract class TypeSerializerSingleton<T> extends TypeSerializer<T>{
@Override
public boolean equals(Object obj) {
- if (obj instanceof TypeSerializerSingleton) {
- TypeSerializerSingleton<?> other = (TypeSerializerSingleton<?>) obj;
-
- return other.canEqual(this);
- } else {
- return false;
- }
+ return obj.getClass().equals(this.getClass());
}
/**
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VoidSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VoidSerializer.java
index d9fae48..808bb9e 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VoidSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VoidSerializer.java
@@ -86,11 +86,6 @@ public final class VoidSerializer extends TypeSerializerSingleton<Void> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof VoidSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<Void> snapshotConfiguration() {
return new VoidSerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BooleanPrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BooleanPrimitiveArraySerializer.java
index a8f76c4..6f24b06 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BooleanPrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BooleanPrimitiveArraySerializer.java
@@ -106,11 +106,6 @@ public final class BooleanPrimitiveArraySerializer extends TypeSerializerSinglet
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof BooleanPrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<boolean[]> snapshotConfiguration() {
return new BooleanPrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BytePrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BytePrimitiveArraySerializer.java
index 803bd53..b8ef6f9 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BytePrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/BytePrimitiveArraySerializer.java
@@ -99,11 +99,6 @@ public final class BytePrimitiveArraySerializer extends TypeSerializerSingleton<
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof BytePrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<byte[]> snapshotConfiguration() {
return new BytePrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/CharPrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/CharPrimitiveArraySerializer.java
index 1e2357a..efdf094 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/CharPrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/CharPrimitiveArraySerializer.java
@@ -105,11 +105,6 @@ public final class CharPrimitiveArraySerializer extends TypeSerializerSingleton<
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof CharPrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<char[]> snapshotConfiguration() {
return new CharPrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/DoublePrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/DoublePrimitiveArraySerializer.java
index 5deddc8..053187a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/DoublePrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/DoublePrimitiveArraySerializer.java
@@ -105,11 +105,6 @@ public final class DoublePrimitiveArraySerializer extends TypeSerializerSingleto
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof DoublePrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<double[]> snapshotConfiguration() {
return new DoublePrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/FloatPrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/FloatPrimitiveArraySerializer.java
index f65bcf3..9f46782 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/FloatPrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/FloatPrimitiveArraySerializer.java
@@ -105,11 +105,6 @@ public final class FloatPrimitiveArraySerializer extends TypeSerializerSingleton
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof FloatPrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<float[]> snapshotConfiguration() {
return new FloatPrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/IntPrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/IntPrimitiveArraySerializer.java
index 7923eca..658f7fd 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/IntPrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/IntPrimitiveArraySerializer.java
@@ -105,11 +105,6 @@ public class IntPrimitiveArraySerializer extends TypeSerializerSingleton<int[]>{
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof IntPrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<int[]> snapshotConfiguration() {
return new IntPrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/LongPrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/LongPrimitiveArraySerializer.java
index 66afb32..858a149 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/LongPrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/LongPrimitiveArraySerializer.java
@@ -105,11 +105,6 @@ public final class LongPrimitiveArraySerializer extends TypeSerializerSingleton<
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof LongPrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<long[]> snapshotConfiguration() {
return new LongPrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/ShortPrimitiveArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/ShortPrimitiveArraySerializer.java
index a93dd71..85b43b2 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/ShortPrimitiveArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/ShortPrimitiveArraySerializer.java
@@ -105,11 +105,6 @@ public final class ShortPrimitiveArraySerializer extends TypeSerializerSingleton
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof ShortPrimitiveArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<short[]> snapshotConfiguration() {
return new ShortPrimitiveArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/StringArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/StringArraySerializer.java
index 7c81aa6..cea3ce8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/StringArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/array/StringArraySerializer.java
@@ -109,11 +109,6 @@ public final class StringArraySerializer extends TypeSerializerSingleton<String[
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof StringArraySerializer;
- }
-
- @Override
public TypeSerializerSnapshot<String[]> snapshotConfiguration() {
return new StringArraySerializerSnapshot();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java
index 2e20415..6973767 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java
@@ -77,7 +77,7 @@ public class PojoField implements Serializable {
if (obj instanceof PojoField) {
PojoField other = (PojoField) obj;
- return other.canEqual(this) && type.equals(other.type) &&
+ return type.equals(other.type) &&
Objects.equals(field, other.field);
} else {
return false;
@@ -88,8 +88,4 @@ public class PojoField implements Serializable {
public int hashCode() {
return Objects.hash(field, type);
}
-
- public boolean canEqual(Object obj) {
- return obj instanceof PojoField;
- }
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java
index 86b925c..513df33 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java
@@ -123,19 +123,12 @@ public final class CopyableValueSerializer<T extends CopyableValue<T>> extends T
@SuppressWarnings("unchecked")
CopyableValueSerializer<T> copyableValueSerializer = (CopyableValueSerializer<T>) obj;
- return copyableValueSerializer.canEqual(this) &&
- valueClass == copyableValueSerializer.valueClass;
- }
- else {
+ return valueClass == copyableValueSerializer.valueClass;
+ } else {
return false;
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof CopyableValueSerializer;
- }
-
// --------------------------------------------------------------------------------------------
// Serializer configuration snapshotting & compatibility
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializer.java
index 0128640..4c5290c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializer.java
@@ -182,8 +182,7 @@ public class EitherSerializer<L, R> extends TypeSerializer<Either<L, R>> {
if (obj instanceof EitherSerializer) {
EitherSerializer<L, R> other = (EitherSerializer<L, R>) obj;
- return other.canEqual(this) &&
- leftSerializer.equals(other.leftSerializer) &&
+ return leftSerializer.equals(other.leftSerializer) &&
rightSerializer.equals(other.rightSerializer);
} else {
return false;
@@ -191,11 +190,6 @@ public class EitherSerializer<L, R> extends TypeSerializer<Either<L, R>> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof EitherSerializer;
- }
-
- @Override
public int hashCode() {
return 17 * leftSerializer.hashCode() + rightSerializer.hashCode();
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
index 05941fa..aa8dc32 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
@@ -236,12 +236,6 @@ public class NullableSerializer<T> extends TypeSerializer<T> {
}
@Override
- public boolean canEqual(Object obj) {
- return (obj != null && obj.getClass() == getClass() &&
- originalSerializer.canEqual(((NullableSerializer) obj).originalSerializer));
- }
-
- @Override
public int hashCode() {
return originalSerializer.hashCode();
}
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 f1dd8fc..882f7bd 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
@@ -565,8 +565,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
if (obj instanceof PojoSerializer) {
PojoSerializer<?> other = (PojoSerializer<?>) obj;
- return other.canEqual(this) &&
- clazz == other.clazz &&
+ return clazz == other.clazz &&
Arrays.equals(fieldSerializers, other.fieldSerializers) &&
Arrays.equals(registeredSerializers, other.registeredSerializers) &&
numFields == other.numFields &&
@@ -576,11 +575,6 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof PojoSerializer;
- }
-
// --------------------------------------------------------------------------------------------
// Serializer configuration snapshotting & compatibility
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java
index df33402..505ce7b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java
@@ -229,7 +229,7 @@ public final class RowSerializer extends TypeSerializer<Row> {
@Override
public boolean equals(Object obj) {
- if (canEqual(obj)) {
+ if (obj instanceof RowSerializer) {
RowSerializer other = (RowSerializer) obj;
if (this.fieldSerializers.length == other.fieldSerializers.length) {
for (int i = 0; i < this.fieldSerializers.length; i++) {
@@ -245,11 +245,6 @@ public final class RowSerializer extends TypeSerializer<Row> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof RowSerializer;
- }
-
- @Override
public int hashCode() {
return Arrays.hashCode(fieldSerializers);
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java
index 42665f3..ddb3b9d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java
@@ -111,17 +111,6 @@ public class Tuple0Serializer extends TupleSerializer<Tuple0> implements SelfRes
@Override
public boolean equals(Object obj) {
- if (obj instanceof Tuple0Serializer) {
- Tuple0Serializer other = (Tuple0Serializer) obj;
-
- return other.canEqual(this);
- } else {
- return false;
- }
- }
-
- @Override
- public boolean canEqual(Object obj) {
return obj instanceof Tuple0Serializer;
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java
index ad8591a..6a294ac 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java
@@ -103,8 +103,7 @@ public abstract class TupleSerializerBase<T> extends TypeSerializer<T> {
if (obj instanceof TupleSerializerBase) {
TupleSerializerBase<?> other = (TupleSerializerBase<?>) obj;
- return other.canEqual(this) &&
- tupleClass == other.tupleClass &&
+ return tupleClass == other.tupleClass &&
Arrays.equals(fieldSerializers, other.fieldSerializers) &&
arity == other.arity;
} else {
@@ -112,11 +111,6 @@ public abstract class TupleSerializerBase<T> extends TypeSerializer<T> {
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof TupleSerializerBase;
- }
-
@VisibleForTesting
public TypeSerializer<Object>[] getFieldSerializers() {
return fieldSerializers;
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java
index 9e02c35..5758f3f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java
@@ -160,17 +160,12 @@ public final class ValueSerializer<T extends Value> extends TypeSerializer<T> {
if (obj instanceof ValueSerializer) {
ValueSerializer<?> other = (ValueSerializer<?>) obj;
- return other.canEqual(this) && type == other.type;
+ return type == other.type;
} else {
return false;
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof ValueSerializer;
- }
-
// --------------------------------------------------------------------------------------------
// Serializer configuration snapshotting & compatibility
// --------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
index af36655..98b07e1 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
@@ -374,8 +374,7 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
if (obj instanceof KryoSerializer) {
KryoSerializer<?> other = (KryoSerializer<?>) obj;
- return other.canEqual(this) &&
- type == other.type &&
+ return type == other.type &&
Objects.equals(kryoRegistrations, other.kryoRegistrations) &&
Objects.equals(defaultSerializerClasses, other.defaultSerializerClasses) &&
Objects.equals(defaultSerializers, other.defaultSerializers);
@@ -384,11 +383,6 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
}
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof KryoSerializer;
- }
-
// --------------------------------------------------------------------------------------------
/**
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshotTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshotTest.java
index 0f77e3d..12601c9 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshotTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshotTest.java
@@ -287,7 +287,7 @@ public class CompositeTypeSerializerSnapshotTest {
@Override
public boolean equals(Object obj) {
- if (canEqual(obj)) {
+ if (obj instanceof TestCompositeTypeSerializer) {
return Arrays.equals(nestedSerializers, ((TestCompositeTypeSerializer) obj).getNestedSerializers());
}
return false;
@@ -297,11 +297,6 @@ public class CompositeTypeSerializerSnapshotTest {
public int hashCode() {
return Arrays.hashCode(nestedSerializers);
}
-
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof TestCompositeTypeSerializer;
- }
}
/**
@@ -438,21 +433,13 @@ public class CompositeTypeSerializerSnapshotTest {
@Override
public boolean equals(Object obj) {
- if (canEqual(obj)) {
- return targetCompatibility == ((NestedSerializer) obj).targetCompatibility;
- }
- return false;
+ return targetCompatibility == ((NestedSerializer) obj).targetCompatibility;
}
@Override
public int hashCode() {
return targetCompatibility.hashCode();
}
-
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof NestedSerializer;
- }
}
/**
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeInformationTestBase.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeInformationTestBase.java
index 164fc95..c23e453 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeInformationTestBase.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeInformationTestBase.java
@@ -54,7 +54,6 @@ public abstract class TypeInformationTestBase<T extends TypeInformation<?>> exte
// compare among test data
for (T otherTypeInfo : testData) {
- assertTrue("canEqual() returns inconsistent results.", typeInfo.canEqual(otherTypeInfo));
// test equality
if (typeInfo == otherTypeInfo) {
assertTrue("hashCode() returns inconsistent results.", typeInfo.hashCode() == otherTypeInfo.hashCode());
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationUtilTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationUtilTest.java
index ed80d28..1a81c61 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationUtilTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationUtilTest.java
@@ -420,11 +420,6 @@ public class TypeSerializerSerializationUtilTest implements Serializable {
}
@Override
- public boolean canEqual(Object obj) {
- return IntSerializer.INSTANCE.canEqual(obj);
- }
-
- @Override
public boolean equals(Object obj) {
return IntSerializer.INSTANCE.equals(obj);
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java
index 4cfd4f0..ece8bc8 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java
@@ -168,11 +168,6 @@ public class TypeSerializerSnapshotTest {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public int hashCode() {
return 0;
}
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/artificialstate/StatefulComplexPayloadSerializer.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/artificialstate/StatefulComplexPayloadSerializer.java
index 294241b..8b04486 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/artificialstate/StatefulComplexPayloadSerializer.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/artificialstate/StatefulComplexPayloadSerializer.java
@@ -131,11 +131,6 @@ public class StatefulComplexPayloadSerializer extends TypeSerializer<ComplexPayl
}
@Override
- public boolean canEqual(Object obj) {
- return getClass().equals(obj.getClass());
- }
-
- @Override
public int hashCode() {
return 42;
}
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java
index da51117..643fb6b 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java
@@ -309,11 +309,6 @@ public class AvroSerializer<T> extends TypeSerializer<T> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj.getClass() == this.getClass();
- }
-
- @Override
public String toString() {
return getClass().getName() + " (" + getType().getName() + ')';
}
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java b/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
index 7b14417..95b4e3b 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
@@ -378,18 +378,13 @@ public class CollectionInputFormatTest {
if (obj instanceof TestSerializer) {
TestSerializer other = (TestSerializer) obj;
- return other.canEqual(this) && failOnRead == other.failOnRead && failOnWrite == other.failOnWrite;
+ return failOnRead == other.failOnRead && failOnWrite == other.failOnWrite;
} else {
return false;
}
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof TestSerializer;
- }
-
- @Override
public int hashCode() {
return Objects.hash(failOnRead, failOnWrite);
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java
index 03881f3..6141bf2 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java
@@ -262,11 +262,6 @@ public class DeweyNumber implements Serializable {
return obj == this || obj.getClass().equals(getClass());
}
- @Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
index 869a641..3ddec5c 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
@@ -995,11 +995,6 @@ public class NFA<T> {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public int hashCode() {
return 37 * sharedBufferSerializer.hashCode() + eventSerializer.hashCode();
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAStateSerializer.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAStateSerializer.java
index 0c19036..c1815e6 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAStateSerializer.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAStateSerializer.java
@@ -141,11 +141,6 @@ public class NFAStateSerializer extends TypeSerializerSingleton<NFAState> {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public TypeSerializerSnapshot<NFAState> snapshotConfiguration() {
return new NFAStateSerializerSnapshot(this);
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java
index 0617310..3ebfac7 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java
@@ -346,11 +346,6 @@ public class SharedBuffer<V> {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public int hashCode() {
return 37 * keySerializer.hashCode() + valueSerializer.hashCode();
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/EventId.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/EventId.java
index 3ac39b5..4edf34a 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/EventId.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/EventId.java
@@ -143,11 +143,6 @@ public class EventId implements Comparable<EventId> {
target.writeLong(source.readLong());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj.getClass().equals(EventIdSerializer.class);
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java
index ec992c8..4dc6a27 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/Lockable.java
@@ -182,11 +182,6 @@ public final class Lockable<T> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj.getClass().equals(LockableTypeSerializer.class);
- }
-
- @Override
public TypeSerializerSnapshot<Lockable<E>> snapshotConfiguration() {
return new LockableTypeSerializerSnapshot<>(this);
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/NodeId.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/NodeId.java
index 2693d4b..6726cd3 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/NodeId.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/NodeId.java
@@ -160,11 +160,6 @@ public class NodeId {
StringValue.copyString(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj.getClass().equals(NodeIdSerializer.class);
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferEdge.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferEdge.java
index d6a95f6..43afdb2 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferEdge.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferEdge.java
@@ -139,11 +139,6 @@ public class SharedBufferEdge {
deweyNumberSerializer.copy(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj.getClass().equals(SharedBufferEdgeSerializer.class);
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferNode.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferNode.java
index 473a342..57e5738 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferNode.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferNode.java
@@ -124,11 +124,6 @@ public class SharedBufferNode {
edgesSerializer.copy(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj.getClass().equals(SharedBufferNodeSerializer.class);
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java
index 6fc7bd0..9b5a8c3 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java
@@ -216,11 +216,6 @@ extends LongValue {
target.writeLong(source.readLong());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof LongValueWithProperHashCodeSerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ByteValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ByteValueArraySerializer.java
index c690543..5cc0ce1 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ByteValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ByteValueArraySerializer.java
@@ -81,11 +81,6 @@ public final class ByteValueArraySerializer extends TypeSerializerSingleton<Byte
ByteValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof ByteValueArraySerializer;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/CharValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/CharValueArraySerializer.java
index 0844986..bb3158e 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/CharValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/CharValueArraySerializer.java
@@ -81,11 +81,6 @@ public final class CharValueArraySerializer extends TypeSerializerSingleton<Char
CharValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof CharValueArraySerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/DoubleValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/DoubleValueArraySerializer.java
index be683b5..bdf068c 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/DoubleValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/DoubleValueArraySerializer.java
@@ -81,11 +81,6 @@ public final class DoubleValueArraySerializer extends TypeSerializerSingleton<Do
DoubleValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof DoubleValueArraySerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/FloatValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/FloatValueArraySerializer.java
index 0fb0f61..0f33bfc 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/FloatValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/FloatValueArraySerializer.java
@@ -81,11 +81,6 @@ public final class FloatValueArraySerializer extends TypeSerializerSingleton<Flo
FloatValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof FloatValueArraySerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/IntValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/IntValueArraySerializer.java
index c00c136..5c997b7 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/IntValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/IntValueArraySerializer.java
@@ -80,11 +80,6 @@ public final class IntValueArraySerializer extends TypeSerializerSingleton<IntVa
IntValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof IntValueArraySerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/LongValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/LongValueArraySerializer.java
index 55a79b3..b9a183a 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/LongValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/LongValueArraySerializer.java
@@ -80,11 +80,6 @@ public final class LongValueArraySerializer extends TypeSerializerSingleton<Long
LongValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof LongValueArraySerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/NullValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/NullValueArraySerializer.java
index 732bb0f..a2fad93 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/NullValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/NullValueArraySerializer.java
@@ -80,11 +80,6 @@ public final class NullValueArraySerializer extends TypeSerializerSingleton<Null
target.write(source, getLength());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof NullValueArraySerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ShortValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ShortValueArraySerializer.java
index 2000799..9a580d6 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ShortValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/ShortValueArraySerializer.java
@@ -81,11 +81,6 @@ public final class ShortValueArraySerializer extends TypeSerializerSingleton<Sho
ShortValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof ShortValueArraySerializer;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/StringValueArraySerializer.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/StringValueArraySerializer.java
index d3dbe70..ce31494 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/StringValueArraySerializer.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/types/valuearray/StringValueArraySerializer.java
@@ -80,11 +80,6 @@ public final class StringValueArraySerializer extends TypeSerializerSingleton<St
StringValueArray.copyInternal(source, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof StringValueArraySerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/VoidNamespaceSerializer.java b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/VoidNamespaceSerializer.java
index e6a9f10..7cf8f2d 100644
--- a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/VoidNamespaceSerializer.java
+++ b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/VoidNamespaceSerializer.java
@@ -91,11 +91,6 @@ public final class VoidNamespaceSerializer extends TypeSerializerSingleton<VoidN
target.write(source.readByte());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof VoidNamespaceSerializer;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java
index d442d0d..b1668fc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java
@@ -132,11 +132,6 @@ final public class ArrayListSerializer<T> extends TypeSerializer<ArrayList<T>> {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public int hashCode() {
return elementSerializer.hashCode();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/JavaSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/JavaSerializer.java
index 60b5fd6..9e00985 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/JavaSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/JavaSerializer.java
@@ -95,11 +95,6 @@ final class JavaSerializer<T extends Serializable> extends TypeSerializerSinglet
serialize(tmp, target);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof JavaSerializer;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/VoidNamespaceSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/VoidNamespaceSerializer.java
index c861acf..aad33e2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/VoidNamespaceSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/VoidNamespaceSerializer.java
@@ -87,11 +87,6 @@ public final class VoidNamespaceSerializer extends TypeSerializerSingleton<VoidN
target.write(source.readByte());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof VoidNamespaceSerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntListSerializer.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntListSerializer.java
index 67d8a26..e810337 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntListSerializer.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntListSerializer.java
@@ -111,17 +111,6 @@ public class IntListSerializer extends TypeSerializer<IntList> {
@Override
public boolean equals(Object obj) {
- if (obj instanceof IntListSerializer) {
- IntListSerializer other = (IntListSerializer) obj;
-
- return other.canEqual(this);
- } else {
- return false;
- }
- }
-
- @Override
- public boolean canEqual(Object obj) {
return obj instanceof IntListSerializer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntPairSerializer.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntPairSerializer.java
index fe456c82..2e40a0e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntPairSerializer.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/IntPairSerializer.java
@@ -92,17 +92,6 @@ public class IntPairSerializer extends TypeSerializer<IntPair> {
@Override
public boolean equals(Object obj) {
- if (obj instanceof IntPairSerializer) {
- IntPairSerializer other = (IntPairSerializer) obj;
-
- return other.canEqual(this);
- } else {
- return false;
- }
- }
-
- @Override
- public boolean canEqual(Object obj) {
return obj instanceof IntPairSerializer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/StringPairSerializer.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/StringPairSerializer.java
index cad78e7..af64a59 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/StringPairSerializer.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/types/StringPairSerializer.java
@@ -88,17 +88,6 @@ public class StringPairSerializer extends TypeSerializer<StringPair> {
@Override
public boolean equals(Object obj) {
- if (obj instanceof StringPairSerializer) {
- StringPairSerializer other = (StringPairSerializer) obj;
-
- return other.canEqual(this);
- } else {
- return false;
- }
- }
-
- @Override
- public boolean canEqual(Object obj) {
return obj instanceof StringPairSerializer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java
index 6a9e543..bd94639 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java
@@ -396,11 +396,6 @@ public class KvStateRegistryTest extends TestLogger {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public int hashCode() {
return 0;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/InternalPriorityQueueTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/InternalPriorityQueueTestBase.java
index 935ebb6..2f4a33e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/InternalPriorityQueueTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/InternalPriorityQueueTestBase.java
@@ -487,11 +487,6 @@ public abstract class InternalPriorityQueueTestBase extends TestLogger {
}
@Override
- public boolean canEqual(Object obj) {
- return false;
- }
-
- @Override
public int hashCode() {
return 4711;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
index 3918303..78e6f4e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
@@ -330,15 +330,6 @@ public class OperatorStateBackendTest {
@Override
public boolean equals(Object obj) {
- if (obj instanceof VerifyingIntSerializer) {
- return ((VerifyingIntSerializer)obj).canEqual(this);
- } else {
- return false;
- }
- }
-
- @Override
- public boolean canEqual(Object obj) {
return obj instanceof VerifyingIntSerializer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
index b636ed0..3177dd4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
@@ -911,11 +911,6 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof CustomVoidNamespaceSerializer;
- }
-
- @Override
public boolean equals(Object obj) {
return obj instanceof CustomVoidNamespaceSerializer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSnapshotTransformerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSnapshotTransformerTest.java
index 42bda6e..044abaa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSnapshotTransformerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSnapshotTransformerTest.java
@@ -271,13 +271,6 @@ class StateSnapshotTransformerTest {
}
@Override
- public boolean canEqual(Object obj) {
- singleThreadAccessChecker.checkSingleThreadAccess();
- return (obj != null && obj.getClass() == getClass() &&
- StringSerializer.INSTANCE.canEqual(obj));
- }
-
- @Override
public int hashCode() {
singleThreadAccessChecker.checkSingleThreadAccess();
return StringSerializer.INSTANCE.hashCode();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/heap/TestDuplicateSerializer.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/heap/TestDuplicateSerializer.java
index 83e249b..a7e4ac8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/heap/TestDuplicateSerializer.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/heap/TestDuplicateSerializer.java
@@ -104,11 +104,6 @@ public class TestDuplicateSerializer extends TypeSerializer<Integer> {
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof TestDuplicateSerializer;
- }
-
- @Override
public int hashCode() {
return getClass().hashCode();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/recordutils/RecordSerializer.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/recordutils/RecordSerializer.java
index 87d91e2..a892bf4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/recordutils/RecordSerializer.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/recordutils/RecordSerializer.java
@@ -126,16 +126,6 @@ public final class RecordSerializer extends TypeSerializer<Record> {
@Override
public boolean equals(Object obj) {
- if (obj instanceof RecordSerializer) {
- RecordSerializer other = (RecordSerializer) obj;
- return other.canEqual(this);
- } else {
- return false;
- }
- }
-
- @Override
- public boolean canEqual(Object obj) {
return obj instanceof RecordSerializer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java
index 52d5521..62e4ccb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java
@@ -243,11 +243,6 @@ public class TestType implements HeapPriorityQueueElement, PriorityComparable<Te
}
@Override
- public boolean canEqual(Object obj) {
- return getClass().equals(obj.getClass());
- }
-
- @Override
public int hashCode() {
return getClass().hashCode();
}
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
index 0427bb3..6e7d65e 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
@@ -101,17 +101,12 @@ class EitherSerializer[A, B](
override def equals(obj: Any): Boolean = {
obj match {
case eitherSerializer: EitherSerializer[_, _] =>
- eitherSerializer.canEqual(this) &&
leftSerializer.equals(eitherSerializer.leftSerializer) &&
rightSerializer.equals(eitherSerializer.rightSerializer)
case _ => false
}
}
- override def canEqual(obj: Any): Boolean = {
- obj.isInstanceOf[EitherSerializer[_, _]]
- }
-
override def hashCode(): Int = {
31 * leftSerializer.hashCode() + rightSerializer.hashCode()
}
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
index b7e21ce..3286095 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
@@ -62,7 +62,7 @@ class EnumValueSerializer[E <: Enumeration](val enum: E) extends TypeSerializer[
override def equals(obj: Any): Boolean = {
obj match {
case enumValueSerializer: EnumValueSerializer[_] =>
- enumValueSerializer.canEqual(this) && enum == enumValueSerializer.enum
+ enum == enumValueSerializer.enum
case _ => false
}
}
@@ -71,10 +71,6 @@ class EnumValueSerializer[E <: Enumeration](val enum: E) extends TypeSerializer[
enum.hashCode()
}
- override def canEqual(obj: scala.Any): Boolean = {
- obj.isInstanceOf[EnumValueSerializer[_]]
- }
-
// --------------------------------------------------------------------------------------------
// Serializer configuration snapshotting & compatibility
// --------------------------------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/NothingSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/NothingSerializer.scala
index eff57b6..aefca07 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/NothingSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/NothingSerializer.scala
@@ -65,15 +65,11 @@ class NothingSerializer extends TypeSerializer[Any] {
override def equals(obj: Any): Boolean = {
obj match {
- case nothingSerializer: NothingSerializer => nothingSerializer.canEqual(this)
+ case nothingSerializer: NothingSerializer => true
case _ => false
}
}
- override def canEqual(obj: scala.Any): Boolean = {
- obj.isInstanceOf[NothingSerializer]
- }
-
override def hashCode(): Int = {
classOf[NothingSerializer].hashCode()
}
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionSerializer.scala
index d41d2e8..d0b01fc 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionSerializer.scala
@@ -84,15 +84,11 @@ class OptionSerializer[A](val elemSerializer: TypeSerializer[A])
override def equals(obj: Any): Boolean = {
obj match {
case optionSerializer: OptionSerializer[_] =>
- optionSerializer.canEqual(this) && elemSerializer.equals(optionSerializer.elemSerializer)
+ elemSerializer.equals(optionSerializer.elemSerializer)
case _ => false
}
}
- override def canEqual(obj: scala.Any): Boolean = {
- obj.isInstanceOf[OptionSerializer[_]]
- }
-
override def hashCode(): Int = {
elemSerializer.hashCode()
}
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TraversableSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TraversableSerializer.scala
index 0f4529a..09aec85 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TraversableSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TraversableSerializer.scala
@@ -160,7 +160,7 @@ class TraversableSerializer[T <: TraversableOnce[E], E](
override def equals(obj: Any): Boolean = {
obj match {
case other: TraversableSerializer[_, _] =>
- other.canEqual(this) && elementSerializer.equals(other.elementSerializer)
+ elementSerializer.equals(other.elementSerializer)
case _ => false
}
}
@@ -169,10 +169,6 @@ class TraversableSerializer[T <: TraversableOnce[E], E](
elementSerializer.hashCode()
}
- override def canEqual(obj: Any): Boolean = {
- obj.isInstanceOf[TraversableSerializer[_, _]]
- }
-
override def snapshotConfiguration(): TraversableSerializerSnapshot[T, E] = {
new TraversableSerializerSnapshot[T, E](this)
}
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala
index 786c128..8c92f77 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala
@@ -87,15 +87,11 @@ class TrySerializer[A](
override def equals(obj: Any): Boolean = {
obj match {
case other: TrySerializer[_] =>
- other.canEqual(this) && elemSerializer.equals(other.elemSerializer)
+ elemSerializer.equals(other.elemSerializer)
case _ => false
}
}
- override def canEqual(obj: Any): Boolean = {
- obj.isInstanceOf[TrySerializer[_]]
- }
-
override def hashCode(): Int = {
31 * elemSerializer.hashCode() + executionConfig.hashCode()
}
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/UnitSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/UnitSerializer.scala
index 8a49b18..33f4bf5 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/UnitSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/UnitSerializer.scala
@@ -59,10 +59,6 @@ class UnitSerializer extends TypeSerializerSingleton[Unit] {
override def hashCode(): Int = classOf[UnitSerializer].hashCode
- override def canEqual(obj: scala.Any): Boolean = {
- obj.isInstanceOf[UnitSerializer]
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java
index 3ef1861..9178a82 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java
@@ -607,18 +607,13 @@ public class CoGroupedStreams<T1, T2> {
if (obj instanceof UnionSerializer) {
UnionSerializer<T1, T2> other = (UnionSerializer<T1, T2>) obj;
- return other.canEqual(this) && oneSerializer.equals(other.oneSerializer) && twoSerializer.equals(other.twoSerializer);
+ return oneSerializer.equals(other.oneSerializer) && twoSerializer.equals(other.twoSerializer);
} else {
return false;
}
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof UnionSerializer;
- }
-
- @Override
public TypeSerializerSnapshot<TaggedUnion<T1, T2>> snapshotConfiguration() {
return new UnionSerializerSnapshot<>(this);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
index fc8e4c1..ce3bd0f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
@@ -757,11 +757,6 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof StateSerializer;
- }
-
- @Override
public boolean equals(Object o) {
if (this == o) {
return true;
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimersSnapshotReaderWriters.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimersSnapshotReaderWriters.java
index b95b048..180e254 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimersSnapshotReaderWriters.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimersSnapshotReaderWriters.java
@@ -391,11 +391,6 @@ public class InternalTimersSnapshotReaderWriters {
}
@Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
- @Override
public int hashCode() {
return getClass().hashCode();
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimerSerializer.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimerSerializer.java
index fb8e0a2..a9774e4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimerSerializer.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimerSerializer.java
@@ -200,11 +200,6 @@ public class TimerSerializer<K, N> extends TypeSerializer<TimerHeapInternalTimer
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof TimerSerializer;
- }
-
- @Override
public TimerSerializerSnapshot<K, N> snapshotConfiguration() {
return new TimerSerializerSnapshot<>(this);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/IntervalJoinOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/IntervalJoinOperator.java
index 8f79c17..84629e4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/IntervalJoinOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/IntervalJoinOperator.java
@@ -461,11 +461,6 @@ public class IntervalJoinOperator<K, T1, T2, OUT>
}
@Override
- public boolean canEqual(Object obj) {
- return obj.getClass().equals(BufferEntrySerializer.class);
- }
-
- @Override
public TypeSerializerSnapshot<BufferEntry<T>> snapshotConfiguration() {
return new BufferEntrySerializerSnapshot<>(this);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/GlobalWindow.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/GlobalWindow.java
index 0e09b2e..459ad50 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/GlobalWindow.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/GlobalWindow.java
@@ -118,11 +118,6 @@ public class GlobalWindow extends Window {
target.writeByte(0);
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof Serializer;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/TimeWindow.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/TimeWindow.java
index a778ebf..25698cd 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/TimeWindow.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/windows/TimeWindow.java
@@ -186,11 +186,6 @@ public class TimeWindow extends Window {
target.writeLong(source.readLong());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof Serializer;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java
index e8af0f9..10c2f68 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java
@@ -254,18 +254,13 @@ public final class StreamElementSerializer<T> extends TypeSerializer<StreamEleme
if (obj instanceof StreamElementSerializer) {
StreamElementSerializer<?> other = (StreamElementSerializer<?>) obj;
- return other.canEqual(this) && typeSerializer.equals(other.typeSerializer);
+ return typeSerializer.equals(other.typeSerializer);
} else {
return false;
}
}
@Override
- public boolean canEqual(Object obj) {
- return obj instanceof StreamElementSerializer;
- }
-
- @Override
public int hashCode() {
return typeSerializer.hashCode();
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/ListViewSerializer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/ListViewSerializer.scala
index 2d48c3d..30257e5 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/ListViewSerializer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/ListViewSerializer.scala
@@ -69,11 +69,9 @@ class ListViewSerializer[T](val listSerializer: TypeSerializer[java.util.List[T]
override def copy(source: DataInputView, target: DataOutputView): Unit =
listSerializer.copy(source, target)
- override def canEqual(obj: scala.Any): Boolean = obj != null && obj.getClass == getClass
-
override def hashCode(): Int = listSerializer.hashCode()
- override def equals(obj: Any): Boolean = canEqual(this) &&
+ override def equals(obj: Any): Boolean =
listSerializer.equals(obj.asInstanceOf[ListViewSerializer[_]].listSerializer)
override def snapshotConfiguration(): ListViewSerializerSnapshot[T] =
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/MapViewSerializer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/MapViewSerializer.scala
index e0067c5..9b7bc28 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/MapViewSerializer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/dataview/MapViewSerializer.scala
@@ -70,11 +70,9 @@ class MapViewSerializer[K, V](val mapSerializer: TypeSerializer[java.util.Map[K,
override def copy(source: DataInputView, target: DataOutputView): Unit =
mapSerializer.copy(source, target)
- override def canEqual(obj: Any): Boolean = obj != null && obj.getClass == getClass
-
override def hashCode(): Int = mapSerializer.hashCode()
- override def equals(obj: Any): Boolean = canEqual(this) &&
+ override def equals(obj: Any): Boolean =
mapSerializer.equals(obj.asInstanceOf[MapViewSerializer[_, _]].mapSerializer)
override def snapshotConfiguration(): MapViewSerializerSnapshot[K, V] =
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/types/CRowSerializer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/types/CRowSerializer.scala
index 02e9160..9a2bfb8 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/types/CRowSerializer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/types/CRowSerializer.scala
@@ -63,16 +63,9 @@ class CRowSerializer(val rowSerializer: TypeSerializer[Row]) extends TypeSeriali
target.writeBoolean(source.readBoolean())
}
- override def canEqual(obj: Any): Boolean = obj.isInstanceOf[CRowSerializer]
-
override def equals(obj: Any): Boolean = {
-
- if (canEqual(obj)) {
- val other = obj.asInstanceOf[CRowSerializer]
- rowSerializer.equals(other.rowSerializer)
- } else {
- false
- }
+ val other = obj.asInstanceOf[CRowSerializer]
+ rowSerializer.equals(other.rowSerializer)
}
override def hashCode: Int = rowSerializer.hashCode() * 13
diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/CheckpointingCustomKvStateProgram.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/CheckpointingCustomKvStateProgram.java
index 67ada5d..3a542bf 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/CheckpointingCustomKvStateProgram.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/CheckpointingCustomKvStateProgram.java
@@ -228,11 +228,6 @@ public class CheckpointingCustomKvStateProgram {
target.writeInt(source.readInt());
}
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof CustomIntSerializer;
- }
-
// -----------------------------------------------------------------------------------
@Override
diff --git a/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java
index c7070b0..3aa3963 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java
@@ -252,11 +252,6 @@ public class TypeSerializerSnapshotMigrationITCase extends SavepointMigrationTes
public boolean equals(Object obj) {
return obj instanceof TestSerializer;
}
-
- @Override
- public boolean canEqual(Object obj) {
- return true;
- }
}
/**