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

Abierto
  • ¿Algo no está bien?
#41,108 5 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

  • #41169 de @stankiewicz — cerrado sin fusionar
  • #41170 de @stankiewicz — cerrado sin fusionar

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
15/100
Tipo de issue
Error
Claridad
Bastante claro
Estado de actividad
Activo
Stack tecnológico
google-cloud, java, kafka
Área
data-engineering, stream-processing

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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
Lenguaje dominante
Java
Estrellas
8.7k
Forks
4.8k
Merge medio
2 d 12 h
PR fusionados (30 d)
371

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/beam

Todos los issues de apache/beam

Issues similares

Más issues de Java

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.