[ErrorProne] Fix EqualsGetClass warnings and enable Error Prone check

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -1541,7 +1541,6 @@ class BeamModulePlugin implements Plugin<Project> {
             "DoNotMockAutoValue",
             "EmptyBlockTag",
             "EmptyCatch",
-            "EqualsGetClass",
             "EqualsUnsafeCast",
             "EscapedEntity",
             "ExtendsAutoValue",

examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergRestCatalogStreamingWriteExample.java

@@ -211,7 +211,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof Accum)) {
         return false;
       }
       Accum accum = (Accum) o;

examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java

@@ -140,7 +140,7 @@ public boolean equals(Object object) {
       if (object == null) {
         return false;
       }
-      if (object.getClass() != getClass()) {
+      if (!(object instanceof StationSpeed)) {
         return false;
       }
       StationSpeed otherStationSpeed = (StationSpeed) object;

examples/java/src/main/java/org/apache/beam/examples/complete/game/UserScore.java

@@ -121,7 +121,7 @@ public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || o.getClass() != this.getClass()) {
+      if (!(o instanceof GameActionInfo)) {
         return false;
       }
 

examples/java/twitter/src/main/java/org/apache/beam/examples/twitterstreamgenerator/ReadFromTwitterDoFn.java

@@ -64,7 +64,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof OffsetHolder)) {
         return false;
       }
       OffsetHolder that = (OffsetHolder) o;

examples/java/twitter/src/main/java/org/apache/beam/examples/twitterstreamgenerator/TwitterConfig.java

@@ -53,7 +53,7 @@ public boolean equals(@Nullable Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (!(o instanceof TwitterConfig)) {
       return false;
     }
     TwitterConfig that = (TwitterConfig) o;

runners/core-java/src/main/java/org/apache/beam/runners/core/construction/SerializablePipelineOptions.java

@@ -89,7 +89,7 @@ public boolean equals(@Nullable Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (!(o instanceof SerializablePipelineOptions)) {
       return false;
     }
     SerializablePipelineOptions that = (SerializablePipelineOptions) o;

runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/BoundedTrieData.java

@@ -286,7 +286,7 @@ public final boolean equals(@Nullable Object other) {
     if (this == other) {
       return true;
     }
-    if (other == null || this.getClass() != other.getClass()) {
+    if (!(other instanceof BoundedTrieData)) {
       return false;
     }
     BoundedTrieData that = (BoundedTrieData) other;
@@ -590,7 +590,7 @@ public boolean equals(@Nullable Object other) {
       if (this == other) {
         return true;
       }
-      if (other == null || getClass() != other.getClass()) {
+      if (!(other instanceof BoundedTrieNode)) {
         return false;
       }
       BoundedTrieNode that = (BoundedTrieNode) other;

runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java

@@ -309,7 +309,7 @@ public Map<String, String> toMap() {
 
     @Override
     public boolean equals(Object obj) {
-      if (obj == null || this.getClass() != obj.getClass()) {
+      if (!(obj instanceof GlobalJobParametersImpl)) {
         return false;
       }
 

runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeInformation.java

@@ -110,7 +110,7 @@ public boolean equals(@Nullable Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (!(o instanceof CoderTypeInformation)) {
       return false;
     }
 

runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java

@@ -339,7 +339,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkBroadcastValueState)) {
         return false;
       }
 
@@ -430,7 +430,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkBroadcastBagState)) {
         return false;
       }
 
@@ -545,7 +545,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkCombiningState)) {
         return false;
       }
 
@@ -677,7 +677,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkCombiningStateWithContext)) {
         return false;
       }
 

runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java

@@ -134,7 +134,7 @@ public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof StateAndNamespaceDescriptor<?>)) {
         return false;
       }
       StateAndNamespaceDescriptor<?> other = (StateAndNamespaceDescriptor<?>) o;
@@ -580,7 +580,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkValueState<?>)) {
         return false;
       }
 
@@ -840,7 +840,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBagState<?>)) {
         return false;
       }
 
