beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [1/2] beam git commit: [BEAM-1517] Garbage collect user state in Flink Runner
Date Wed, 01 Mar 2017 17:01:28 GMT
Repository: beam
Updated Branches:
  refs/heads/master d66029caf -> 79b1395c2


[BEAM-1517] Garbage collect user state in Flink Runner


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/6e9f7ad4
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/6e9f7ad4
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/6e9f7ad4

Branch: refs/heads/master
Commit: 6e9f7ad46bcb341d37ac5d7804465dd82ef3b56e
Parents: d66029c
Author: JingsongLi <lzljs3620320@aliyun.com>
Authored: Tue Feb 28 16:08:38 2017 +0800
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Wed Mar 1 17:31:36 2017 +0100

----------------------------------------------------------------------
 .../apache/beam/runners/core/DoFnRunners.java   |  39 +++
 .../beam/runners/core/StatefulDoFnRunner.java   | 233 +++++++++++++++++
 .../runners/core/StatefulDoFnRunnerTest.java    | 255 +++++++++++++++++++
 .../wrappers/streaming/DoFnOperator.java        |   4 +
 4 files changed, 531 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/6e9f7ad4/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
index f3972ae..9455eea 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
@@ -19,9 +19,15 @@ package org.apache.beam.runners.core;
 
 import java.util.List;
 import org.apache.beam.runners.core.ExecutionContext.StepContext;
+import org.apache.beam.runners.core.StatefulDoFnRunner.CleanupTimer;
+import org.apache.beam.runners.core.StatefulDoFnRunner.StateCleaner;
+import org.apache.beam.runners.core.StatefulDoFnRunner.StateInternalsStateCleaner;
+import org.apache.beam.runners.core.StatefulDoFnRunner.TimeInternalsCleanupTimer;
+import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.transforms.Aggregator;
 import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.SideInputReader;
 import org.apache.beam.sdk.util.WindowedValue;
@@ -116,4 +122,37 @@ public class DoFnRunners {
         stepContext.timerInternals(),
         droppedDueToLatenessAggregator);
   }
