Merge pull request #7523: Apply spotless across Beam

Author: reuvenlax

examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java

@@ -186,7 +186,8 @@ static void runWindowedWordCount(Options options) throws IOException {
         pipeline
             /* Read from the GCS file. */
             .apply(TextIO.read().from(options.getInputFile()))
-            // Concept #2: Add an element timestamp, using an artificial time just to show windowing.
+            // Concept #2: Add an element timestamp, using an artificial time just to show
+            // windowing.
             // See AddTimestampFn for more detail on this.
             .apply(ParDo.of(new AddTimestampFn(minTimestamp, maxTimestamp)));
 

examples/java/src/main/java/org/apache/beam/examples/common/ExampleUtils.java

@@ -224,7 +224,8 @@ private static Bigquery.Builder newBigQueryClient(BigQueryOptions options) {
             Transport.getJsonFactory(),
             chainHttpRequestInitializer(
                 options.getGcpCredential(),
-                // Do not log 404. It clutters the output and is possibly even required by the caller.
+                // Do not log 404. It clutters the output and is possibly even required by the
+                // caller.
                 new RetryHttpRequestInitializer(ImmutableList.of(404))))
         .setApplicationName(options.getAppName())
         .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
@@ -237,7 +238,8 @@ private static Pubsub.Builder newPubsubClient(PubsubOptions options) {
             Transport.getJsonFactory(),
             chainHttpRequestInitializer(
                 options.getGcpCredential(),
-                // Do not log 404. It clutters the output and is possibly even required by the caller.
+                // Do not log 404. It clutters the output and is possibly even required by the
+                // caller.
                 new RetryHttpRequestInitializer(ImmutableList.of(404))))
         .setRootUrl(options.getPubsubRootUrl())
         .setApplicationName(options.getAppName())

examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java

@@ -122,7 +122,7 @@
  * and then exits.
  */
 public class TriggerExample {
-  //Numeric value of fixed window duration, in minutes
+  // Numeric value of fixed window duration, in minutes
   public static final int WINDOW_DURATION = 30;
   // Constants used in triggers.
   // Speeding up ONE_MINUTE or FIVE_MINUTES helps you get an early approximation of results.
@@ -189,18 +189,22 @@ public PCollectionList<TableRow> expand(PCollection<KV<String, Integer>> flowInf
               .apply(
                   "Default",
                   Window
-                      // The default window duration values work well if you're running the default input
+                      // The default window duration values work well if you're running the default
+                      // input
                       // file. You may want to adjust the window duration otherwise.
                       .<KV<String, Integer>>into(
                           FixedWindows.of(Duration.standardMinutes(windowDuration)))
-                      // The default trigger first emits output when the system's watermark passes the end
+                      // The default trigger first emits output when the system's watermark passes
+                      // the end
                       // of the window.
                       .triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow()))
                       // Late data is dropped
                       .withAllowedLateness(Duration.ZERO)
                       // Discard elements after emitting each pane.
-                      // With no allowed lateness and the specified trigger there will only be a single
-                      // pane, so this doesn't have a noticeable effect. See concept 2 for more details.
+                      // With no allowed lateness and the specified trigger there will only be a
+                      // single
+                      // pane, so this doesn't have a noticeable effect. See concept 2 for more
+                      // details.
                       .discardingFiredPanes())
               .apply(new TotalFlow("default"));
 
@@ -229,7 +233,8 @@ public PCollectionList<TableRow> expand(PCollection<KV<String, Integer>> flowInf
                           FixedWindows.of(Duration.standardMinutes(windowDuration)))
                       // Late data is emitted as it arrives
                       .triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow()))
-                      // Once the output is produced, the pane is dropped and we start preparing the next
+                      // Once the output is produced, the pane is dropped and we start preparing the
+                      // next
                       // pane for the window
                       .discardingFiredPanes()
                       // Late data is handled up to one day
