nsivabalan opened a new issue, #20400:
URL: https://github.com/apache/hudi/issues/20400
### Bug Description
`HoodieStreamingSink` cannot keep an embedded timeline server alive across
micro-batches, so Spark Structured Streaming writes run without one.
`HoodieStreamer` does not have this problem. This issue proposes aligning the
sink with `HoodieStreamer`: **one embedded timeline server per sink instance
(per streaming query on the driver), reused across every commit, and stopped
when the query terminates.**
**Current behaviour (master `60caf5f4`)**
1. The write client is created on the first micro-batch inside
`HoodieSparkSqlWriter`. It starts its own embedded timeline server (in the
constructor via `BaseHoodieClient.startEmbeddedServerView`) and is therefore
its owner.
2. At the end of each write, `HoodieSparkSqlWriterInternal.closeWriteClient`
closes the client unless async compaction or async clustering is enabled.
Closing the owner stops the timeline server and resets the view-storage config.
3. The sink keeps the returned client (`writeClient = Some(client)`) and
hands it back on the next batch, including when it has already been closed. A
client only starts a timeline server in its constructor, so batch 2 onwards
runs without one.
4. To avoid that, the sink forces `hoodie.embed.timeline.server=false`
unless the user sets it (HUDI-3636, #9266), and always forces `DIRECT` markers
(comment in `addBatch`: *"timeline server is closed after first micro-batch and
subsequent micro-batches do not have timeline server running"*).
Consequences:
- Streaming writes never get the timeline server's file-system view caching
or timeline-server-based markers.
- A user who enables the server with `REMOTE_ONLY` fails on the second
micro-batch with `HoodieRemoteException: Connect to localhost:<port> ...
Connection refused` (reported in #12652).
- With async compaction or clustering enabled, the client (and any server it
owns) is never closed when the query stops. Cleanup only happens through JVM
shutdown hooks.
**How `HoodieStreamer` does it**
`StreamSync` owns the timeline server. It creates it once
(`EmbeddedTimelineServerHelper.createEmbeddedTimelineService`) and passes it
into every `SparkRDDWriteClient` it builds. When a client is rebuilt (for
example on a schema change), only the client is closed. The server keeps
running, because a client that was handed a server does not stop it. The server
is stopped once, in `StreamSync.close()` (`stopForBasePath`).
**Proposed fix**
- `HoodieStreamingSink` owns an `EmbeddedTimelineService`, started lazily on
the first batch from that batch's write config, and passed into the write
client (non-owning, as `StreamSync` does).
- In streaming mode the writer no longer closes the client after each batch.
The sink owns the client's lifecycle, and async compaction and clustering keep
sharing it.
- The server is stopped when the query terminates. A V1 `Sink` has no close
hook, so this would use a `StreamingQueryListener` keyed on the sink's query
id. It is also stopped on `reset`.
- Drop the HUDI-3636 default override and the forced `DIRECT` markers.
`hoodie.embed.timeline.server=false` and an explicit
`hoodie.write.markers.type` keep working as user opt-outs.
**Why now:** Spark Declarative Pipelines (RFC-112, #19997) runs every
streaming flow as a separate `HoodieStreamingSink` on one long-lived Spark
Connect driver. Each flow needs a correctly scoped timeline server that is
released when its query stops, not when the JVM exits.
### Environment
Behaviour verified against apache/hudi `master` (`60caf5f4`); originally
reported on Hudi 0.15.0 / Spark 3.3.0 in #12652.
### Steps to reproduce
Run any `writeStream.format("hudi")` query with
`hoodie.embed.timeline.server=true`, `hoodie.filesystem.view.type=REMOTE_ONLY`,
`hoodie.filesystem.remote.backup.view.enable=false`. Batch 0 succeeds. Batch 1
fails with `Connection refused` against the stopped server's port. Full repro
and stack trace in #12652.
### Code pointers
-
`hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala`:
`addBatch` (marker and timeline-server overrides, client reuse), `reset`
-
`hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala`:
`closeWriteClient` and the `finally` in `writeInternal`
-
`hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java`:
`startEmbeddedServerView` / `stopEmbeddedServerView` (ownership via
`shouldStopTimelineServer`)
-
`hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java`:
`reInitWriteClient`, `close` (reference behaviour)
Related: #12652, #9266 (HUDI-3636), #20370, #19997.
--
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]
