Make WindowedValue a public interface and receivers for it

When we introduce OutputBuilder, it will implement WindowedValue,
and every time we create an OutputBuilder we will want to provide
the following:

1. the default metadata for the output
2. the receiver for the output

This change unblocks 2 and reduces the size of the final change.

Author: kennknowles

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

@@ -1233,6 +1233,7 @@ class BeamModulePlugin implements Plugin<Project> {
           "-AskipUses=${skipUsesCombinedRegex}",
           "-AnoWarnMemoryConstraints",
           "-AsuppressWarnings=annotation.not.completed,keyfor",
+          "-implicit:none"
         ]
 
         project.dependencies {

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

@@ -20,7 +20,7 @@
 import org.apache.beam.sdk.state.TimeDomain;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValue;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Instant;
 

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

@@ -29,10 +29,11 @@
 import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
 import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.util.WindowedValueMultiReceiver;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollectionView;
 import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.WindowedValue;
 import org.apache.beam.sdk.values.WindowingStrategy;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -41,12 +42,6 @@
   "nullness" // TODO(https://github.com/apache/beam/issues/20497)
 })
 public class DoFnRunners {
-  /** Information about how to create output receivers and output to them. */
-  public interface OutputManager {
-    /** Outputs a single element to the receiver indicated by the given {@link TupleTag}. */
-    <T> void output(TupleTag<T> tag, WindowedValue<T> output);
-  }
-
   /**
    * Returns an implementation of {@link DoFnRunner} that for a {@link DoFn}.
    *
@@ -58,7 +53,7 @@ public static <InputT, OutputT> DoFnRunner<InputT, OutputT> simpleRunner(
       PipelineOptions options,
       DoFn<InputT, OutputT> fn,
       SideInputReader sideInputReader,
-      OutputManager outputManager,
+      WindowedValueMultiReceiver outputManager,
       TupleTag<OutputT> mainOutputTag,
       List<TupleTag<?>> additionalOutputTags,
       StepContext stepContext,
@@ -168,7 +163,7 @@ ProcessFnRunner<InputT, OutputT, RestrictionT> newProcessFnRunner(
           PipelineOptions options,
           Collection<PCollectionView<?>> views,
           ReadyCheckingSideInputReader sideInputReader,
-          OutputManager outputManager,
+          WindowedValueMultiReceiver outputManager,
           TupleTag<OutputT> mainOutputTag,
           List<TupleTag<?>> additionalOutputTags,
           StepContext stepContext,

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

@@ -17,20 +17,17 @@
  */
 package org.apache.beam.runners.core;
 
-import java.util.Collection;
 import org.apache.beam.model.pipeline.v1.RunnerApi;
 import org.apache.beam.runners.core.triggers.ExecutableTriggerStateMachine;
 import org.apache.beam.runners.core.triggers.TriggerStateMachines;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.util.SystemDoFnInternal;
-import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.util.WindowedValueMultiReceiver;
 import org.apache.beam.sdk.util.construction.TriggerTranslation;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.WindowingStrategy;
-import org.joda.time.Instant;
 
 /**
  * A general {@link GroupAlsoByWindowsAggregators}. This delegates all of the logic to the {@link
@@ -51,7 +48,7 @@ DoFn<KeyedWorkItem<K, InputT>, KV<K, OutputT>> create(
           TimerInternalsFactory<K> timerInternalsFactory,
           SideInputReader sideInputReader,
           SystemReduceFn<K, InputT, ?, OutputT, W> reduceFn,
-          DoFnRunners.OutputManager outputManager,
+          WindowedValueMultiReceiver outputManager,
           TupleTag<KV<K, OutputT>> mainTag) {
     return new GroupAlsoByWindowViaWindowSetNewDoFn<>(
         strategy,
@@ -68,7 +65,7 @@ DoFn<KeyedWorkItem<K, InputT>, KV<K, OutputT>> create(
   private transient StateInternalsFactory<K> stateInternalsFactory;
   private transient TimerInternalsFactory<K> timerInternalsFactory;
   private transient SideInputReader sideInputReader;
-  private transient DoFnRunners.OutputManager outputManager;
+  private transient WindowedValueMultiReceiver outputManager;
   private TupleTag<KV<K, OutputT>> mainTag;
 
   public GroupAlsoByWindowViaWindowSetNewDoFn(
@@ -77,7 +74,7 @@ public GroupAlsoByWindowViaWindowSetNewDoFn(
       TimerInternalsFactory<K> timerInternalsFactory,
       SideInputReader sideInputReader,
       SystemReduceFn<K, InputT, ?, OutputT, W> reduceFn,
-      DoFnRunners.OutputManager outputManager,
+      WindowedValueMultiReceiver outputManager,
       TupleTag<KV<K, OutputT>> mainTag) {
     this.timerInternalsFactory = timerInternalsFactory;
     this.sideInputReader = sideInputReader;
@@ -91,29 +88,6 @@ public GroupAlsoByWindowViaWindowSetNewDoFn(
     this.triggerProto = TriggerTranslation.toProto(windowingStrategy.getTrigger());
   }
 
-  private OutputWindowedValue<KV<K, OutputT>> outputWindowedValue() {
-    return new OutputWindowedValue<KV<K, OutputT>>() {
-      @Override
-      public void outputWindowedValue(
-          KV<K, OutputT> output,
-          Instant timestamp,
-          Collection<? extends BoundedWindow> windows,
-          PaneInfo pane) {
-        outputManager.output(mainTag, WindowedValue.of(output, timestamp, windows, pane));
-      }
-
-      @Override
-      public <AdditionalOutputT> void outputWindowedValue(
-          TupleTag<AdditionalOutputT> tag,
-          AdditionalOutputT output,
-          Instant timestamp,
-          Collection<? extends BoundedWindow> windows,
-          PaneInfo pane) {
-        outputManager.output(tag, WindowedValue.of(output, timestamp, windows, pane));
-      }
-    };
-  }
-
   @ProcessElement
   public void processElement(ProcessContext c) throws Exception {
     KeyedWorkItem<K, InputT> keyedWorkItem = c.element();
@@ -130,7 +104,7 @@ public void processElement(ProcessContext c) throws Exception {
                 TriggerStateMachines.stateMachineForTrigger(triggerProto)),
             stateInternals,
             timerInternals,
-            outputWindowedValue(),
+            windowedValue -> outputManager.output(mainTag, windowedValue),
             sideInputReader,
             reduceFn,
             c.getPipelineOptions());

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

@@ -29,10 +29,10 @@
 import org.apache.beam.sdk.transforms.GroupByKey;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.WindowedValue.WindowedValueCoder;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValues.WindowedValueCoder;
 import org.apache.beam.sdk.values.WindowingStrategy;
 
 /**

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

@@ -18,7 +18,7 @@
 package org.apache.beam.runners.core;
 
 import org.apache.beam.runners.core.TimerInternals.TimerData;
-import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValue;
 
 /**
  * Interface that contains all the timers and elements associated with a specific work item.

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

@@ -28,8 +28,8 @@
 import org.apache.beam.sdk.coders.IterableCoder;
 import org.apache.beam.sdk.coders.StructuredCoder;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.WindowedValue.FullWindowedValueCoder;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValues.FullWindowedValueCoder;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 
 /** A {@link Coder} for {@link KeyedWorkItem KeyedWorkItems}. */

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

@@ -20,7 +20,7 @@
 import java.util.Collections;
 import java.util.Objects;
 import org.apache.beam.runners.core.TimerInternals.TimerData;
-import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValue;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
 import org.checkerframework.checker.nullness.qual.Nullable;

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

@@ -25,8 +25,9 @@
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.WindowTracing;
-import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValues;
 import org.apache.beam.sdk.values.WindowingStrategy;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.joda.time.Instant;
@@ -140,7 +141,7 @@ public <K, InputT> Iterable<WindowedValue<InputT>> filter(
                 timerInternals.currentOutputWatermarkTime());
           } else {
             nonLateElements.add(
-                WindowedValue.of(
+                WindowedValues.of(
                     element.getValue(), element.getTimestamp(), window, element.getPane()));
           }
         }

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

@@ -21,7 +21,7 @@
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
 import org.apache.beam.sdk.util.WindowTracing;
-import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValue;
 import org.apache.beam.sdk.values.WindowingStrategy;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.FluentIterable;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;