[ErrorProne] Enable UnnecessaryParentheses validation and fix all violations (#37913)

* Enable UnnecessaryParentheses validation and fix all violations

* Enable UnnecessaryParentheses validation and fix all violations

* fix avro format function

Author: stankiewicz

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

@@ -1551,7 +1551,6 @@ class BeamModulePlugin implements Plugin<Project> {
             "ThreadPriorityCheck",
             "UndefinedEquals",
             "UnescapedEntity",
-            "UnnecessaryParentheses",
             "UnrecognisedJavadocTag",
             // errorprone 3.2.0+ checks
             "DirectInvocationOnMock",

it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java

@@ -244,7 +244,7 @@ public void testWriteAndRead() throws IOException {
                 .withAvroFormatFunction(
                     new AvroFormatFn(
                         configuration.numColumns,
-                        !("STORAGE_WRITE_API".equalsIgnoreCase(configuration.writeMethod))));
+                        !"STORAGE_WRITE_API".equalsIgnoreCase(configuration.writeMethod)));
         break;
       case JSON:
         writeIO =

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

@@ -302,7 +302,7 @@ private BundleOutputManager(Map<TupleTag<?>, UncommittedBundle<?>> bundles) {
     @Override
     public <OutputT> void output(TupleTag<OutputT> tag, WindowedValue<OutputT> output) {
       checkArgument(bundles.containsKey(tag), "Unknown output tag %s", tag);
-      (bundles.get(tag)).add((WindowedValue) output);
+      bundles.get(tag).add((WindowedValue) output);
     }
   }
 }

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/unbounded/FlinkUnboundedSourceReader.java

@@ -128,7 +128,7 @@ public void start() {
         pipelineOptions.as(FlinkPipelineOptions.class).getAutoWatermarkInterval();
     if (watermarkInterval == null) {
       watermarkInterval =
-          (pipelineOptions.as(FlinkPipelineOptions.class).getMaxBundleTimeMills()) / 5L;
+          pipelineOptions.as(FlinkPipelineOptions.class).getMaxBundleTimeMills() / 5L;
       watermarkInterval =
           (watermarkInterval > MIN_WATERMARK_EMIT_INTERVAL_MS)
               ? watermarkInterval

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

@@ -287,7 +287,7 @@ private void testTriggerCombinerLiftingDisabled(Trigger trigger) throws Exceptio
 
   @Test
   public void testRepeatedCountTriggerDisablesCombinerLifting() throws IOException, Exception {
-    testTriggerCombinerLiftingDisabled(Repeatedly.forever((AfterPane.elementCountAtLeast(1))));
+    testTriggerCombinerLiftingDisabled(Repeatedly.forever(AfterPane.elementCountAtLeast(1)));
   }
 
   @Test

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

@@ -478,7 +478,7 @@ private synchronized void recordActiveMessageInProcessingTimesMap() {
         return;
       }
       int processingTime =
-          (int) (this.activeMessageMetadata.stopwatch().elapsed(TimeUnit.MILLISECONDS));
+          (int) this.activeMessageMetadata.stopwatch().elapsed(TimeUnit.MILLISECONDS);
       this.processingTimesByStep.compute(
           this.activeMessageMetadata.userStepName(),
           (k, v) -> {

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/ShuffleEntry.java

@@ -96,10 +96,10 @@ public boolean equals(@Nullable Object o) {
     }
     if (o instanceof ShuffleEntry) {
       ShuffleEntry that = (ShuffleEntry) o;
-      return (Objects.equals(this.position, that.position))
-          && (Objects.equals(this.key, that.key))
-          && (Objects.equals(this.secondaryKey, that.secondaryKey))
-          && (Objects.equals(this.value, that.value));
+      return Objects.equals(this.position, that.position)
+          && Objects.equals(this.key, that.key)
+          && Objects.equals(this.secondaryKey, that.secondaryKey)
+          && Objects.equals(this.value, that.value);
     }
     return false;
   }

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/getdata/ApplianceGetDataClient.java

@@ -104,7 +104,7 @@ private void issueReadBatch(ReadBatch batch) {
     } catch (InterruptedException e) {
       // We don't expect this thread to be interrupted. To simplify handling, we just fall through
       // to issuing the call.
-      assert (false);
+      assert false;
       Thread.currentThread().interrupt();
     } catch (ExecutionException e) {
       // startRead is a SettableFuture so this should never occur.
@@ -185,7 +185,7 @@ private void issueReadBatch(ReadBatch batch) {
     synchronized (this) {
       ReadBatch batch;
       if (activeReadThreads < MAX_ACTIVE_READS) {
-        assert (pendingReadBatches.isEmpty());
+        assert pendingReadBatches.isEmpty();
         activeReadThreads += 1;
         // fall through to below synchronized block
       } else if (pendingReadBatches.isEmpty()

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillMultimap.java

@@ -548,7 +548,7 @@ private Map<Object, Triple<K, Boolean, ConcatIterables<V>>> mergedCachedEntries(
       keyStateMap
           .entrySet()
           .removeIf(
-              (entry -> {
+              entry -> {
                 Object structuralKey = entry.getKey();
                 KeyState keyState = entry.getValue();
                 if (complete && keyState.existence == KeyExistence.KNOWN_EXIST) {
@@ -574,7 +574,7 @@ private Map<Object, Triple<K, Boolean, ConcatIterables<V>>> mergedCachedEntries(
                 return (keyState.existence == KeyExistence.KNOWN_NONEXISTENT
                         && !keyState.removedLocally)
                     || keyState.existence == KeyExistence.UNKNOWN_EXISTENCE;
-              }));
+              });
       return cachedEntries;
     }
 

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

@@ -222,7 +222,7 @@ private <K, InputT, AccumT, OutputT> ParDoFn createCombineValuesFn(
   public void testCombineValuesFnAll() throws Exception {
     TestReceiver receiver = new TestReceiver();
 
-    Combine.CombineFn<Integer, CountSum, String> combiner = (new MeanInts());
+    Combine.CombineFn<Integer, CountSum, String> combiner = new MeanInts();
 
     ParDoFn combineParDoFn =
         createCombineValuesFn(