[Bug]: Cached ContinuationIterable permanently poisons state reads after its continuation token becomes invalid
- Dominant language
- Java
- Stars
- 8.7k
- Forks
- 4.7k
- Avg merge
- 1d 20h
- Merged PRs (30d)
- 196
Description
### 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:
```python
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
- [X] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [X] Component: Google Cloud Dataflow Runner
Contributor guide
Assessment
This issue has not been assessed yet.