[ErrorProne] Fix UnusedVariable warnings across codebase

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

@@ -1587,7 +1587,6 @@ class BeamModulePlugin implements Plugin<Project> {
             "StringCharset",
             "SuperCallToObjectMethod",
             "UnnecessaryLongToIntConversion",
-            "UnusedVariable",
             // intended suppressions emerged in newer protobuf versions
             "AutoValueBoxedValues",
             // For backward compatibility. Public method checked in before this check impl

runners/core-java/src/test/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvokerTest.java

@@ -90,7 +90,7 @@ public ProcessContinuation process(
     }
 
     @GetInitialRestriction
-    public OffsetRange getInitialRestriction(@Element Void element) {
+    public OffsetRange getInitialRestriction(@SuppressWarnings("unused") @Element Void element) {
       throw new UnsupportedOperationException("Should not be called in this test");
     }
   }
@@ -209,7 +209,8 @@ public void process(ProcessContext c, RestrictionTracker<OffsetRange, Long> trac
           }
 
           @GetInitialRestriction
-          public OffsetRange getInitialRestriction(@Element Void element) {
+          public OffsetRange getInitialRestriction(
+              @SuppressWarnings("unused") @Element Void element) {
             throw new UnsupportedOperationException("Should not be called in this test");
           }
         };
@@ -228,7 +229,8 @@ public void process(ProcessContext c, RestrictionTracker<OffsetRange, Long> trac
           }
 
           @GetInitialRestriction
-          public OffsetRange getInitialRestriction(@Element Void element) {
+          public OffsetRange getInitialRestriction(
+              @SuppressWarnings("unused") @Element Void element) {
             throw new UnsupportedOperationException("Should not be called in this test");
           }
         };

runners/core-java/src/test/java/org/apache/beam/runners/core/SimplePushbackSideInputDoFnRunnerTest.java

@@ -536,7 +536,8 @@ private static class MyDoFn extends DoFn<KV<String, Integer>, Integer> {
     public final StateSpec<ValueState<Integer>> intState = StateSpecs.value(VarIntCoder.of());
 
     @ProcessElement
-    public void processElement(ProcessContext c, @StateId(stateId) ValueState<Integer> state) {
+    public void processElement(
+        @SuppressWarnings("unused") ProcessContext c, @StateId(stateId) ValueState<Integer> state) {
       Integer currentValue = MoreObjects.firstNonNull(state.read(), 0);
       state.write(currentValue + 1);
     }

runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoProcessFnTest.java

@@ -287,7 +287,8 @@ public void process(ProcessContext c, RestrictionTracker<SomeRestriction, Void>
     }
 
     @GetInitialRestriction
-    public SomeRestriction getInitialRestriction(@Element Integer elem) {
+    public SomeRestriction getInitialRestriction(
+        @SuppressWarnings("unused") @Element Integer elem) {
       return new SomeRestriction();
     }
   }
@@ -342,7 +343,7 @@ public void process(
     }
 
     @GetInitialRestriction
-    public OffsetRange getInitialRestriction(@Element Instant elem) {
+    public OffsetRange getInitialRestriction(@SuppressWarnings("unused") @Element Instant elem) {
       throw new IllegalStateException("Expected to be supplied explicitly in this test");
     }
 
@@ -402,7 +403,8 @@ public ProcessContinuation process(
     }
 
     @GetInitialRestriction
-    public SomeRestriction getInitialRestriction(@Element Integer elem) {
+    public SomeRestriction getInitialRestriction(
+        @SuppressWarnings("unused") @Element Integer elem) {
       return new SomeRestriction();
     }
   }
@@ -465,7 +467,7 @@ public ProcessContinuation process(
     }
 
     @GetInitialRestriction
-    public OffsetRange getInitialRestriction(@Element Integer elem) {
+    public OffsetRange getInitialRestriction(@SuppressWarnings("unused") @Element Integer elem) {
       throw new UnsupportedOperationException("Expected to be supplied explicitly in this test");
     }
   }
@@ -586,12 +588,15 @@ private enum State {
     private State state = State.BEFORE_SETUP;
 
     @ProcessElement
-    public void process(ProcessContext c, RestrictionTracker<SomeRestriction, Void> tracker) {
+    public void process(
+        @SuppressWarnings("unused") ProcessContext c,
+        @SuppressWarnings("unused") RestrictionTracker<SomeRestriction, Void> tracker) {
       assertEquals(State.INSIDE_BUNDLE, state);
     }
 
     @GetInitialRestriction
-    public SomeRestriction getInitialRestriction(@Element Integer element) {
+    public SomeRestriction getInitialRestriction(
+        @SuppressWarnings("unused") @Element Integer element) {
       return new SomeRestriction();
     }
 

runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java

@@ -142,7 +142,9 @@ public ResultT extractLatestAttempted() {
      * @param bundle The bundle being committed.
      * @param finalCumulative The final cumulative value for the given bundle.
      */
-    public void commitLogical(final CommittedBundle<?> bundle, final UpdateT finalCumulative) {
+    public void commitLogical(
+        @SuppressWarnings("unused") final CommittedBundle<?> bundle,
+        final UpdateT finalCumulative) {
       UpdateT current;
       do {
         current = finishedCommitted.get();

runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagerRemovingTransformEvaluatorTest.java

@@ -144,6 +144,6 @@ public void removesOnExceptionInFinishBundle() throws Exception {
 
   private static class TestFn extends DoFn<Object, Object> {
     @ProcessElement
-    public void processElement(ProcessContext c) throws Exception {}
+    public void processElement(@SuppressWarnings("unused") ProcessContext c) throws Exception {}
   }
 }

runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagerTest.java

@@ -219,7 +219,7 @@ public void setup() {
     }
 
     @ProcessElement
-    public void processElement(ProcessContext c) throws Exception {}
+    public void processElement(@SuppressWarnings("unused") ProcessContext c) throws Exception {}
 
     @Teardown
     public void teardown() {

runners/direct-java/src/test/java/org/apache/beam/runners/direct/DoFnLifecycleManagersTest.java

@@ -111,7 +111,7 @@ private ThrowsInCleanupFn(String message) {
     }
 
     @ProcessElement
-    public void processElement(ProcessContext c) throws Exception {}
+    public void processElement(@SuppressWarnings("unused") ProcessContext c) throws Exception {}
 
     @Teardown
     public void teardown() throws Exception {
@@ -165,6 +165,6 @@ public void describeTo(Description description) {
 
   private static class EmptyFn extends DoFn<Object, Object> {
     @ProcessElement
-    public void processElement(ProcessContext c) throws Exception {}
+    public void processElement(@SuppressWarnings("unused") ProcessContext c) throws Exception {}
   }
 }

runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java

@@ -496,7 +496,10 @@ private static class TestUnboundedSource<T> extends UnboundedSource<T, TestCheck
 
     static int readerCreatedCount;
     static int readerClosedCount;
+
+    @SuppressWarnings("unused")
     static int readerAdvancedCount;
+
     private final Coder<T> coder;
     private final List<T> elems;
     private boolean dedupes = false;

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

@@ -47,7 +47,6 @@
 import org.apache.flink.configuration.StateBackendOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
-import org.apache.flink.runtime.state.StateBackend;
 import org.apache.flink.runtime.util.EnvironmentInformation;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
@@ -390,7 +389,6 @@ private static void configureCheckpointing(
   }
 
   private static void configureStateBackend(FlinkPipelineOptions options, Configuration config) {
-    final StateBackend stateBackend;
     if (options.getStateBackend() != null) {
       final String storagePath = options.getStateBackendStoragePath();
       Preconditions.checkArgument(

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

@@ -1306,12 +1306,12 @@ private static void initializeUserState(
 
   private static class NoOpDoFn<InputT, OutputT> extends DoFn<InputT, OutputT> {
     @ProcessElement
-    public void doNothing(ProcessContext context) {}
+    public void doNothing(@SuppressWarnings("unused") ProcessContext context) {}
   }
 
   private static class StableNoOpDoFn<InputT, OutputT> extends DoFn<InputT, OutputT> {
     @RequiresStableInput
     @ProcessElement
-    public void doNothing(ProcessContext context) {}
+    public void doNothing(@SuppressWarnings("unused") ProcessContext context) {}
   }
 }

runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourcePortableTest.java

@@ -169,6 +169,7 @@ public UnboundedReader<Long> createReader(
         PipelineOptions options, @Nullable Checkpoint checkpointMark) {
 
       return new UnboundedReader<Long>() {
+        @SuppressWarnings("unused")
         int pos = -1;
 
         @Override
@@ -230,6 +231,7 @@ public Coder<Checkpoint> getCheckpointMarkCoder() {
     }
 
     private static class Checkpoint implements CheckpointMark, Serializable {
+      @SuppressWarnings("unused")
       final int pos;
 
       Checkpoint(int pos) {

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

@@ -1690,12 +1690,14 @@ private static void assertAllStepOutputsHaveUniqueIds(Job job) throws Exception
   private static class TestSplittableFn extends DoFn<String, Integer> {
 
     @ProcessElement
-    public void process(ProcessContext c, RestrictionTracker<OffsetRange, Long> tracker) {
+    public void process(
+        @SuppressWarnings("unused") ProcessContext c,
+        @SuppressWarnings("unused") RestrictionTracker<OffsetRange, Long> tracker) {
       // noop
     }
 
     @GetInitialRestriction
-    public OffsetRange getInitialRange(@Element String element) {
+    public OffsetRange getInitialRange(@SuppressWarnings("unused") @Element String element) {
       return null;
     }
   }

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/status/DebugCapture.java

@@ -79,11 +79,13 @@ public Page(String name, String content) {
     }
 
     // Implicitly read. Not a bug.
+    @SuppressWarnings("unused")
     @SuppressFBWarnings("URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD")
     @Key
     public String name;
 
     // Implicitly read. Not a bug.
+    @SuppressWarnings("unused")
     @SuppressFBWarnings("URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD")
     @Key
     public String content;

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

@@ -63,7 +63,7 @@ private static class TestDoFn extends DoFn<Integer, String> {
     }
 
     @ProcessElement
-    public void processElement(ProcessContext c) {
+    public void processElement(@SuppressWarnings("unused") ProcessContext c) {
       // This is called to ensure the SimpleParDoFn is fully initialized
     }
   }

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

@@ -48,7 +48,8 @@ private static class TestFn extends DoFn<Object, Object> {
     boolean tornDown = false;
 
     @ProcessElement
-    public void processElement(ProcessContext processContext) throws Exception {}
+    public void processElement(@SuppressWarnings("unused") ProcessContext processContext)
+        throws Exception {}
 
     @Teardown
     public void teardown() throws Exception {

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

@@ -239,6 +239,7 @@ public long getEstimatedSizeBytes(PipelineOptions options) {
   }
 
   private static class SourceWithLargeObject extends MockSource {
+    @SuppressWarnings("unused")
     byte[] array;
 
     public SourceWithLargeObject(int sourceObjectSize) {

runners/java-job-service/src/test/java/org/apache/beam/runners/jobsubmission/PortablePipelineJarCreatorTest.java

@@ -122,7 +122,7 @@ public void testCopyResourcesFromJar_ignoresDuplicates() throws IOException {
   }
 
   private static class FakePipelineRunnner {
-    public static void main(String[] args) {
+    public static void main(@SuppressWarnings("unused") String[] args) {
       System.out.println("Hello world");
     }
   }
@@ -145,7 +145,7 @@ public void testCreateManifest_withoutMainMethod() {
 
   @SuppressWarnings("IncorrectMainMethod") // intended
   private static class EvilPipelineRunner {
-    public static int main(String[] args) {
+    public static int main(@SuppressWarnings("unused") String[] args) {
       return 0;
     }
   }

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

@@ -558,6 +558,6 @@ public void processElement(WindowedValue<RawUnionValue> inputElement, OpEmitter<
 
   private static class NoOpDoFn<InT, OutT> extends DoFn<InT, OutT> {
     @ProcessElement
-    public void doNothing(ProcessContext context) {}
+    public void doNothing(@SuppressWarnings("unused") ProcessContext context) {}
   }
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Select.java

@@ -135,8 +135,7 @@ private static class SelectDoFn<T> extends DoFn<T, Row> {
     @FieldAccess("selectFields")
     final FieldAccessDescriptor fieldAccess = FieldAccessDescriptor.withAllFields();
 
-    public SelectDoFn(
-        FieldAccessDescriptor fieldAccessDescriptor, Schema inputSchema, Schema outputSchema) {
+    public SelectDoFn(FieldAccessDescriptor fieldAccessDescriptor, Schema inputSchema) {
       this.rowSelector = new RowSelectorContainer(inputSchema, fieldAccessDescriptor, true);
     }
 
@@ -204,7 +203,7 @@ public PCollection<Row> expand(PCollection<T> input) {
                 + input.getSchema());
       }
       return input
-          .apply(ParDo.of(new SelectDoFn<>(resolved, inputSchema, outputSchema)))
+          .apply(ParDo.of(new SelectDoFn<>(resolved, inputSchema)))
           .setRowSchema(outputSchema);
     }
   }
@@ -316,7 +315,7 @@ public PCollection<Row> expand(PCollection<T> input) {
         outputSchema = inferredOutputSchema;
       }
       return input
-          .apply(ParDo.of(new SelectDoFn<>(fieldAccessDescriptor, inputSchema, outputSchema)))
+          .apply(ParDo.of(new SelectDoFn<>(fieldAccessDescriptor, inputSchema)))
           .setRowSchema(outputSchema);
     }
   }

sdks/java/core/src/test/java/org/apache/beam/sdk/coders/CoderRegistryTest.java

@@ -282,7 +282,7 @@ private static class PTransformOutputingMySerializableGeneric
     private static class OutputDoFn
         extends DoFn<String, KV<String, MySerializableGeneric<String>>> {
       @ProcessElement
-      public void processElement(ProcessContext c) {}
+      public void processElement(@SuppressWarnings("unused") ProcessContext c) {}
     }
 
     @Override
@@ -339,7 +339,7 @@ private static class GenericOutputMySerializedGeneric<T extends Serializable>
 
     private class OutputDoFn extends DoFn<String, KV<String, MySerializableGeneric<T>>> {
       @ProcessElement
-      public void processElement(ProcessContext c) {}
+      public void processElement(@SuppressWarnings("unused") ProcessContext c) {}
     }
 
     @Override

sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java

@@ -1957,7 +1957,8 @@ private String emptyStringErrorMessage(String type) {
   }
 
   private static class RegisteredTestRunner extends PipelineRunner<PipelineResult> {
-    public static PipelineRunner<PipelineResult> fromOptions(PipelineOptions options) {
+    public static PipelineRunner<PipelineResult> fromOptions(
+        @SuppressWarnings("unused") PipelineOptions options) {
       return new RegisteredTestRunner();
     }
 

sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java

@@ -1247,6 +1247,7 @@ public void testDisplayDataExcludesJsonIgnoreOptions() {
   }
 
   private static class CapturesOptions implements Serializable {
+    @SuppressWarnings("unused")
     PipelineOptions options = PipelineOptionsFactory.create();
   }
 

sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/JsonUtilsTest.java

@@ -94,7 +94,9 @@ private static <T extends RowEncodable> TestCase<T> testCase(T cat) {
 
   private static class TestCase<T extends RowEncodable> {
 
+    @SuppressWarnings("unused")
     final String name;
+
     final T userT;
     final String jsonString;
     final byte[] jsonBytes;

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTesterTest.java

@@ -375,7 +375,7 @@ public void startBundle() {
     }
 
     @ProcessElement
-    public void processElement(ProcessContext c) {
+    public void processElement(@SuppressWarnings("unused") ProcessContext c) {
       elements++;
     }
 

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/WaitTest.java

@@ -462,7 +462,7 @@ public void process(ProcessContext c) {
         ParDo.of(
             new DoFn<Long, Long>() {
               @ProcessElement
-              public void process(ProcessContext c, BoundedWindow w) {
+              public void process(ProcessContext c, @SuppressWarnings("unused") BoundedWindow w) {
                 while (true) {
                   Instant maxMainTimestamp = TEST_WAIT_MAX_MAIN_TIMESTAMP.get();
                   Instant newMaxTimestamp =

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java

@@ -1217,7 +1217,7 @@ public void testCanSerializeItemSpecReference() {
   }
 
   private static class HoldsItemSpecReference implements Serializable {
-    public HoldsItemSpecReference(DisplayData.ItemSpec<?> spec) {}
+    public HoldsItemSpecReference(@SuppressWarnings("unused") DisplayData.ItemSpec<?> spec) {}
   }
 
   @Test

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DefaultPackageTest.java

@@ -47,6 +47,6 @@ public void defaultPackageInvoker() {
 
   private static class MyFn extends DoFn<String, String> {
     @ProcessElement
-    public void processElement(ProcessContext context) {}
+    public void processElement(@SuppressWarnings("unused") ProcessContext context) {}
   }
 }