Remove StepContext.noteOutput
Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/bed1c53f Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/bed1c53f Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/bed1c53f Branch: refs/heads/master Commit: bed1c53fb47e2d623d6671ce69b82579992df642 Parents: 98a7555 Author: Kenneth Knowles <[email protected]> Authored: Mon May 22 15:30:33 2017 -0700 Committer: Kenneth Knowles <[email protected]> Committed: Tue May 23 11:16:27 2017 -0700 ---------------------------------------------------------------------- .../apex/translation/utils/NoOpStepContext.java | 8 -------- .../beam/runners/core/BaseExecutionContext.java | 16 ---------------- .../apache/beam/runners/core/ExecutionContext.java | 16 ---------------- .../apache/beam/runners/core/SimpleDoFnRunner.java | 6 ------ .../org/apache/beam/runners/core/StepContext.java | 15 --------------- .../translation/functions/FlinkNoOpStepContext.java | 10 ---------- .../wrappers/streaming/DoFnOperator.java | 6 ------ .../spark/translation/SparkProcessContext.java | 6 ------ .../beam/fn/harness/fake/FakeStepContext.java | 8 -------- 9 files changed, 91 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java ---------------------------------------------------------------------- diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java index 241a985..51e843b 100644 --- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java +++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java @@ -44,14 +44,6 @@ public class NoOpStepContext implements StepContext, Serializable { } @Override - public void noteOutput(WindowedValue<?> output) { - } - - @Override - public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) { - } - - @Override public <T, W extends BoundedWindow> void writePCollectionViewData(TupleTag<?> tag, Iterable<WindowedValue<T>> data, Coder<Iterable<WindowedValue<T>>> dataCoder, W window, Coder<W> windowCoder) throws http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java ---------------------------------------------------------------------- diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java index ed37143..a006999 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java @@ -104,12 +104,6 @@ public abstract class BaseExecutionContext<T extends StepContext> return Collections.unmodifiableCollection(cachedStepContexts.values()); } - @Override - public void noteOutput(WindowedValue<?> output) {} - - @Override - public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {} - /** * Base class for implementations of {@link org.apache.beam.runners.core.StepContext}. * @@ -138,16 +132,6 @@ public abstract class BaseExecutionContext<T extends StepContext> } @Override - public void noteOutput(WindowedValue<?> output) { - executionContext.noteOutput(output); - } - - @Override - public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) { - executionContext.noteOutput(tag, output); - } - - @Override public <T, W extends BoundedWindow> void writePCollectionViewData( TupleTag<?> tag, Iterable<WindowedValue<T>> data, Coder<Iterable<WindowedValue<T>>> dataCoder, http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java ---------------------------------------------------------------------- diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java index f431c92..eac3599 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java @@ -18,9 +18,6 @@ package org.apache.beam.runners.core; import java.util.Collection; -import org.apache.beam.sdk.transforms.DoFn.WindowedContext; -import org.apache.beam.sdk.util.WindowedValue; -import org.apache.beam.sdk.values.TupleTag; /** * Context for the current execution. This is guaranteed to exist during processing, @@ -36,17 +33,4 @@ public interface ExecutionContext { * Returns a collection view of all of the {@link StepContext}s. */ Collection<? extends StepContext> getAllStepContexts(); - - /** - * Hook for subclasses to implement that will be called whenever - * {@link WindowedContext#output(TupleTag, Object)} is called. - */ - void noteOutput(WindowedValue<?> output); - - /** - * Hook for subclasses to implement that will be called whenever - * {@link WindowedContext#output(TupleTag, Object)} is called. - */ - void noteOutput(TupleTag<?> tag, WindowedValue<?> output); - } http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java ---------------------------------------------------------------------- diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java index adbe62e..97b0b33 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java @@ -352,9 +352,6 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out void outputWindowedValue(WindowedValue<OutputT> windowedElem) { outputManager.output(mainOutputTag, windowedElem); - if (stepContext != null) { - stepContext.noteOutput(windowedElem); - } } private <T> void outputWindowedValue( @@ -380,9 +377,6 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out } outputManager.output(tag, windowedElem); - if (stepContext != null) { - stepContext.noteOutput(tag, windowedElem); - } } } http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/core-java/src/main/java/org/apache/beam/runners/core/StepContext.java ---------------------------------------------------------------------- diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StepContext.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StepContext.java index a414830..fd2575d 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StepContext.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StepContext.java @@ -19,7 +19,6 @@ package org.apache.beam.runners.core; import java.io.IOException; import org.apache.beam.sdk.coders.Coder; -import org.apache.beam.sdk.transforms.DoFn.WindowedContext; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.util.WindowedValue; import org.apache.beam.sdk.values.TupleTag; @@ -40,20 +39,6 @@ public interface StepContext { String getTransformName(); /** - * Hook for subclasses to implement that will be called whenever - * {@link WindowedContext#output} - * is called. - */ - void noteOutput(WindowedValue<?> output); - - /** - * Hook for subclasses to implement that will be called whenever - * {@link WindowedContext#output} - * is called. - */ - void noteOutput(TupleTag<?> tag, WindowedValue<?> output); - - /** * Writes the given {@code PCollectionView} data to a globally accessible location. */ <T, W extends BoundedWindow> void writePCollectionViewData( http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java ---------------------------------------------------------------------- diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java index c394ebd..d999494 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java @@ -42,16 +42,6 @@ public class FlinkNoOpStepContext implements StepContext { } @Override - public void noteOutput(WindowedValue<?> output) { - - } - - @Override - public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) { - - } - - @Override public <T, W extends BoundedWindow> void writePCollectionViewData( TupleTag<?> tag, Iterable<WindowedValue<T>> data, http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java ---------------------------------------------------------------------- diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java index c9f106a..2bb9c20 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java @@ -688,12 +688,6 @@ public class DoFnOperator<InputT, FnOutputT, OutputT> } @Override - public void noteOutput(WindowedValue<?> output) {} - - @Override - public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {} - - @Override public <T, W extends BoundedWindow> void writePCollectionViewData( TupleTag<?> tag, Iterable<WindowedValue<T>> data, http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java ---------------------------------------------------------------------- diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java index 9147422..31e616c 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java @@ -110,12 +110,6 @@ class SparkProcessContext<FnInputT, FnOutputT, OutputT> { } @Override - public void noteOutput(WindowedValue<?> output) { } - - @Override - public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) { } - - @Override public <T, W extends BoundedWindow> void writePCollectionViewData( TupleTag<?> tag, Iterable<WindowedValue<T>> data, http://git-wip-us.apache.org/repos/asf/beam/blob/bed1c53f/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java ---------------------------------------------------------------------- diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java index b206bc7..750c167 100644 --- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java +++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java @@ -42,14 +42,6 @@ public class FakeStepContext implements StepContext { } @Override - public void noteOutput(WindowedValue<?> output) { - } - - @Override - public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) { - } - - @Override public <T, W extends BoundedWindow> void writePCollectionViewData( TupleTag<?> tag, Iterable<WindowedValue<T>> data,
