stankiewicz opened a new issue, #41294:
URL: https://github.com/apache/beam/issues/41294
### What happened?
In the Dataflow Java Runner v1 worker
(`org.apache.beam.runners.dataflow.worker.windmill.state`), calling
`.readLater()` on `ValueState` (`WindmillValue`) or `BagState` (`WindmillBag`)
checks the worker's in-memory state cache first and is a no-op when the state
is already cached.
By contrast, calling `mapState.get(k).readLater()` (`WindmillMapViaMultimap`
/ `WindmillMap`) or `multimapState.get(k).readLater()` (`WindmillMultimap`)
skips all in-memory cache checks and unconditionally queues a backend
`StateTag` lookup into `WindmillStateReader.pendingLookups`. If any single
state read in that bundle misses the worker cache and calls
`WrappedFuture.get()`, `WindmillStateReader.performReads()` flushes the entire
`pendingLookups` queue over gRPC to Windmill, re-fetching and re-deserializing
every key that had `.readLater()` called on it, even keys that were already
cached in JVM memory or deleted in the same bundle.
---
### Root Cause in Code
#### 1. `WindmillBag` and `WindmillValue` (Guarded)
In `WindmillBag.java` (and `WindmillValue.java`), `readLater()` calls
`getFuture()`, which checks `cachedValues != null` before registering a lookup
with `WindmillStateReader`:
```java
//
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillBag.java
@Override
public WindmillBag<T> readLater() {
getFuture();
return this;
}
private Future<Iterable<T>> getFuture() {
return cachedValues != null
? null
: reader.bagFuture(stateKey.byteString(), stateFamily, elemCoder);
}
```
#### 2. `WindmillMultimap` and `WindmillMapViaMultimap` (Unguarded)
On Streaming Engine, `MapState<K, V>` is backed by `WindmillMapViaMultimap`,
which delegates `mapState.get(key).readLater()` to
`WindmillMultimap.ReadResultReadableState.readLater()`.
In `WindmillMultimap.java`, `read()` checks `KNOWN_NONEXISTENT`,
`removedLocally`, `valuesCached`, and `complete` before touching
`WindmillStateReader`, whereas `readLater()` bypasses all of those checks:
```java
//
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillMultimap.java
@Override
public Iterable<V> read() {
KeyState keyState = ...;
if (keyState.existence == KeyExistence.KNOWN_NONEXISTENT) {
return Collections.emptyList();
}
if (keyState.removedLocally) {
return Iterables.unmodifiableIterable(localNewValues);
}
if (keyState.valuesCached || complete) {
// Cache hit: returns from JVM memory without touching
WindmillStateReader
return Iterables.unmodifiableIterable(
Iterables.concat(
Iterables.limit(keyState.values, keyState.valuesSize),
localNewValues));
}
Future<Iterable<V>> persistedData =
necessaryKeyEntriesFromStorageFuture(key);
...
}
@Override
@SuppressWarnings("FutureReturnValueIgnored")
public ReadableState<Iterable<V>> readLater() {
// Skips all cache checks and unconditionally adds StateTag to
reader.pendingLookups
WindmillMultimap.this.necessaryKeyEntriesFromStorageFuture(key);
return this;
}
```
Note that `WindmillMultimap.KeysReadableState.readLater()` has a similar gap
(`necessaryEntriesFromStorageFuture(true)` checks `if (complete)` but not `if
(allKeysKnown)`), and `WindmillMap.WindmillMapReadResultReadableState` calls
`getFutureForKey(key)` unconditionally in both `readLater()` and the first line
of `read()`.
---
### Impact: Contagious Cache-Miss Read Amplification
1. Calling `necessaryKeyEntriesFromStorageFuture(key)` or
`getFutureForKey(key)` calls `WindmillStateReader.stateFuture()`, which appends
a `StateTag` to `WindmillStateReader.pendingLookups` and returns a
`WrappedFuture`.
2. In `WrappedFuture.get()`, the first unfulfilled future that is awaited
triggers `reader.performReads()` for all pending tags in the reader:
```java
@Override
public T get() throws InterruptedException, ExecutionException {
if (!delegate().isDone() && reader != null) {
reader.performReads();
}
reader = null;
return super.get();
}
```
3. **Example with 9 warm keys and 1 cold key**:
Suppose an `@OnTimer` callback prefetches 10 sub-keys via
`multimapState.get(k).readLater()` or `mapState.get(k).readLater()`, where 9 of
the 10 sub-keys are already cached in `WindmillStateCache`
(`keyState.valuesCached == true`) and 1 sub-key is a cache miss (or another
state family in the same `DoFn` misses cache):
- All 10 `StateTag`s are queued into `reader.pendingLookups` by
`.readLater()`.
- Calling `.read()` on the 9 warm sub-keys returns their cached values
from memory without calling `WrappedFuture.get()`.
- Calling `.read()` on the 1 cold sub-key calls `persistedData.get()`
(`WrappedFuture.get()`).
- `reader.performReads()` drains the entire `pendingLookups` queue and
sends a `GetData` request to Windmill for **all 10 sub-keys** (the 1 cache miss
plus the 9 cache hits that were already served from memory), reading all 10
from backend storage and re-decoding all 10 responses in
`WindmillStateReader.consumeResponse()`.
---
### Current Workaround (and Why It Is Insufficient)
The only user-side workaround is to remove `.readLater()` calls on
`MapState.get(k)` and `MultimapState.get(k)` and call `.read()` directly on
each key in a loop.
However, removing `.readLater()` creates a severe penalty whenever keys
actually miss the worker cache (such as after worker startup, key-range
rebalance, or cache eviction): without `.readLater()`, reading $N$ cold keys in
a loop issues **$N$ sequential, blocking `GetData` RPCs** to Windmill instead
of batching them into a single `GetData` request. Users are forced to choose
between:
- **Calling `.readLater()`**: Batches cold reads into 1 RPC, but any single
cache miss in the bundle re-fetches and re-deserializes all warm keys from
backend storage.
- **Omitting `.readLater()`**: Avoids re-fetching warm keys, but degrades to
$N$ serial RPC round-trips whenever keys are cold.
---
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [x] 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: Prism Runner
- [ ] Component: Hazelcast Jet Runner
- [x] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
