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

Ouverte
  • Quelque chose ne va pas ?
#41,108 5 commentaires 0 réactions 0 personnes assignées Voir sur GitHub

Les mainteneurs répondent en général sous 1 jour

Personne n'a encore pris cette issue.

  • #41169 par @stankiewicz — fermée sans fusion
  • #41170 par @stankiewicz — fermée sans fusion

Évaluation

Difficulté
4/5
Temps estimé
3-5 jours
Accessibilité débutants
15/100
Type d'issue
Bug
Clarté
Plutôt claire
Activité
Active
Stack technique
google-cloud, java, kafka
Domaine
data-engineering, stream-processing

Piste de recherche

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.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

Description

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
Langage dominant
Java
Étoiles
8.7k
Forks
4.8k
Merge moyen
2 j 12 h
PR mergées (30 j)
371

Préparer son environnement

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de apache/beam

Toutes les issues de apache/beam

Issues similaires

Plus d'issues Java

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.