apache / apache/beam

[Bug]: Cached ContinuationIterable permanently poisons state reads after its continuation token becomes invalid

Open
#39,383 1 comment 0 reactions 1 assignee Claimed by @addenergyx View on GitHub
dataflow P2 python
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.