[BEAM-2140] Don't use StatefulDoFnRunner when running SDF in FlinkRunner

mransley · Feb 23, 2018 · d15979f · d15979f
1 parent 3eecb38
commit d15979f
Showing 1 changed file with 10 additions and 0 deletions.
diff --git a/.../org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java b/.../org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java
@@ -26,6 +26,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import org.apache.beam.runners.core.DoFnRunner;
 import org.apache.beam.runners.core.KeyedWorkItem;
 import org.apache.beam.runners.core.KeyedWorkItems;
 import org.apache.beam.runners.core.OutputAndTimeBoundedSplittableProcessElementInvoker;
@@ -87,6 +88,15 @@ public SplittableDoFnOperator(
  keyCoder);
  }
 
+ @Override
+ protected DoFnRunner<
+ KeyedWorkItem<String, KV<InputT, RestrictionT>>, OutputT> createWrappingDoFnRunner(
+ DoFnRunner<KeyedWorkItem<String, KV<InputT, RestrictionT>>, OutputT> wrappedRunner) {
+ // don't wrap in anything because we don't need state cleanup because ProcessFn does
+ // all that
+ return wrappedRunner;
+ }
+
  @Override
  public void open() throws Exception {
  super.open();