What happened?
GlobalCachingStateHandler (apache_beam/runners/worker/sdk_worker.py) can permanently poison its cross-bundle state cache.
blocking_get() caches the result of _partially_cached_iterable() under (state_key, cache_token). When the state cell is too large for the runner to return in one response, the cached object is a ContinuationIterable whose lazy tail holds the continuation token from that first read:
return self.ContinuationIterable(
head,
functools.partial(
self._lazy_iterator, state_key, coder, continuation_token))Continuation tokens can become permanently invalid. Dataflow, for example, revokes them together with the work item that created them. When that happens:
- iterating the cached object raises, but nothing invalidates the cache entry
- the user-state cache token is unchanged (nothing was written), so every later bundle reading that cell gets the same cached object and replays the dead token
- a fresh read (no token) would succeed, but never happens
The stage retries forever and its watermark stalls.
Observed on: Dataflow, Python streaming, Streaming Engine, Runner v2, Beam 2.74.0, with a stateful BatchElements whose bag state exceeded one read response. Every retry raised:
RuntimeError: INTERNAL: The work item requesting state read is no longer valid on the backend.
The work has already completed or will be retried. This is expected during autoscaling events.
The autoscaling wording is boilerplate; no resize events occurred while the errors and backlog accumulated. Errors grew from 1.6k to 17k/day while the job stayed JOB_STATE_RUNNING; only downstream data freshness revealed the stall.
Workaround: the runner only issues a continuation token when the state cell does not fit in a single response, so keeping bag state small makes this path unreachable. Flushing small batches frequently (max_batch_duration_secs=5, modest min_batch_size) eliminated the failures for us.
Proposed fix: invalidate the cache entry when iterating the continuation fails, so the next read fetches fresh state. A transient failure costs one extra reload; a permanent one self-heals instead of retrying indefinitely. PR with a deterministic unit repro (no Dataflow required): #39384.
Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components
What happened?
GlobalCachingStateHandler(apache_beam/runners/worker/sdk_worker.py) can permanently poison its cross-bundle state cache.blocking_get()caches the result of_partially_cached_iterable()under(state_key, cache_token). When the state cell is too large for the runner to return in one response, the cached object is aContinuationIterablewhose lazy tail holds the continuation token from that first read:Continuation tokens can become permanently invalid. Dataflow, for example, revokes them together with the work item that created them. When that happens:
The stage retries forever and its watermark stalls.
Observed on: Dataflow, Python streaming, Streaming Engine, Runner v2, Beam 2.74.0, with a stateful
BatchElementswhose bag state exceeded one read response. Every retry raised:The autoscaling wording is boilerplate; no resize events occurred while the errors and backlog accumulated. Errors grew from 1.6k to 17k/day while the job stayed
JOB_STATE_RUNNING; only downstream data freshness revealed the stall.Workaround: the runner only issues a continuation token when the state cell does not fit in a single response, so keeping bag state small makes this path unreachable. Flushing small batches frequently (
max_batch_duration_secs=5, modestmin_batch_size) eliminated the failures for us.Proposed fix: invalidate the cache entry when iterating the continuation fails, so the next read fetches fresh state. A transient failure costs one extra reload; a permanent one self-heals instead of retrying indefinitely. PR with a deterministic unit repro (no Dataflow required): #39384.
Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components