@@ -264,8 +269,10 @@ public PCollectionList<TableRow> expand(PCollection<KV<String, Integer>> flowInf
                               AfterProcessingTime.pastFirstElementInPane()
                                   // Speculative every ONE_MINUTE
                                   .plusDelayOf(ONE_MINUTE)))
-                      // After emitting each pane, it will continue accumulating the elements so that each
-                      // approximation includes all of the previous data in addition to the newly arrived
+                      // After emitting each pane, it will continue accumulating the elements so
+                      // that each
+                      // approximation includes all of the previous data in addition to the newly
+                      // arrived
                       // data.
                       .accumulatingFiredPanes()
                       .withAllowedLateness(ONE_DAY))
@@ -414,7 +421,7 @@ public void processElement(ProcessContext c) throws Exception {
         return;
       }
       if (laneInfo.length < VALID_NUM_FIELDS) {
-        //Skip the invalid input.
+        // Skip the invalid input.
         return;
       }
       String freeway = laneInfo[2];

examples/java/src/test/java/org/apache/beam/examples/complete/game/LeaderBoardTest.java

@@ -129,14 +129,16 @@ public void testTeamScoresSpeculative() {
             .addElements(
                 event(TestUser.BLUE_ONE, 3, Duration.standardSeconds(3)),
                 event(TestUser.BLUE_ONE, 2, Duration.standardMinutes(1)))
-            // Some time passes within the runner, which causes a speculative pane containing the blue
+            // Some time passes within the runner, which causes a speculative pane containing the
+            // blue
             // team's score to be emitted
             .advanceProcessingTime(Duration.standardMinutes(10))
             .addElements(event(TestUser.RED_TWO, 5, Duration.standardMinutes(3)))
             // Some additional time passes and we get a speculative pane for the red team
             .advanceProcessingTime(Duration.standardMinutes(12))
             .addElements(event(TestUser.BLUE_TWO, 3, Duration.standardSeconds(22)))
-            // More time passes and a speculative pane containing a refined value for the blue pane is
+            // More time passes and a speculative pane containing a refined value for the blue pane
+            // is
             // emitted
             .advanceProcessingTime(Duration.standardMinutes(10))
             // Some more events occur
@@ -238,7 +240,8 @@ public void testTeamScoresObservablyLate() {
                 event(TestUser.RED_TWO, 2, Duration.ZERO),
                 event(TestUser.RED_TWO, 5, Duration.standardMinutes(1)),
                 event(TestUser.RED_TWO, 3, Duration.standardMinutes(3)))
-            // A late refinement is emitted due to the advance in processing time, but the window has
+            // A late refinement is emitted due to the advance in processing time, but the window
+            // has
             // not yet closed because the watermark has not advanced
             .advanceProcessingTime(Duration.standardMinutes(12))
             // These elements should appear in the final pane
@@ -303,7 +306,8 @@ public void testTeamScoresDroppablyLate() {
                     .plus(ALLOWED_LATENESS)
                     .plus(TEAM_WINDOW_DURATION)
                     .plus(Duration.standardMinutes(1)))
-            // These elements within the expired window are droppably late, and will not appear in the
+            // These elements within the expired window are droppably late, and will not appear in
+            // the
             // output
             .addElements(
                 event(

runners/apex/src/main/java/org/apache/beam/runners/apex/TestApexRunner.java

@@ -33,7 +33,7 @@ public class TestApexRunner extends PipelineRunner<ApexRunnerResult> {
 
   private TestApexRunner(ApexPipelineOptions options) {
     options.setEmbeddedExecution(true);
-    //options.setEmbeddedExecutionDebugMode(false);
+    // options.setEmbeddedExecutionDebugMode(false);
     this.delegate = ApexRunner.fromOptions(options);
   }
 

runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoTranslator.java

@@ -77,9 +77,7 @@ public void translate(ParDo.MultiOutput<InputT, OutputT> transform, TranslationC
     List<PCollectionView<?>> sideInputs = transform.getSideInputs();
 
     Map<TupleTag<?>, Coder<?>> outputCoders =
-        outputs
-            .entrySet()
-            .stream()
+        outputs.entrySet().stream()
             .filter(e -> e.getValue() instanceof PCollection)
             .collect(
                 Collectors.toMap(e -> e.getKey(), e -> ((PCollection) e.getValue()).getCoder()));
@@ -138,9 +136,7 @@ public void translate(
       List<PCollectionView<?>> sideInputs = transform.getSideInputs();
 
       Map<TupleTag<?>, Coder<?>> outputCoders =
-          outputs
-              .entrySet()
-              .stream()
+          outputs.entrySet().stream()
               .filter(e -> e.getValue() instanceof PCollection)
               .collect(
                   Collectors.toMap(e -> e.getKey(), e -> ((PCollection) e.getValue()).getCoder()));
@@ -221,8 +217,8 @@ private static PCollection<?> unionSideInputs(
           .getWindowingStrategy()
           .equals(firstSideInput.getWindowingStrategy())) {
         // TODO: check how to handle this in stream codec
-        //String msg = "Multiple side inputs with different window strategies.";
-        //throw new UnsupportedOperationException(msg);
+        // String msg = "Multiple side inputs with different window strategies.";
+        // throw new UnsupportedOperationException(msg);
         LOG.warn(
             "Side inputs union with different windowing strategies {} {}",
             firstSideInput.getWindowingStrategy(),

runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java

@@ -382,7 +382,7 @@ private void processWatermark(ApexStreamTuple.WatermarkTuple<?> mark) {
     checkState(
         minEventTimeTimer >= currentInputWatermark,
         "Event time timer processing generates new timer(s) behind watermark.");
-    //LOG.info("Processing time timer {} registered behind watermark {}", minProcessingTimeTimer,
+    // LOG.info("Processing time timer {} registered behind watermark {}", minProcessingTimeTimer,
     //    currentInputWatermark);
 
     // TODO: is this the right way to trigger processing time timers?

runners/apex/src/test/java/org/apache/beam/runners/apex/ApexYarnLauncherTest.java

@@ -56,8 +56,8 @@ public void testGetYarnDeployDependencies() throws Exception {
     List<File> deps = ApexYarnLauncher.getYarnDeployDependencies();
     String depsToString = deps.toString();
     // the beam dependencies are not present as jar when running within the Maven build reactor
-    //assertThat(depsToString, containsString("beam-runners-core-"));
-    //assertThat(depsToString, containsString("beam-runners-apex-"));
+    // assertThat(depsToString, containsString("beam-runners-core-"));
+    // assertThat(depsToString, containsString("beam-runners-apex-"));
     assertThat(depsToString, containsString("apex-common-"));
     assertThat(depsToString, not(containsString("hadoop-")));
     assertThat(depsToString, not(containsString("zookeeper-")));

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

@@ -92,7 +92,8 @@ public static class Registrar implements TransformPayloadTranslatorRegistrar {
   /** Produces a {@link RunnerApi.CombinePayload} from a {@link Combine}. */
   static <K, InputT, OutputT> CombinePayload payloadForCombine(
       final AppliedPTransform<
-              PCollection<KV<K, InputT>>, PCollection<KV<K, OutputT>>,
+              PCollection<KV<K, InputT>>,
+              PCollection<KV<K, OutputT>>,
               Combine.PerKey<K, InputT, OutputT>>
           combine,
       final SdkComponents components)

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

@@ -53,7 +53,8 @@ public class CreatePCollectionViewTranslation {
   @Deprecated
   public static <ElemT, ViewT> PCollectionView<ViewT> getView(
       AppliedPTransform<
-              PCollection<ElemT>, PCollection<ElemT>,
+              PCollection<ElemT>,
+              PCollection<ElemT>,
               PTransform<PCollection<ElemT>, PCollection<ElemT>>>
           application)
       throws IOException {