wollowizard opened a new issue, #41108:
URL: https://github.com/apache/beam/issues/41108
### What happened?
**What happened?**
Since Dataflow turns on Windmill DirectPath by default for Java SDK 2.74.0
and later, our streaming pipelines show frequent storms of reader
invalidations. During these storms work is processed again and again without
making progress. The same pipeline without DirectPath runs normally.
**Setup**
• Beam Java 2.77.0, Dataflow Runner, Streaming Engine, Runner v1 (not Runner
v2).
• KafkaIO.read() using the legacy UnboundedSource path, so readers are
cached in ReaderCache .
• 1 worker, stateful processing (keyed state).
• enable_windmill_service_direct_path is added automatically by the
service; we didn't set it.
**Symptoms (DirectPath on)**
• Bursts in which every cached reader is closed with Asynchronously closing
reader for … as it is no longer valid , several times within minutes.
• The same key is often invalidated twice about 2s apart. First
acquireReader finds a cache-token mismatch, then onCompleteCommit sees a
non-OK commit status and calls invalidateReader .
• Readers restart from the same source position repeatedly. In one 2-hour
window, 492 of 1,118 reader starts resumed at an offset that had already been
used, so progress was being discarded.
• During storms CPU sits at 97–100% and system lag grows from a few seconds
to about 8 minutes. Input rate is normal, and committed state writes fall to
about 40–60% of normal.
• No client-side commit errors or aborts are logged: no Aborting … commits
, no Error occurred sending commits .
• Each invalidation closes and recreates the Kafka consumers, which led to a
large spike in new broker connections.
**Comparison ("no longer valid" reader closes)**
- Same code, SDK 2.77.0, DirectPath **on**: **400** over ~2 days
- Same code, SDK 2.77.0, run side by side with
`--experiments=disable_windmill_service_direct_path`: **16** (startup plus one
isolated event)
- Earlier pipeline version on SDK 2.73.0 (no DirectPath): **11** over ~28h
- Same pipeline on SDK 2.77.0 with DirectPath, same 28h: **583**
With DirectPath disabled there are no storms, no CPU saturation and no lag
spikes.
With DirectPath disabled there are no storms, no CPU saturation and no lag
spikes.
**Observations and questions**
1. StreamingDataflowWorker.onCompleteCommit invalidates the reader and
state cache on any non-OK CommitStatus without logging the status. This makes
it very hard to diagnose. Could the status be logged, or counted in a metric?
2. Is there a known issue with cache-token stability or commit rejections on
the DirectPath / FanOutStreamingEngineWorkerHarness path in 2.74–2.77?
3. Should DirectPath stay on by default while this is investigated?
### 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
- [x] 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]
