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

Đang mở
  • Có gì không ổn?
#41,108 5 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

  • #41169 của @stankiewicz — đã đóng, không merge
  • #41170 của @stankiewicz — đã đóng, không merge

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
15/100
Loại issue
Lỗi
Độ rõ ràng
Khá rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
google-cloud, java, kafka
Lĩnh vực
data-engineering, stream-processing

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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
Ngôn ngữ chính
Java
Star
8.7k
Fork
4.8k
Merge trung bình
2 ngày 12 giờ
Pull request đã merge (30 ngày)
371

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/beam

Tất cả issue của apache/beam

Issue tương tự

Thêm issue về Java

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.