+
+  /**
+   * Returns an implementation of {@link DoFnRunner} that handles
+   * late data dropping and garbage collection for stateful {@link DoFn DoFns}.
+   *
+   * <p>It registers a timer by TimeInternals, and clean all states by StateInternals.
+   */
+  public static <InputT, OutputT, W extends BoundedWindow>
+      DoFnRunner<InputT, OutputT> defaultStatefulDoFnRunner(
+          DoFn<InputT, OutputT> fn,
+          DoFnRunner<InputT, OutputT> doFnRunner,
+          StepContext stepContext,
+          AggregatorFactory aggregatorFactory,
+          WindowingStrategy<?, ?> windowingStrategy) {
+    Aggregator<Long, Long> droppedDueToLateness = aggregatorFactory.createAggregatorForDoFn(
+        fn.getClass(), stepContext, StatefulDoFnRunner.DROPPED_DUE_TO_LATENESS_COUNTER,
+        Sum.ofLongs());
+
+    CleanupTimer cleanupTimer =
+        new TimeInternalsCleanupTimer(stepContext.timerInternals(), windowingStrategy);
+
+    Coder<W> windowCoder = (Coder<W>) windowingStrategy.getWindowFn().windowCoder();
+    StateCleaner<W> stateCleaner =
+        new StateInternalsStateCleaner<>(fn, stepContext.stateInternals(), windowCoder);
+
+    return new StatefulDoFnRunner<>(
+        doFnRunner,
+        windowingStrategy,
+        cleanupTimer,
+        stateCleaner,
+        droppedDueToLateness);
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/6e9f7ad4/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
new file mode 100644
index 0000000..154d8bc
--- /dev/null
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core;
+
+import java.util.Map;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.Aggregator;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.NonMergingWindowFn;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.util.TimeDomain;
+import org.apache.beam.sdk.util.WindowTracing;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.util.WindowingStrategy;
+import org.apache.beam.sdk.util.state.State;
+import org.apache.beam.sdk.util.state.StateSpec;
+import org.joda.time.Instant;
+
+/**
+ * A customized {@link DoFnRunner} that handles late data dropping and garbage collection
for
+ * stateful {@link DoFn DoFns}. It registers a GC timer in {@link #processElement(WindowedValue)}
+ * and does cleanup in {@link #onTimer(String, BoundedWindow, Instant, TimeDomain)}
+ *
+ * @param <InputT> the type of the {@link DoFn} (main) input elements
+ * @param <OutputT> the type of the {@link DoFn} (main) output elements
+ */
+public class StatefulDoFnRunner<InputT, OutputT, W extends BoundedWindow>
+    implements DoFnRunner<InputT, OutputT> {
+
+  public static final String GC_TIMER_ID = "__StatefulParDoGcTimerId";
+  public static final String DROPPED_DUE_TO_LATENESS_COUNTER = "StatefulParDoDropped";
+
+  private final DoFnRunner<InputT, OutputT> doFnRunner;
+  private final WindowingStrategy<?, ?> windowingStrategy;
+  private final Aggregator<Long, Long> droppedDueToLateness;
+  private final CleanupTimer cleanupTimer;
+  private final StateCleaner stateCleaner;
+
+  public StatefulDoFnRunner(
+      DoFnRunner<InputT, OutputT> doFnRunner,
+      WindowingStrategy<?, ?> windowingStrategy,
+      CleanupTimer cleanupTimer,
+      StateCleaner<W> stateCleaner,
+      Aggregator<Long, Long> droppedDueToLateness) {
+    this.doFnRunner = doFnRunner;
+    this.windowingStrategy = windowingStrategy;
+    this.cleanupTimer = cleanupTimer;
+    this.stateCleaner = stateCleaner;
+    WindowFn<?, ?> windowFn = windowingStrategy.getWindowFn();
+    rejectMergingWindowFn(windowFn);
+    this.droppedDueToLateness = droppedDueToLateness;
+  }
+
+  private void rejectMergingWindowFn(WindowFn<?, ?> windowFn) {
+    if (!(windowFn instanceof NonMergingWindowFn)) {
+      throw new UnsupportedOperationException(
+          "MergingWindowFn is not supported for stateful DoFns, WindowFn is: "
+              + windowFn);
+    }
+  }
+
+  @Override
+  public void startBundle() {
+    doFnRunner.startBundle();
+  }
+
+  @Override
+  public void processElement(WindowedValue<InputT> compressedElem) {
+
+    // StatefulDoFnRunner always observes windows, so we need to explode
+    for (WindowedValue<InputT> value : compressedElem.explodeWindows()) {
+
+      BoundedWindow window = value.getWindows().iterator().next();
+
+      if (!dropLateData(window)) {
+        cleanupTimer.setForWindow(window);
+        doFnRunner.processElement(value);
+      }
+    }
+  }
+
+  private boolean dropLateData(BoundedWindow window) {
+    Instant gcTime = window.maxTimestamp().plus(windowingStrategy.getAllowedLateness());
+    Instant inputWM = cleanupTimer.currentInputWatermarkTime();
+    if (gcTime.isBefore(inputWM)) {
+      // The element is too late for this window.
+      droppedDueToLateness.addValue(1L);
+      WindowTracing.debug(
+          "StatefulDoFnRunner.processElement/onTimer: Dropping element for window:{} "
+              + "since too far behind inputWatermark:{}", window, inputWM);
+      return true;
+    } else {
+      return false;
+    }
+  }
+
+  @Override
+  public void onTimer(
+      String timerId, BoundedWindow window, Instant timestamp, TimeDomain timeDomain) {
+    boolean isEventTimer = timeDomain.equals(TimeDomain.EVENT_TIME);
+    Instant gcTime = window.maxTimestamp().plus(windowingStrategy.getAllowedLateness());
+    if (isEventTimer && GC_TIMER_ID.equals(timerId) && gcTime.equals(timestamp))
{
+      stateCleaner.clearForWindow(window);
+      // There should invoke the onWindowExpiration of DoFn
+    } else {
+      if (isEventTimer || !dropLateData(window)) {
+        doFnRunner.onTimer(timerId, window, timestamp, timeDomain);
+      }
+    }
+  }
+
+  @Override
+  public void finishBundle() {
+    doFnRunner.finishBundle();
+  }
+
+  /**
+   * A cleaner for deciding when to clean state of window.
+   *
+   * <p>A runner might either (a) already know that it always has a timer set
+   * for the expiration time or (b) not need a timer at all because it is
+   * a batch runner that discards state when it is done.
+   */
+  public interface CleanupTimer {
+
+    /**
+     * Return the current, local input watermark timestamp for this computation
+     * in the {@link TimeDomain#EVENT_TIME} time domain.
+     */
+    Instant currentInputWatermarkTime();
+
+    /**
+     * Set the garbage collect time of the window to timer.
+     */
+    void setForWindow(BoundedWindow window);
+  }
+
+  /**
+   * A cleaner to clean all states of the window.
+   */
+  public interface StateCleaner<W extends BoundedWindow> {
+
+    void clearForWindow(W window);
+  }
+
+  /**
+   * A {@link CleanupTimer} implemented by TimerInternals.
+   */
+  public static class TimeInternalsCleanupTimer implements CleanupTimer {
+
+    private final TimerInternals timerInternals;
+    private final WindowingStrategy<?, ?> windowingStrategy;
+    private final Coder<BoundedWindow> windowCoder;
+
+    public TimeInternalsCleanupTimer(
+        TimerInternals timerInternals,
+        WindowingStrategy<?, ?> windowingStrategy) {
+      this.windowingStrategy = windowingStrategy;
+      WindowFn<?, ?> windowFn = windowingStrategy.getWindowFn();
+      windowCoder = (Coder<BoundedWindow>) windowFn.windowCoder();
+      this.timerInternals = timerInternals;
+    }
+
+    @Override
+    public Instant currentInputWatermarkTime() {
+      return timerInternals.currentInputWatermarkTime();
+    }
+
+    @Override
+    public void setForWindow(BoundedWindow window) {
+      Instant gcTime = window.maxTimestamp().plus(windowingStrategy.getAllowedLateness());
+      timerInternals.setTimer(StateNamespaces.window(windowCoder, window),
+          GC_TIMER_ID, gcTime, TimeDomain.EVENT_TIME);
+    }
+
+  }
+
+  /**
+   * A {@link StateCleaner} implemented by StateInternals.
+   */
+  public static class StateInternalsStateCleaner<W extends BoundedWindow>
+      implements StateCleaner<W> {
+
+    private final DoFn<?, ?> fn;
+    private final DoFnSignature signature;
+    private final StateInternals<?> stateInternals;
+    private final Coder<W> windowCoder;
+
+    public StateInternalsStateCleaner(
+        DoFn<?, ?> fn,
+        StateInternals<?> stateInternals,
+        Coder<W> windowCoder) {
+      this.fn = fn;
+      this.signature = DoFnSignatures.getSignature(fn.getClass());
+      this.stateInternals = stateInternals;
+      this.windowCoder = windowCoder;
+    }
+
+    @Override
+    public void clearForWindow(W window) {
+      for (Map.Entry<String, DoFnSignature.StateDeclaration> entry :
+          signature.stateDeclarations().entrySet()) {
+        try {
+          StateSpec<?, ?> spec = (StateSpec<?, ?>) entry.getValue().field().get(fn);
+          State state = stateInternals.state(StateNamespaces.window(windowCoder, window),
+              StateTags.tagForSpec(entry.getKey(), (StateSpec) spec));
+          state.clear();
+        } catch (IllegalAccessException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/6e9f7ad4/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java
new file mode 100644
index 0000000..54ac77e
--- /dev/null
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/StatefulDoFnRunnerTest.java
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.when;
+
+import com.google.common.base.MoreObjects;
+import java.util.Collections;
+import org.apache.beam.runners.core.BaseExecutionContext.StepContext;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.transforms.Aggregator;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Sum;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.NullSideInputReader;
+import org.apache.beam.sdk.util.TimeDomain;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.util.WindowingStrategy;
+import org.apache.beam.sdk.util.state.StateSpec;
+import org.apache.beam.sdk.util.state.StateSpecs;
+import org.apache.beam.sdk.util.state.ValueState;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.TupleTag;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/** Tests for {@link StatefulDoFnRunnerTest}. */
+@RunWith(JUnit4.class)
+public class StatefulDoFnRunnerTest {
+
+  private static final long WINDOW_SIZE = 10;
+  private static final long ALLOWED_LATENESS = 1;
+
+  private static final WindowingStrategy<?, ?> WINDOWING_STRATEGY =
+      WindowingStrategy
+          .of(FixedWindows.of(Duration.millis(WINDOW_SIZE)))
+          .withAllowedLateness(Duration.millis(ALLOWED_LATENESS));
+
+  private static final IntervalWindow WINDOW_1 =
+      new IntervalWindow(new Instant(0), new Instant(10));
+
+  private static final IntervalWindow WINDOW_2 =
+      new IntervalWindow(new Instant(10), new Instant(20));
+
+  @Mock StepContext mockStepContext;
+
+  private InMemoryLongSumAggregator droppedDueToLateness;
+  private AggregatorFactory aggregatorFactory;
+  private InMemoryStateInternals<String> stateInternals;
+  private InMemoryTimerInternals timerInternals;
+
+  private static StateNamespace windowNamespace(IntervalWindow window) {
+    return StateNamespaces.<IntervalWindow>window(
+        (Coder) WINDOWING_STRATEGY.getWindowFn().windowCoder(), window);
+  }
+
+  @Before
+  public void setup() {
+    MockitoAnnotations.initMocks(this);
+    when(mockStepContext.timerInternals()).thenReturn(timerInternals);
+    droppedDueToLateness = new InMemoryLongSumAggregator("droppedDueToLateness");
+
+    aggregatorFactory = new AggregatorFactory() {
+      @Override
+      public <InputT, AccumT, OutputT> Aggregator<InputT, OutputT> createAggregatorForDoFn(
+          Class<?> fnClass, ExecutionContext.StepContext stepContext, String aggregatorName,
+          Combine.CombineFn<InputT, AccumT, OutputT> combine) {
+        return (Aggregator<InputT, OutputT>) droppedDueToLateness;
+      }
+    };
+
+    stateInternals = new InMemoryStateInternals<>("hello");
+    timerInternals = new InMemoryTimerInternals();
+
+    when(mockStepContext.stateInternals()).thenReturn((StateInternals) stateInternals);
+    when(mockStepContext.timerInternals()).thenReturn(timerInternals);
+  }
+
+  @Test
+  public void testLateDropping() throws Exception {
+
+    timerInternals.advanceInputWatermark(new Instant(BoundedWindow.TIMESTAMP_MAX_VALUE));
+    timerInternals.advanceOutputWatermark(new Instant(BoundedWindow.TIMESTAMP_MAX_VALUE));
+
+    DoFn<KV<String, Integer>, Integer> fn = new MyDoFn();
+
+    DoFnRunner<KV<String, Integer>, Integer> runner = DoFnRunners.defaultStatefulDoFnRunner(
+        fn, getDoFnRunner(fn), mockStepContext, aggregatorFactory, WINDOWING_STRATEGY);
+
+    runner.startBundle();
+
+    IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(0L + WINDOW_SIZE));
+    Instant timestamp = new Instant(0);
+
+    runner.processElement(
+        WindowedValue.of(KV.of("hello", 1), timestamp, window, PaneInfo.NO_FIRING));
+    assertEquals(1L, droppedDueToLateness.sum);
+
+    runner.onTimer("processTimer", window, timestamp, TimeDomain.PROCESSING_TIME);
+    assertEquals(2L, droppedDueToLateness.sum);
+
+    runner.onTimer("synchronizedProcessTimer", window, timestamp,
+        TimeDomain.SYNCHRONIZED_PROCESSING_TIME);
+    assertEquals(3L, droppedDueToLateness.sum);
+
+    runner.finishBundle();
+  }
+
+  @Test
+  public void testGarbageCollect() throws Exception {
+    timerInternals.advanceInputWatermark(new Instant(1L));
+
+    MyDoFn fn = new MyDoFn();
+    StateTag<Object, ValueState<Integer>> stateTag = StateTags.tagForSpec(fn.stateId,
fn.intState);
+
+    DoFnRunner<KV<String, Integer>, Integer> runner = DoFnRunners.defaultStatefulDoFnRunner(
+        fn, getDoFnRunner(fn), mockStepContext, aggregatorFactory, WINDOWING_STRATEGY);
+
+    Instant elementTime = new Instant(1);
+
+    // first element, key is hello, WINDOW_1
+    runner.processElement(
+        WindowedValue.of(KV.of("hello", 1), elementTime, WINDOW_1, PaneInfo.NO_FIRING));
+
+    assertEquals(
+        1, (int) stateInternals.state(windowNamespace(WINDOW_1), stateTag).read());
+
+    // second element, key is hello, WINDOW_2
+    runner.processElement(
+        WindowedValue.of(
+            KV.of("hello", 1), elementTime.plus(WINDOW_SIZE), WINDOW_2, PaneInfo.NO_FIRING));
+
+    runner.processElement(
+        WindowedValue.of(
+            KV.of("hello", 1), elementTime.plus(WINDOW_SIZE), WINDOW_2, PaneInfo.NO_FIRING));
+
+    assertEquals(
+        2, (int) stateInternals.state(windowNamespace(WINDOW_2), stateTag).read());
+
+    // advance watermark past end of WINDOW_1 + allowed lateness
+    advanceInputWatermark(
+        timerInternals, WINDOW_1.maxTimestamp().plus(ALLOWED_LATENESS + 1), runner);
+    assertTrue(
+        stateInternals.isEmptyForTesting(
+            stateInternals.state(windowNamespace(WINDOW_1), stateTag)));
+
+    assertEquals(
+        2, (int) stateInternals.state(windowNamespace(WINDOW_2), stateTag).read());
+
+    // advance watermark past end of WINDOW_2 + allowed lateness
+    advanceInputWatermark(
+        timerInternals, WINDOW_2.maxTimestamp().plus(ALLOWED_LATENESS + 1), runner);
+    assertTrue(
+        stateInternals.isEmptyForTesting(
+            stateInternals.state(windowNamespace(WINDOW_2), stateTag)));
+  }
+
+  private DoFnRunner<KV<String, Integer>, Integer> getDoFnRunner(
+      DoFn<KV<String, Integer>, Integer> fn) {
+    return new SimpleDoFnRunner<>(
+        null,
+        fn,
+        NullSideInputReader.empty(),
+        null,
+        null,
+        Collections.<TupleTag<?>>emptyList(),
+        mockStepContext,
+        null,
+        WINDOWING_STRATEGY);
+  }
+
+  private static void advanceInputWatermark(
+      InMemoryTimerInternals timerInternals,
+      Instant newInputWatermark,
+      DoFnRunner<?, ?> toTrigger) throws Exception {
+    timerInternals.advanceInputWatermark(newInputWatermark);
+    TimerInternals.TimerData timer;
+    while ((timer = timerInternals.removeNextEventTimer()) != null) {
+      StateNamespace namespace = timer.getNamespace();
+      checkArgument(namespace instanceof StateNamespaces.WindowNamespace);
+      BoundedWindow window = ((StateNamespaces.WindowNamespace) namespace).getWindow();
+      toTrigger.onTimer(timer.getTimerId(), window, timer.getTimestamp(), timer.getDomain());
+    }
+  }
+
+  private static class MyDoFn extends DoFn<KV<String, Integer>, Integer> {
+
+    public final String stateId = "foo";
+
+    @StateId(stateId)
+    public final StateSpec<Object, ValueState<Integer>> intState =
+        StateSpecs.value(VarIntCoder.of());
+
+    @ProcessElement
+    public void processElement(
+        ProcessContext c, @StateId(stateId) ValueState<Integer> state) {
+      Integer currentValue = MoreObjects.firstNonNull(state.read(), 0);
+      state.write(currentValue + 1);
+    }
+  };
+
+  private static class InMemoryLongSumAggregator implements Aggregator<Long, Long>
{
+    private final String name;
+    private long sum = 0;
+
+    public InMemoryLongSumAggregator(String name) {
+      this.name = name;
+    }
+
+    @Override
+    public void addValue(Long value) {
+      sum += value;
+    }
+
+    @Override
+    public String getName() {
+      return name;
+    }
+
+    @Override
+    public Combine.CombineFn<Long, ?, Long> getCombineFn() {
+      return Sum.ofLongs();
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/6e9f7ad4/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
index 29b6fbc..c4622ba 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
@@ -291,6 +291,10 @@ public class DoFnOperator<InputT, FnOutputT, OutputT>
           stepContext,
           windowingStrategy,
           ((GroupAlsoByWindowViaWindowSetNewDoFn) doFn).getDroppedDueToLatenessAggregator());
+    } else if (keyCoder != null) {
+      // It is a stateful DoFn
+      doFnRunner = DoFnRunners.defaultStatefulDoFnRunner(
+          doFn, doFnRunner, stepContext, aggregatorFactory, windowingStrategy);
     }
 
     pushbackDoFnRunner =


Mime
View raw message