@@ -1002,7 +1002,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkCombiningState<?, ?, ?, ?>)) {
         return false;
       }
 
@@ -1167,7 +1167,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkCombiningStateWithContext<?, ?, ?, ?>)) {
         return false;
       }
 
@@ -1289,7 +1289,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkStateInternals.FlinkWatermarkHoldState)) {
         return false;
       }
 
@@ -1511,7 +1511,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkMapState<?, ?>)) {
         return false;
       }
 
@@ -1660,7 +1660,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkSetState<?>)) {
         return false;
       }
 

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java

@@ -336,7 +336,7 @@ public Map<String, String> toMap() {
 
     @Override
     public boolean equals(Object obj) {
-      if (obj == null || this.getClass() != obj.getClass()) {
+      if (!(obj instanceof GlobalJobParametersImpl)) {
         return false;
       }
 

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeInformation.java

@@ -109,7 +109,7 @@ public boolean equals(@Nullable Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (!(o instanceof CoderTypeInformation)) {
       return false;
     }
 

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java

@@ -137,7 +137,7 @@ public boolean equals(@Nullable Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (!(o instanceof CoderTypeSerializer)) {
       return false;
     }
 

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java

@@ -339,7 +339,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkBroadcastValueState)) {
         return false;
       }
 
@@ -430,7 +430,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkBroadcastBagState)) {
         return false;
       }
 
@@ -545,7 +545,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkCombiningState)) {
         return false;
       }
 
@@ -677,7 +677,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBroadcastStateInternals.FlinkCombiningStateWithContext)) {
         return false;
       }
 

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java

@@ -134,7 +134,7 @@ public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof StateAndNamespaceDescriptor<?>)) {
         return false;
       }
       StateAndNamespaceDescriptor<?> other = (StateAndNamespaceDescriptor<?>) o;
@@ -580,7 +580,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkValueState<?>)) {
         return false;
       }
 
@@ -840,7 +840,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkBagState<?>)) {
         return false;
       }
 
@@ -1002,7 +1002,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkCombiningState<?, ?, ?, ?>)) {
         return false;
       }
 
@@ -1167,7 +1167,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkCombiningStateWithContext<?, ?, ?, ?>)) {
         return false;
       }
 
@@ -1289,7 +1289,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkStateInternals.FlinkWatermarkHoldState)) {
         return false;
       }
 
@@ -1511,7 +1511,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkMapState<?, ?>)) {
         return false;
       }
 
@@ -1660,7 +1660,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof FlinkSetState<?>)) {
         return false;
       }
 

runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/PrimitiveParDoSingleFactoryTest.java

@@ -156,7 +156,7 @@ public void toLong(ProcessContext ctxt) {
 
     @Override
     public boolean equals(@Nullable Object other) {
-      return other != null && other.getClass().equals(getClass());
+      return other instanceof ToLongFn;
     }
 
     @Override

runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/CloudObjectsTest.java

@@ -276,7 +276,7 @@ public Object decode(InputStream inStream) throws CoderException, IOException {
 
     @Override
     public boolean equals(@Nullable Object other) {
-      return other != null && getClass().equals(other.getClass());
+      return other instanceof ObjectCoder;
     }
 
     @Override
@@ -321,7 +321,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      return o != null && getClass() == o.getClass();
+      return o instanceof RowIdentity;
     }
   }
 }

runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ReferenceCountingExecutableStageContextFactory.java

@@ -222,7 +222,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof WrappedContext)) {
         return false;
       }
       WrappedContext that = (WrappedContext) o;

runners/jet/src/main/java/org/apache/beam/runners/jet/Utils.java

@@ -280,7 +280,7 @@ public boolean equals(@Nullable Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (!(o instanceof ByteArrayKey)) {
         return false;
       }
       ByteArrayKey that = (ByteArrayKey) o;

runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/KeyedTimerData.java

@@ -121,7 +121,7 @@ public boolean equals(@Nullable Object o) {
       return true;
     }
 
-    if (o == null || getClass() != o.getClass()) {
+    if (!(o instanceof KeyedTimerData)) {
       return false;
     }