This is an automated email from the ASF dual-hosted git repository.
scwhittle 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 3cca8fa3dac [Dataflow Streaming] Access state/timerinternals via
StepContext (#38991)
3cca8fa3dac is described below
commit 3cca8fa3dac2679f42a791884e0d9aa00431bccd
Author: Arun Pandian <[email protected]>
AuthorDate: Mon Jun 29 01:08:11 2026 -0700
[Dataflow Streaming] Access state/timerinternals via StepContext (#38991)
---
..._ValidatesRunner_Dataflow_Streaming_Engine.json | 2 +-
.../org/apache/beam/runners/core/DoFnRunners.java | 4 +-
.../runners/core/LateDataDroppingDoFnRunner.java | 17 +++---
.../core/LateDataDroppingDoFnRunnerTest.java | 4 +-
.../apache/beam/runners/core/ReduceFnTester.java | 20 +++++++-
.../wrappers/streaming/WindowDoFnOperator.java | 2 +-
.../dataflow/worker/GroupAlsoByWindowsParDoFn.java | 2 +-
.../worker/StreamingModeExecutionContext.java | 20 ++++++++
.../dataflow/worker/WindmillTimerInternals.java | 23 +++++++++
.../windmill/state/WindmillStateInternals.java | 17 ++++++
.../worker/StreamingGroupAlsoByWindowFnsTest.java | 3 +-
.../worker/StreamingModeExecutionContextTest.java | 60 ++++++++++++++++++++++
12 files changed, 156 insertions(+), 18 deletions(-)
diff --git
a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming_Engine.json
b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming_Engine.json
index e623d3373a9..50d17c108f2 100644
---
a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming_Engine.json
+++
b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming_Engine.json
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run!",
- "modification": 1,
+ "modification": 2,
}
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 2cab758d36e..0608ea24b04 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
@@ -85,9 +85,9 @@ public class DoFnRunners {
public static <K, InputT, OutputT, W extends BoundedWindow>
DoFnRunner<KeyedWorkItem<K, InputT>, KV<K, OutputT>>
lateDataDroppingRunner(
DoFnRunner<KeyedWorkItem<K, InputT>, KV<K, OutputT>> wrappedRunner,
- TimerInternals timerInternals,
+ StepContext stepContext,
WindowingStrategy<?, W> windowingStrategy) {
- return new LateDataDroppingDoFnRunner<>(wrappedRunner, windowingStrategy,
timerInternals);
+ return new LateDataDroppingDoFnRunner<>(wrappedRunner, windowingStrategy,
stepContext);
}
/**
diff --git
a/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
index dfab198f893..1d5908c2497 100644
---
a/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
+++
b/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
@@ -58,9 +58,9 @@ public class LateDataDroppingDoFnRunner<K, InputT, OutputT, W
extends BoundedWin
public LateDataDroppingDoFnRunner(
DoFnRunner<KeyedWorkItem<K, InputT>, KV<K, OutputT>> doFnRunner,
WindowingStrategy<?, ?> windowingStrategy,
- TimerInternals timerInternals) {
+ StepContext stepContext) {
this.doFnRunner = doFnRunner;
- lateDataFilter = new LateDataFilter(windowingStrategy, timerInternals);
+ lateDataFilter = new LateDataFilter(windowingStrategy, stepContext);
}
@Override
@@ -116,13 +116,12 @@ public class LateDataDroppingDoFnRunner<K, InputT,
OutputT, W extends BoundedWin
@VisibleForTesting
static class LateDataFilter {
private final WindowingStrategy<?, ?> windowingStrategy;
- private final TimerInternals timerInternals;
+ private final StepContext stepContext;
private final Counter droppedDueToLateness;
- public LateDataFilter(
- WindowingStrategy<?, ?> windowingStrategy, TimerInternals
timerInternals) {
+ public LateDataFilter(WindowingStrategy<?, ?> windowingStrategy,
StepContext stepContext) {
this.windowingStrategy = windowingStrategy;
- this.timerInternals = timerInternals;
+ this.stepContext = stepContext;
this.droppedDueToLateness =
Metrics.counter(LateDataDroppingDoFnRunner.class,
DROPPED_DUE_TO_LATENESS);
}
@@ -146,8 +145,8 @@ public class LateDataDroppingDoFnRunner<K, InputT, OutputT,
W extends BoundedWin
element.getTimestamp(),
key,
window,
- timerInternals.currentInputWatermarkTime(),
- timerInternals.currentOutputWatermarkTime());
+ stepContext.timerInternals().currentInputWatermarkTime(),
+ stepContext.timerInternals().currentOutputWatermarkTime());
} else {
nonLateElements.add(
WindowedValues.of(
@@ -173,7 +172,7 @@ public class LateDataDroppingDoFnRunner<K, InputT, OutputT,
W extends BoundedWin
* @return True if element can be dropped.
*/
private boolean canDropDueToExpiredWindow(BoundedWindow window) {
- Instant inputWM = timerInternals.currentInputWatermarkTime();
+ Instant inputWM =
stepContext.timerInternals().currentInputWatermarkTime();
return LateDataUtils.garbageCollectionTime(window,
windowingStrategy).isBefore(inputWM);
}
}
diff --git
a/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunnerTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunnerTest.java
index 5a6257ba0a1..3ce098ee05e 100644
---
a/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunnerTest.java
+++
b/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunnerTest.java
@@ -49,6 +49,7 @@ public class LateDataDroppingDoFnRunnerTest {
private static final FixedWindows WINDOW_FN =
FixedWindows.of(Duration.millis(10));
@Mock private TimerInternals mockTimerInternals;
+ @Mock private StepContext mockStepContext;
@Before
public void setUp() {
@@ -60,9 +61,10 @@ public class LateDataDroppingDoFnRunnerTest {
MetricsContainerImpl container = new MetricsContainerImpl("any");
MetricsEnvironment.setCurrentContainer(container);
when(mockTimerInternals.currentInputWatermarkTime()).thenReturn(new
Instant(15L));
+ when(mockStepContext.timerInternals()).thenReturn(mockTimerInternals);
LateDataFilter lateDataFilter =
- new LateDataFilter(WindowingStrategy.of(WINDOW_FN),
mockTimerInternals);
+ new LateDataFilter(WindowingStrategy.of(WINDOW_FN), mockStepContext);
Iterable<WindowedValue<Integer>> actual =
lateDataFilter.filter(
diff --git
a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
index fe2c0a25346..6ff3b99b860 100644
---
a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
+++
b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
@@ -47,6 +47,7 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.state.TimeDomain;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
import org.apache.beam.sdk.transforms.CombineWithContext.CombineFnWithContext;
+import org.apache.beam.sdk.transforms.DoFn.BundleFinalizer;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
@@ -94,6 +95,23 @@ public class ReduceFnTester<InputT, OutputT, W extends
BoundedWindow> {
private final TestInMemoryStateInternals<String> stateInternals =
new TestInMemoryStateInternals<>(KEY);
private final InMemoryTimerInternals timerInternals = new
InMemoryTimerInternals();
+ private final StepContext stepContext =
+ new StepContext() {
+ @Override
+ public StateInternals stateInternals() {
+ return stateInternals;
+ }
+
+ @Override
+ public TimerInternals timerInternals() {
+ return timerInternals;
+ }
+
+ @Override
+ public BundleFinalizer bundleFinalizer() {
+ throw new UnsupportedOperationException();
+ }
+ };
private final WindowFn<Object, W> windowFn;
private final TestWindowedValueReceiver testOutputter;
@@ -577,7 +595,7 @@ public class ReduceFnTester<InputT, OutputT, W extends
BoundedWindow> {
ReduceFnRunner<String, InputT, OutputT, W> runner = createRunner();
runner.processElements(
- new LateDataDroppingDoFnRunner.LateDataFilter(objectStrategy,
timerInternals)
+ new LateDataDroppingDoFnRunner.LateDataFilter(objectStrategy,
stepContext)
.filter(KEY, inputs));
// Persist after each bundle.
diff --git
a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
index fad97e2d7df..5043ed9e530 100644
---
a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
+++
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
@@ -118,7 +118,7 @@ public class WindowDoFnOperator<K, InputT, OutputT>
// for some K, V
return DoFnRunners.lateDataDroppingRunner(
- (DoFnRunner) doFnRunner, timerInternals, windowingStrategy);
+ (DoFnRunner) doFnRunner, stepContext, windowingStrategy);
}
@Override
diff --git
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupAlsoByWindowsParDoFn.java
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupAlsoByWindowsParDoFn.java
index e204a78a7d2..6c92b7f10ea 100644
---
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupAlsoByWindowsParDoFn.java
+++
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupAlsoByWindowsParDoFn.java
@@ -216,7 +216,7 @@ public class GroupAlsoByWindowsParDoFn<InputT, K, V, W
extends BoundedWindow> im
}
return (DoFnRunner<InputT, KV<K, Iterable<V>>>)
DoFnRunners.<K, V, Iterable<V>, W>lateDataDroppingRunner(
- streamingGABWRunner, stepContext.timerInternals(),
windowingStrategy);
+ streamingGABWRunner, stepContext, windowingStrategy);
} else {
if (hasStreamingSideInput) {
return new StreamingSideInputDoFnRunner<>(
diff --git
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java
index d8886474564..66df2233394 100644
---
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java
+++
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java
@@ -891,6 +891,26 @@ public class StreamingModeExecutionContext extends
DataflowExecutionContext<Step
checkStateNotNull(systemTimerInternals).persistTo(builder);
checkStateNotNull(userTimerInternals).persistTo(builder);
}
+ poisonStateAndTimerInternals();
+ }
+
+ /**
+ * Poisons the state and timer internals to prevent any subsequent (stale)
usage.
+ *
+ * <p>This ensures that if these key-specific internals are incorrectly
cached and used after
+ * the key's execution context has finished, it will fail fast with a
clear error rather than
+ * silently corrupting state.
+ */
+ private void poisonStateAndTimerInternals() {
+ if (stateInternals != null) {
+ stateInternals.poison();
+ }
+ if (systemTimerInternals != null) {
+ systemTimerInternals.poison();
+ }
+ if (userTimerInternals != null) {
+ userTimerInternals.poison();
+ }
}
@Override
diff --git
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java
index 07ce62d5933..9b8757769db 100644
---
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java
+++
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java
@@ -64,6 +64,19 @@ class WindmillTimerInternals implements TimerInternals {
private final Consumer<TimerData> onTimerModified;
private final WindmillTagEncoding windmillTagEncoding;
+ private boolean poisoned = false;
+
+ public void poison() {
+ this.poisoned = true;
+ }
+
+ private void checkNotPoisoned() {
+ if (poisoned) {
+ throw new IllegalStateException(
+ "WindmillTimerInternals is poisoned and cannot be used after
flushState().");
+ }
+ }
+
public WindmillTimerInternals(
String stateFamily, // unique identifies a step
WindmillTimerType type,
@@ -80,12 +93,14 @@ class WindmillTimerInternals implements TimerInternals {
}
public WindmillTimerInternals withType(WindmillTimerType type) {
+ checkNotPoisoned();
return new WindmillTimerInternals(
stateFamily, type, processingTime, watermarks, windmillTagEncoding,
onTimerModified);
}
@Override
public void setTimer(TimerData timerKey) {
+ checkNotPoisoned();
String timerDataKey = getTimerDataKey(timerKey.getTimerId(),
timerKey.getTimerFamilyId());
timerMap.put(
new SimpleEntry<>(timerDataKey, timerKey.getNamespace()),
@@ -101,6 +116,7 @@ class WindmillTimerInternals implements TimerInternals {
Instant timestamp,
Instant outputTimestamp,
TimeDomain timeDomain) {
+ checkNotPoisoned();
TimerData timer =
TimerData.of(
timerId,
@@ -124,6 +140,7 @@ class WindmillTimerInternals implements TimerInternals {
@Override
public void deleteTimer(TimerData timerKey) {
+ checkNotPoisoned();
String timerDataKey = getTimerDataKey(timerKey.getTimerId(),
timerKey.getTimerFamilyId());
timerMap.put(
new SimpleEntry<>(timerDataKey, timerKey.getNamespace()),
@@ -139,6 +156,7 @@ class WindmillTimerInternals implements TimerInternals {
@Override
public void deleteTimer(
StateNamespace namespace, String timerId, String timerFamilyId,
TimeDomain timeDomain) {
+ checkNotPoisoned();
deleteTimer(
TimerData.of(
timerId,
@@ -152,12 +170,14 @@ class WindmillTimerInternals implements TimerInternals {
@Override
public Instant currentProcessingTime() {
+ checkNotPoisoned();
Instant now = Instant.now();
return processingTime.isAfter(now) ? processingTime : now;
}
@Override
public @Nullable Instant currentSynchronizedProcessingTime() {
+ checkNotPoisoned();
return watermarks.synchronizedProcessingTime();
}
@@ -172,6 +192,7 @@ class WindmillTimerInternals implements TimerInternals {
*/
@Override
public Instant currentInputWatermarkTime() {
+ checkNotPoisoned();
return watermarks.inputDataWatermark();
}
@@ -186,10 +207,12 @@ class WindmillTimerInternals implements TimerInternals {
*/
@Override
public @Nullable Instant currentOutputWatermarkTime() {
+ checkNotPoisoned();
return watermarks.outputDataWatermark();
}
public void persistTo(Windmill.WorkItemCommitRequest.Builder outputBuilder) {
+ checkNotPoisoned();
for (Entry<TimerData, Boolean> value : timerMap.values()) {
// Regardless of whether it is set or not, it must have some TimerData
stored so we
// can know its time domain
diff --git
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternals.java
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternals.java
index db036bee43c..f10ba70212e 100644
---
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternals.java
+++
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternals.java
@@ -55,6 +55,19 @@ public class WindmillStateInternals<K> implements
StateInternals {
private final CachingStateTable workItemDerivedState;
private final Supplier<Closeable> scopedReadStateSupplier;
+ private boolean poisoned = false;
+
+ public void poison() {
+ this.poisoned = true;
+ }
+
+ private void checkNotPoisoned() {
+ if (poisoned) {
+ throw new IllegalStateException(
+ "WindmillStateInternals is poisoned and cannot be used after
flushState().");
+ }
+ }
+
public WindmillStateInternals(
@Nullable K key,
String stateFamily,
@@ -78,6 +91,7 @@ public class WindmillStateInternals<K> implements
StateInternals {
@Override
public @Nullable K getKey() {
+ checkNotPoisoned();
return key;
}
@@ -104,6 +118,7 @@ public class WindmillStateInternals<K> implements
StateInternals {
}
public void persist(final Windmill.WorkItemCommitRequest.Builder
commitBuilder) {
+ checkNotPoisoned();
List<Future<WorkItemCommitRequest>> commitsToMerge = new ArrayList<>();
// Call persist on each first, which may schedule some futures for reading.
@@ -126,12 +141,14 @@ public class WindmillStateInternals<K> implements
StateInternals {
@Override
public <T extends State> T state(StateNamespace namespace, StateTag<T>
address) {
+ checkNotPoisoned();
return workItemState.get(namespace, address, StateContexts.nullContext());
}
@Override
public <T extends State> T state(
StateNamespace namespace, StateTag<T> address, StateContext<?> c) {
+ checkNotPoisoned();
return workItemState.get(namespace, address, c);
}
diff --git
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingGroupAlsoByWindowFnsTest.java
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingGroupAlsoByWindowFnsTest.java
index c31c39de9cc..fe5ba60acec 100644
---
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingGroupAlsoByWindowFnsTest.java
+++
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingGroupAlsoByWindowFnsTest.java
@@ -753,8 +753,7 @@ public class StreamingGroupAlsoByWindowFnsTest {
NullSideInputReader.empty(),
outputManager,
stepContext);
- return DoFnRunners.lateDataDroppingRunner(
- doFnRunner, stepContext.timerInternals(), windowingStrategy);
+ return DoFnRunners.lateDataDroppingRunner(doFnRunner, stepContext,
windowingStrategy);
}
private IntervalWindow window(long start, long end) {
diff --git
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContextTest.java
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContextTest.java
index 13601410bfd..056185c587f 100644
---
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContextTest.java
+++
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContextTest.java
@@ -42,6 +42,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.StateInternals;
import org.apache.beam.runners.core.StateNamespaceForTest;
import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.core.TimerInternals.TimerData;
@@ -460,4 +461,63 @@ public class StreamingModeExecutionContextTest {
assertEquals(1234, outputBuilder.getSourceBacklogBytes());
}
+
+ @Test
+ public void testInternalsPoisonedAfterFlushState() throws Exception {
+ Windmill.WorkItemCommitRequest.Builder outputBuilder =
+ Windmill.WorkItemCommitRequest.newBuilder();
+ NameContext nameContext = NameContextsForTests.nameContextForTest();
+ DataflowOperationContext operationContext =
+ executionContext.createOperationContext(nameContext);
+ StreamingModeExecutionContext.StepContext stepContext =
+ executionContext.getStepContext(operationContext);
+
+ executionContext.start(
+ "key",
+ createMockWork(
+
Windmill.WorkItem.newBuilder().setKey(ByteString.EMPTY).setWorkToken(17L).build(),
+ Watermarks.builder().setInputDataWatermark(new
Instant(1000)).build()),
+ stateReader,
+ sideInputStateFetcher,
+ outputBuilder,
+ workExecutor);
+
+ TimerInternals timerInternals = stepContext.timerInternals();
+ StateInternals stateInternals = stepContext.stateInternals();
+
+ executionContext.finishKey();
+ executionContext.flushState();
+
+ // Verify timerInternals is poisoned
+ try {
+ timerInternals.currentProcessingTime();
+ org.junit.Assert.fail("Expected IllegalStateException");
+ } catch (IllegalStateException e) {
+ assertThat(e.getMessage(), Matchers.containsString("poisoned"));
+ }
+
+ // Verify stateInternals is poisoned
+ try {
+ stateInternals.getKey();
+ org.junit.Assert.fail("Expected IllegalStateException");
+ } catch (IllegalStateException e) {
+ assertThat(e.getMessage(), Matchers.containsString("poisoned"));
+ }
+
+ // Verify stepContext.stateInternals() returns poisoned instance
+ try {
+ stepContext.stateInternals().getKey();
+ org.junit.Assert.fail("Expected IllegalStateException");
+ } catch (IllegalStateException e) {
+ assertThat(e.getMessage(), Matchers.containsString("poisoned"));
+ }
+
+ // Verify stepContext.timerInternals() returns poisoned instance
+ try {
+ stepContext.timerInternals().currentProcessingTime();
+ org.junit.Assert.fail("Expected IllegalStateException");
+ } catch (IllegalStateException e) {
+ assertThat(e.getMessage(), Matchers.containsString("poisoned"));
+ }
+ }
}
