[Bug]: Dataflow streaming on SDK 2.74+ (Windmill DirectPath on by default) causes frequent UnboundedSource reader invalidations, re-processing and CPU saturation

未关闭
  • 哪里不对？
#41,108 5 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

维护者通常 1 天内回复

还没有人认领这个 Issue。

  • #41169 来自 @stankiewicz —— 已关闭，未合并
  • #41170 来自 @stankiewicz —— 已关闭，未合并

评估

难度
4/5
预计耗时
3-5 天
新手友好度
15/100
Issue 类型
缺陷
描述清晰度
基本清楚
活跃度
活跃
技术栈
google-cloud, java, kafka
领域
data-engineering, stream-processing

调研方向

Start in StreamingDataflowWorker.onCompleteCommit, where a non-OK CommitStatus invalidates the reader and state cache without logging, and follow the cache-token check in acquireReader plus the FanOutStreamingEngineWorkerHarness DirectPath path. A first step is logging or counting the commit status, which the issue asks for. Done means that change is in place and the root cause of the cache-token mismatch or commit rejection is identified, which the issue leaves open.

由索引模型根据 Issue 内容生成。

描述

awaiting triage bug dataflow io java P2
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
  • 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
  • Component: Google Cloud Dataflow Runner
主要语言
Java
星标
8.7k
派生
4.8k
平均合并
2 天 12 小时
30 天内合并 PR
371

环境准备

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

apache/beam 的其他 Issue

查看 apache/beam 的全部 Issue

相似的 Issue

更多 Java Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。