[ErrorProne] Fix BadInstanceof and enable check (#37753)

Author: stankiewicz

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

@@ -1535,7 +1535,6 @@ class BeamModulePlugin implements Plugin<Project> {
             "AutoValueImmutableFields",
             "AutoValueSubclassLeaked",
             "BadImport",
-            "BadInstanceof",
             "BigDecimalEquals",
             "ComparableType",
             "DoNotMockAutoValue",

runners/flink/2.0/src/test/java/org/apache/beam/runners/flink/translation/functions/ImpulseSourceFunctionTest.java

@@ -201,8 +201,7 @@ private static class ImpulseElementMatcher implements ArgumentMatcher<WindowedVa
 
     @Override
     public boolean matches(WindowedValue<byte[]> o) {
-      return o instanceof WindowedValue
-          && Arrays.equals((byte[]) ((WindowedValue) o).getValue(), new byte[] {});
+      return o != null && Arrays.equals(o.getValue(), new byte[] {});
     }
   }
 }

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

@@ -123,7 +123,7 @@ public <T> Coder<T> getInputCoder(PCollection<T> collection) {
 
   public Map<TupleTag<?>, Coder<?>> getOutputCoders() {
     return currentTransform.getOutputs().entrySet().stream()
-        .filter(e -> e.getValue() instanceof PCollection)
+        .filter(e -> e.getValue() != null)
         .collect(Collectors.toMap(e -> e.getKey(), e -> ((PCollection) e.getValue()).getCoder()));
   }
 

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

@@ -74,7 +74,7 @@ public void visitPrimitiveTransform(TransformHierarchy.Node node) {
 
   private boolean hasUnboundedOutput(AppliedPTransform<?, ?, ?> transform) {
     return transform.getOutputs().values().stream()
-        .filter(value -> value instanceof PCollection)
+        .filter(value -> value != null)
         .map(value -> (PCollection<?>) value)
         .anyMatch(collection -> collection.isBounded() == IsBounded.UNBOUNDED);
   }

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/utils/LookupPipelineVisitor.java

@@ -92,7 +92,7 @@ public <T extends PValue> T getOutput(PTransform<?, T> transform) {
   @SuppressWarnings("unchecked")
   public Map<TupleTag<?>, Coder<?>> getOutputCoders(PTransform<?, ?> transform) {
     return getOutputs(transform).entrySet().stream()
-        .filter(e -> e.getValue() instanceof PCollection)
+        .filter(e -> e.getValue() != null)
         .collect(Collectors.toMap(Map.Entry::getKey, e -> ((PCollection) e.getValue()).getCoder()));
   }
 }

runners/flink/src/test/java/org/apache/beam/runners/flink/translation/functions/ImpulseSourceFunctionTest.java

@@ -201,8 +201,7 @@ private static class ImpulseElementMatcher implements ArgumentMatcher<WindowedVa
 
     @Override
     public boolean matches(WindowedValue<byte[]> o) {
-      return o instanceof WindowedValue
-          && Arrays.equals((byte[]) ((WindowedValue) o).getValue(), new byte[] {});
+      return o != null && Arrays.equals(o.getValue(), new byte[] {});
     }
   }
 }

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

@@ -147,7 +147,7 @@ static <T> Map<T, Coder> getCoders(
   static Map<TupleTag<?>, Coder<?>> getOutputValueCoders(
       AppliedPTransform<?, ?, ?> appliedTransform) {
     return appliedTransform.getOutputs().entrySet().stream()
-        .filter(e -> e.getValue() instanceof PCollection)
+        .filter(e -> e.getValue() != null)
         .collect(Collectors.toMap(Map.Entry::getKey, e -> ((PCollection) e.getValue()).getCoder()));
   }
 

runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ParDoBoundMultiTranslator.java

@@ -113,7 +113,7 @@ private static <InT, OutT> void doTranslate(
     final PCollection<? extends InT> input = ctx.getInput(transform);
     final Map<TupleTag<?>, Coder<?>> outputCoders =
         ctx.getCurrentTransform().getOutputs().entrySet().stream()
-            .filter(e -> e.getValue() instanceof PCollection)
+            .filter(e -> e.getValue() != null)
             .collect(
                 Collectors.toMap(e -> e.getKey(), e -> ((PCollection<?>) e.getValue()).getCoder()));
 
@@ -145,7 +145,7 @@ private static <InT, OutT> void doTranslate(
       final Map.Entry<TupleTag<?>, PCollection<?>> taggedOutput = outputs.get(index);
       tagToIndexMap.put(taggedOutput.getKey(), index);
 
-      if (!(taggedOutput.getValue() instanceof PCollection)) {
+      if (taggedOutput.getValue() == null) {
         throw new IllegalArgumentException(
             "Expected side output to be PCollection, but was: " + taggedOutput.getValue());
       }

runners/samza/src/main/java/org/apache/beam/runners/samza/translation/SplittableParDoTranslators.java

@@ -78,7 +78,7 @@ public void translate(
         final Map.Entry<TupleTag<?>, PCollection<?>> taggedOutput = outputs.get(index);
         tagToIndexMap.put(taggedOutput.getKey(), index);
 
-        if (!(taggedOutput.getValue() instanceof PCollection)) {
+        if (taggedOutput.getValue() == null) {
           throw new IllegalArgumentException(
               "Expected side output to be PCollection, but was: " + taggedOutput.getValue());
         }

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java

@@ -144,7 +144,7 @@ public Map<TupleTag<?>, PCollection<?>> getOutputs(PTransform<?, ?> transform) {
 
   public Map<TupleTag<?>, Coder<?>> getOutputCoders() {
     return currentTransform.getOutputs().entrySet().stream()
-        .filter(e -> e.getValue() instanceof PCollection)
+        .filter(e -> e.getValue() != null)
         .collect(Collectors.toMap(Map.Entry::getKey, e -> ((PCollection) e.getValue()).getCoder()));
   }