review feedback - remove duplicated code and unused variables

Author: stankiewicz

runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java

@@ -479,12 +479,7 @@ public <T> void outputWindowedValue(
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
-      noteOutput();
-      if (watermarkEstimator instanceof TimestampObservingWatermarkEstimator) {
-        ((TimestampObservingWatermarkEstimator) watermarkEstimator)
-            .observeTimestamp(windowedValue.getTimestamp());
-      }
-      outputReceiver.output(mainOutputTag, windowedValue);
+      outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

@@ -445,8 +445,7 @@ public <T> void output(TupleTag<T> tag, T output) {
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
-      checkTimestamp(elem.getTimestamp(), windowedValue.getTimestamp());
-      SimpleDoFnRunner.this.outputWindowedValue(mainOutputTag, windowedValue);
+      outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override
@@ -1041,8 +1040,7 @@ public <T> void outputWindowedValue(
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
-      checkTimestamp(timestamp(), windowedValue.getTimestamp());
-      SimpleDoFnRunner.this.outputWindowedValue(mainOutputTag, windowedValue);
+      outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override
@@ -1312,8 +1310,7 @@ public <T> void outputWindowedValue(
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
-      checkTimestamp(this.timestamp, windowedValue.getTimestamp());
-      SimpleDoFnRunner.this.outputWindowedValue(mainOutputTag, windowedValue);
+      outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override

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

@@ -651,18 +651,7 @@ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
-      for (BoundedWindow w : windowedValue.getWindows()) {
-        getMutableOutput(mainOutputTag)
-            .add(
-                ValueInSingleWindow.of(
-                    windowedValue.getValue(),
-                    windowedValue.getTimestamp(),
-                    w,
-                    windowedValue.getPaneInfo(),
-                    windowedValue.getRecordId(),
-                    windowedValue.getRecordOffset(),
-                    windowedValue.causedByDrain()));
-      }
+      outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override

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

@@ -30,9 +30,6 @@
 import org.junit.runners.JUnit4;
 
 public class MetadataPropagationTest {
-  private static final Integer[] EMPTY = new Integer[] {};
-  private static final Integer[] DATA = new Integer[] {1, 2, 3, 4, 5};
-  private static final Integer[] REPEATED_DATA = new Integer[] {1, 1, 2, 2, 3, 3, 4, 4, 5, 5};
 
   @RunWith(JUnit4.class)
   public static class MiscTest {