[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.

flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java

@@ -1228,11 +1228,6 @@ public void copy(
  target.writeShort(source.readShort());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof TransactionStateSerializer;
- }
-
  // ------------------------------------------------------------------------
 
  @Override
@@ -1327,11 +1322,6 @@ public void copy(
  }
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof ContextStateSerializer;
- }
-
  // ------------------------------------------------------------------------
 
  @Override

flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java

@@ -1234,11 +1234,6 @@ public void copy(
  target.writeShort(source.readShort());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof FlinkKafkaProducer.TransactionStateSerializer;
- }
-
  // -----------------------------------------------------------------------------------
 
  @Override
@@ -1334,11 +1329,6 @@ public void copy(
  }
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof FlinkKafkaProducer.ContextStateSerializer;
- }
-
  // -----------------------------------------------------------------------------------
 
  @Override

flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java

@@ -146,17 +146,12 @@ public boolean equals(Object obj) {
  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
  // --------------------------------------------------------------------------------------------

flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializer.java

@@ -190,18 +190,9 @@ public int hashCode() {
  @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

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();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeDeserializerAdapter.java

@@ -88,10 +88,6 @@ public boolean equals(Object obj) {
  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();
  }

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();
 
  // --------------------------------------------------------------------------------------------

flink-core/src/main/java/org/apache/flink/api/common/typeutils/UnloadableDummyTypeSerializer.java

@@ -118,11 +118,6 @@ public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot<?
  throw new UnsupportedOperationException("This object is a dummy TypeSerializer.");
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return false;
- }
-
  @Override
  public boolean equals(Object o) {
  if (this == o) {

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigDecSerializer.java

@@ -110,11 +110,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  }
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof BigDecSerializer;
- }
-
  // --------------------------------------------------------------------------------------------
  // Static Helpers for BigInteger Serialization
  // --------------------------------------------------------------------------------------------

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BigIntSerializer.java

@@ -82,11 +82,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  copyBigInteger(source, target);
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof BigIntSerializer;
- }
-
  // --------------------------------------------------------------------------------------------
  // Static Helpers for BigInteger Serialization
  // --------------------------------------------------------------------------------------------

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanSerializer.java

@@ -84,11 +84,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeBoolean(source.readBoolean());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof BooleanSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Boolean> snapshotConfiguration() {
  return new BooleanSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/BooleanValueSerializer.java

@@ -83,11 +83,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeBoolean(source.readBoolean());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof BooleanValueSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<BooleanValue> snapshotConfiguration() {
  return new BooleanValueSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteSerializer.java

@@ -84,11 +84,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeByte(source.readByte());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof ByteSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Byte> snapshotConfiguration() {
  return new ByteSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ByteValueSerializer.java

@@ -81,11 +81,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeByte(source.readByte());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof ByteValueSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<ByteValue> snapshotConfiguration() {
  return new ByteValueSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharSerializer.java

@@ -84,11 +84,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeChar(source.readChar());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof CharSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Character> snapshotConfiguration() {
  return new CharSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CharValueSerializer.java

@@ -81,11 +81,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeChar(source.readChar());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof CharValueSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<CharValue> snapshotConfiguration() {
  return new CharValueSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DateSerializer.java

@@ -99,11 +99,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeLong(source.readLong());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof DateSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Date> snapshotConfiguration() {
  return new DateSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleSerializer.java

@@ -84,11 +84,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeDouble(source.readDouble());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof DoubleSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Double> snapshotConfiguration() {
  return new DoubleSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/DoubleValueSerializer.java

@@ -81,11 +81,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeDouble(source.readDouble());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof DoubleValueSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<DoubleValue> snapshotConfiguration() {
  return new DoubleValueSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java

@@ -134,17 +134,12 @@ public boolean equals(Object obj) {
  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();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatSerializer.java

@@ -84,11 +84,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeFloat(source.readFloat());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof FloatSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Float> snapshotConfiguration() {
  return new FloatSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/FloatValueSerializer.java

@@ -81,11 +81,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeFloat(source.readFloat());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof FloatValueSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<FloatValue> snapshotConfiguration() {
  return new FloatValueSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java

@@ -183,19 +183,13 @@ public boolean equals(Object obj) {
  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;
  }
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof GenericArraySerializer;
- }
-
  @Override
  public String toString() {
  return "Serializer " + componentClass.getName() + "[]";

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/InstantSerializer.java

@@ -32,6 +32,9 @@
  */
 @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;
 
@@ -100,11 +103,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeInt(source.readInt());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof InstantSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Instant> snapshotConfiguration() {
  return new InstantSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntSerializer.java

@@ -84,11 +84,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeInt(source.readInt());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof IntSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<Integer> snapshotConfiguration() {
  return new IntSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/IntValueSerializer.java

@@ -81,11 +81,6 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
  target.writeInt(source.readInt());
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return obj instanceof IntValueSerializer;
- }
-
  @Override
  public TypeSerializerSnapshot<IntValue> snapshotConfiguration() {
  return new IntValueSerializerSnapshot();

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java

@@ -159,11 +159,6 @@ public boolean equals(Object obj) {
  elementSerializer.equals(((ListSerializer<?>) obj).elementSerializer));
  }
 
- @Override
- public boolean canEqual(Object obj) {
- return true;
- }
-
  @Override
  public int hashCode() {
  return elementSerializer.hashCode();