You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by mx...@apache.org on 2019/04/29 13:36:24 UTC
[beam] branch master updated: [BEAM-7112] Timer race with state
cleanup - take two
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new aca9523 [BEAM-7112] Timer race with state cleanup - take two
new 2b111c4 Merge pull request #8399: [BEAM-7112] Prevent race condition between user and cleanup timers
aca9523 is described below
commit aca952327cf04eec84050a46bd322ce1bc68733b
Author: Thomas Weise <th...@apache.org>
AuthorDate: Thu Apr 25 00:07:23 2019 -0700
[BEAM-7112] Timer race with state cleanup - take two
---
.../streaming/ExecutableStageDoFnOperator.java | 92 +++++++------
.../streaming/ExecutableStageDoFnOperatorTest.java | 145 ++++++++++++++++++++-
2 files changed, 184 insertions(+), 53 deletions(-)
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java
index 064b87a..48f9792 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java
@@ -35,6 +35,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
+import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleProgressResponse;
@@ -42,7 +43,6 @@ import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleResponse;
import org.apache.beam.model.fnexecution.v1.BeamFnApi.StateKey.TypeCase;
import org.apache.beam.model.pipeline.v1.RunnerApi;
import org.apache.beam.runners.core.DoFnRunner;
-import org.apache.beam.runners.core.DoFnRunners;
import org.apache.beam.runners.core.LateDataUtils;
import org.apache.beam.runners.core.StateInternals;
import org.apache.beam.runners.core.StateNamespace;
@@ -118,7 +118,6 @@ public class ExecutableStageDoFnOperator<InputT, OutputT> extends DoFnOperator<I
private transient SdkHarnessDoFnRunner<InputT, OutputT> sdkHarnessRunner;
private transient FlinkMetricContainer flinkMetricContainer;
private transient long backupWatermarkHold = Long.MIN_VALUE;
- private transient ArrayDeque<InternalTimer<?, TimerInternals.TimerData>> cleanupTimers;
/** Constructor. */
public ExecutableStageDoFnOperator(
@@ -189,7 +188,6 @@ public class ExecutableStageDoFnOperator<InputT, OutputT> extends DoFnOperator<I
flinkMetricContainer.updateMetrics(stepName, response.getMonitoringInfosList());
}
};
- cleanupTimers = new ArrayDeque<>();
// This will call {@code createWrappingDoFnRunner} which needs the above dependencies.
super.open();
@@ -398,35 +396,8 @@ public class ExecutableStageDoFnOperator<InputT, OutputT> extends DoFnOperator<I
}
}
- @SuppressWarnings("ByteBufferBackingArray")
- private void fireCleanupTimers() {
- while (!cleanupTimers.isEmpty()) {
- InternalTimer<?, TimerInternals.TimerData> timer = cleanupTimers.remove();
- final ByteBuffer encodedKey = (ByteBuffer) timer.getKey();
- try {
- // still need to process as timer, see CleanupTimer
- stateBackendLock.lock();
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "State cleanup for {} {}",
- Arrays.toString(encodedKey.array()),
- timer.getNamespace().getNamespace());
- }
- getKeyedStateBackend().setCurrentKey(encodedKey);
- super.fireTimer(timer);
- } finally {
- stateBackendLock.unlock();
- }
- }
- }
-
@Override
public void fireTimer(InternalTimer<?, TimerInternals.TimerData> timer) {
- if (CleanupTimer.GC_TIMER_ID.equals(timer.getNamespace().getTimerId())) {
- // hold cleanup until bundle is complete
- cleanupTimers.add(timer);
- return;
- }
final ByteBuffer encodedKey = (ByteBuffer) timer.getKey();
// We have to synchronize to ensure the state backend is not concurrently accessed by the state
// requests
@@ -526,9 +497,6 @@ public class ExecutableStageDoFnOperator<InputT, OutputT> extends DoFnOperator<I
// at this point the bundle is finished, allow the watermark to pass
// we are restoring the previous hold in case it was already set for side inputs
setPushedBackWatermark(backupWatermarkHold);
- // fire cleanup timers, they can only execute after the bundle is complete
- // as they remove the state that the timer callback may rely on
- fireCleanupTimers();
super.processWatermark(mark);
} catch (Exception e) {
throw new RuntimeException(
@@ -538,10 +506,6 @@ public class ExecutableStageDoFnOperator<InputT, OutputT> extends DoFnOperator<I
}
}
super.processWatermark(mark);
- // if this was the final watermark, then no callback was scheduled
- if (mark.getTimestamp() >= BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()) {
- fireCleanupTimers();
- }
}
private static class SdkHarnessDoFnRunner<InputT, OutputT>
@@ -741,10 +705,29 @@ public class ExecutableStageDoFnOperator<InputT, OutputT> extends DoFnOperator<I
executableStage.getUserStates().stream()
.map(UserStateReference::localName)
.collect(Collectors.toList());
- StateCleaner stateCleaner = new StateCleaner(userStates, windowCoder, keyedStateInternals);
- return DoFnRunners.defaultStatefulDoFnRunner(
- doFn, sdkHarnessRunner, windowingStrategy, cleanupTimer, stateCleaner);
+ KeyedStateBackend<ByteBuffer> stateBackend = getKeyedStateBackend();
+ StateCleaner stateCleaner =
+ new StateCleaner(userStates, windowCoder, () -> stateBackend.getCurrentKey());
+
+ return new StatefulDoFnRunner<InputT, OutputT, BoundedWindow>(
+ sdkHarnessRunner, windowingStrategy, cleanupTimer, stateCleaner) {
+ @Override
+ public void finishBundle() {
+ // Before cleaning up state, first finish bundle for all underlying DoFnRunners
+ super.finishBundle();
+ // execute cleanup after the bundle is complete
+ if (!stateCleaner.cleanupQueue.isEmpty()) {
+ try {
+ stateBackendLock.lock();
+ stateCleaner.cleanupState(
+ keyedStateInternals, (key) -> stateBackend.setCurrentKey(key));
+ } finally {
+ stateBackendLock.unlock();
+ }
+ }
+ }
+ };
}
static class CleanupTimer<InputT> implements StatefulDoFnRunner.CleanupTimer<InputT> {
@@ -812,21 +795,36 @@ public class ExecutableStageDoFnOperator<InputT, OutputT> extends DoFnOperator<I
private final List<String> userStateNames;
private final Coder windowCoder;
- private final StateInternals stateInternals;
+ private final ArrayDeque<KV<ByteBuffer, BoundedWindow>> cleanupQueue;
+ private final Supplier<ByteBuffer> keyedStateBackend;
- StateCleaner(List<String> userStateNames, Coder windowCoder, StateInternals stateInternals) {
+ StateCleaner(
+ List<String> userStateNames, Coder windowCoder, Supplier<ByteBuffer> keyedStateBackend) {
this.userStateNames = userStateNames;
this.windowCoder = windowCoder;
- this.stateInternals = stateInternals;
+ this.keyedStateBackend = keyedStateBackend;
+ this.cleanupQueue = new ArrayDeque<>();
}
@Override
public void clearForWindow(BoundedWindow window) {
// Executed in the context of onTimer(..) where the correct key will be set
- for (String userState : userStateNames) {
- StateNamespace namespace = StateNamespaces.window(windowCoder, window);
- BagState<?> state = stateInternals.state(namespace, StateTags.bag(userState, null));
- state.clear();
+ cleanupQueue.add(KV.of(keyedStateBackend.get(), window));
+ }
+
+ @SuppressWarnings("ByteBufferBackingArray")
+ void cleanupState(StateInternals stateInternals, Consumer<ByteBuffer> keyContextConsumer) {
+ while (!cleanupQueue.isEmpty()) {
+ KV<ByteBuffer, BoundedWindow> kv = cleanupQueue.remove();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("State cleanup for {} {}", Arrays.toString(kv.getKey().array()), kv.getValue());
+ }
+ keyContextConsumer.accept(kv.getKey());
+ for (String userState : userStateNames) {
+ StateNamespace namespace = StateNamespaces.window(windowCoder, kv.getValue());
+ BagState<?> state = stateInternals.state(namespace, StateTags.bag(userState, null));
+ state.clear();
+ }
}
}
}
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java
index 373a724..dc01ca3 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java
@@ -18,23 +18,30 @@
package org.apache.beam.runners.flink.translation.wrappers.streaming;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.collection.IsIterableContainingInOrder.contains;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import javax.annotation.Nullable;
import org.apache.beam.model.pipeline.v1.RunnerApi.Components;
@@ -45,6 +52,7 @@ import org.apache.beam.runners.core.InMemoryTimerInternals;
import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.StateTags;
import org.apache.beam.runners.core.StatefulDoFnRunner;
+import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.flink.FlinkPipelineOptions;
import org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate;
import org.apache.beam.runners.flink.translation.functions.FlinkExecutableStageContext;
@@ -64,7 +72,10 @@ import org.apache.beam.sdk.fn.data.FnDataReceiver;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.state.BagState;
import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.TupleTag;
@@ -74,6 +85,7 @@ import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableLis
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableMap;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
import org.apache.commons.lang3.SerializationUtils;
+import org.apache.commons.lang3.mutable.MutableObject;
import org.apache.flink.api.common.cache.DistributedCache;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -83,6 +95,8 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.OutputTag;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -365,7 +379,12 @@ public class ExecutableStageDoFnOperatorTest {
new DoFnOperator.MultiOutputOutputManagerFactory(mainOutput, VarIntCoder.of());
VarIntCoder keyCoder = VarIntCoder.of();
ExecutableStageDoFnOperator<Integer, Integer> operator =
- getOperator(mainOutput, Collections.emptyList(), outputManagerFactory, keyCoder);
+ getOperator(
+ mainOutput,
+ Collections.emptyList(),
+ outputManagerFactory,
+ WindowingStrategy.globalDefault(),
+ keyCoder);
KeyedOneInputStreamOperatorTestHarness<Integer, WindowedValue<Integer>, WindowedValue<Integer>>
testHarness =
@@ -391,7 +410,7 @@ public class ExecutableStageDoFnOperatorTest {
}
@Test
- public void testEnsureStateCleanupWithKeyedInputCleanupTimer() throws Exception {
+ public void testEnsureStateCleanupWithKeyedInputCleanupTimer() {
InMemoryTimerInternals inMemoryTimerInternals = new InMemoryTimerInternals();
KeyedStateBackend keyedStateBackend = Mockito.mock(KeyedStateBackend.class);
Lock stateBackendLock = Mockito.mock(Lock.class);
@@ -420,7 +439,7 @@ public class ExecutableStageDoFnOperatorTest {
}
@Test
- public void testEnsureStateCleanupWithKeyedInputStateCleaner() throws Exception {
+ public void testEnsureStateCleanupWithKeyedInputStateCleaner() {
GlobalWindow.Coder windowCoder = GlobalWindow.Coder.INSTANCE;
InMemoryStateInternals<String> stateInternals = InMemoryStateInternals.forKey("key");
List<String> userStateNames = ImmutableList.of("state1", "state2");
@@ -435,14 +454,20 @@ public class ExecutableStageDoFnOperatorTest {
}
ImmutableList<BagState<String>> bagStates = bagStateBuilder.build();
+ MutableObject<ByteBuffer> key =
+ new MutableObject<>(
+ ByteBuffer.wrap(stateInternals.getKey().getBytes(StandardCharsets.UTF_8)));
+
// Test that state is cleaned up correctly
ExecutableStageDoFnOperator.StateCleaner stateCleaner =
- new ExecutableStageDoFnOperator.StateCleaner(userStateNames, windowCoder, stateInternals);
+ new ExecutableStageDoFnOperator.StateCleaner(
+ userStateNames, windowCoder, () -> key.getValue());
for (BagState<String> bagState : bagStates) {
assertThat(Iterables.size(bagState.read()), is(1));
}
stateCleaner.clearForWindow(GlobalWindow.INSTANCE);
+ stateCleaner.cleanupState(stateInternals, (k) -> key.setValue(k));
for (BagState<String> bagState : bagStates) {
assertThat(Iterables.size(bagState.read()), is(0));
@@ -450,6 +475,108 @@ public class ExecutableStageDoFnOperatorTest {
}
@Test
+ public void testEnsureDeferredStateCleanupTimerFiring() throws Exception {
+ TupleTag<Integer> mainOutput = new TupleTag<>("main-output");
+ DoFnOperator.MultiOutputOutputManagerFactory<Integer> outputManagerFactory =
+ new DoFnOperator.MultiOutputOutputManagerFactory(mainOutput, VoidCoder.of());
+ StringUtf8Coder keyCoder = StringUtf8Coder.of();
+
+ WindowingStrategy windowingStrategy =
+ WindowingStrategy.of(FixedWindows.of(Duration.millis(1000)));
+
+ ExecutableStageDoFnOperator<Integer, Integer> operator =
+ getOperator(
+ mainOutput, Collections.emptyList(), outputManagerFactory, windowingStrategy, keyCoder);
+
+ @SuppressWarnings("unchecked")
+ RemoteBundle bundle = Mockito.mock(RemoteBundle.class);
+ when(stageBundleFactory.getBundle(any(), any(), any())).thenReturn(bundle);
+
+ String timerInputId = "timerInput";
+ AtomicBoolean timerInputReceived = new AtomicBoolean();
+ IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(1000));
+ IntervalWindow.IntervalWindowCoder windowCoder = IntervalWindow.IntervalWindowCoder.of();
+ WindowedValue<KV<String, Integer>> one =
+ WindowedValue.of(
+ KV.of("one", 1), window.maxTimestamp(), ImmutableList.of(window), PaneInfo.NO_FIRING);
+
+ FnDataReceiver<WindowedValue<?>> receiver = Mockito.mock(FnDataReceiver.class);
+ FnDataReceiver<WindowedValue<?>> timerReceiver = Mockito.mock(FnDataReceiver.class);
+ doAnswer(
+ (invocation) -> {
+ timerInputReceived.set(true);
+ return null;
+ })
+ .when(timerReceiver)
+ .accept(any());
+
+ when(bundle.getInputReceivers())
+ .thenReturn(ImmutableMap.of("input", receiver, timerInputId, timerReceiver));
+
+ KeyedOneInputStreamOperatorTestHarness<
+ String, WindowedValue<KV<String, Integer>>, WindowedValue<KV<String, Integer>>>
+ testHarness =
+ new KeyedOneInputStreamOperatorTestHarness(
+ operator, val -> val, new CoderTypeInformation<>(keyCoder));
+
+ testHarness.open();
+
+ Lock stateBackendLock = (Lock) Whitebox.getInternalState(operator, "stateBackendLock");
+ stateBackendLock.lock();
+
+ KeyedStateBackend<ByteBuffer> keyedStateBackend = operator.getKeyedStateBackend();
+ ByteBuffer key =
+ FlinkKeyUtils.encodeKey(one.getValue().getKey(), keyCoder, Coder.Context.NESTED);
+ keyedStateBackend.setCurrentKey(key);
+
+ DoFnOperator.FlinkTimerInternals timerInternals =
+ (DoFnOperator.FlinkTimerInternals) Whitebox.getInternalState(operator, "timerInternals");
+
+ Object doFnRunner = Whitebox.getInternalState(operator, "doFnRunner");
+ Object delegate = Whitebox.getInternalState(doFnRunner, "delegate");
+ Object stateCleaner = Whitebox.getInternalState(delegate, "stateCleaner");
+ Collection<?> cleanupTimers =
+ (Collection) Whitebox.getInternalState(stateCleaner, "cleanupQueue");
+
+ // user timer that fires after the end of the window and after state cleanup
+ TimerInternals.TimerData userTimer =
+ TimerInternals.TimerData.of(
+ timerInputId,
+ StateNamespaces.window(windowCoder, window),
+ window.maxTimestamp().plus(1),
+ TimeDomain.EVENT_TIME);
+ timerInternals.setTimer(userTimer);
+
+ // start of bundle
+ testHarness.processElement(new StreamRecord<>(one));
+ verify(receiver).accept(one);
+
+ // move watermark past cleanup and user timer while bundle in progress
+ operator.processWatermark(new Watermark(window.maxTimestamp().plus(2).getMillis()));
+
+ // due to watermark hold the timers won't fire at this point
+ assertFalse("Watermark must be held back until bundle is complete.", timerInputReceived.get());
+ assertThat(cleanupTimers, hasSize(0));
+
+ // upon finish bundle, watermark advances and timers can fire
+ // Note that this will finish the current bundle, but will also start a new one
+ // when timers fire as part of advancing the watermark
+ operator.invokeFinishBundle();
+
+ // the user timer was scheduled to fire after cleanup, but executes first
+ assertTrue("Timer should have been triggered.", timerInputReceived.get());
+ // cleanup will be executed after the bundle is complete
+ assertThat(cleanupTimers, hasSize(1));
+
+ verifyNoMoreInteractions(receiver);
+
+ operator.invokeFinishBundle();
+ assertThat(cleanupTimers, hasSize(0));
+
+ testHarness.close();
+ }
+
+ @Test
public void testSerialization() {
WindowedValue.ValueOnlyWindowedValueCoder<Integer> coder =
WindowedValue.getValueOnlyCoder(VarIntCoder.of());
@@ -514,13 +641,19 @@ public class ExecutableStageDoFnOperatorTest {
TupleTag<Integer> mainOutput,
List<TupleTag<?>> additionalOutputs,
DoFnOperator.MultiOutputOutputManagerFactory<Integer> outputManagerFactory) {
- return getOperator(mainOutput, additionalOutputs, outputManagerFactory, null);
+ return getOperator(
+ mainOutput,
+ additionalOutputs,
+ outputManagerFactory,
+ WindowingStrategy.globalDefault(),
+ null);
}
private ExecutableStageDoFnOperator<Integer, Integer> getOperator(
TupleTag<Integer> mainOutput,
List<TupleTag<?>> additionalOutputs,
DoFnOperator.MultiOutputOutputManagerFactory<Integer> outputManagerFactory,
+ WindowingStrategy windowingStrategy,
@Nullable Coder keyCoder) {
FlinkExecutableStageContext.Factory contextFactory =
@@ -544,7 +677,7 @@ public class ExecutableStageDoFnOperatorTest {
jobInfo,
contextFactory,
createOutputMap(mainOutput, additionalOutputs),
- WindowingStrategy.globalDefault(),
+ windowingStrategy,
keyCoder,
null);