danny0405 commented on code in PR #20415:
URL: https://github.com/apache/hudi/pull/20415#discussion_r4373969334
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala:
##########
@@ -288,20 +294,157 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
}
- private def reset(force: Boolean): Unit = this.synchronized {
+ /**
+ * Creates the write client for this sink with the sink's embedded timeline
server, starting the server on first
+ * use. The client does not own the server, so the server keeps running for
as long as the sink does.
+ */
+ private def createWriteClient(context: HoodieSparkEngineContext,
writeConfig: HoodieWriteConfig): SparkRDDWriteClient[_] = {
+ // register the release first, so that a timeline server started for a
client that then fails to build is
+ // still stopped when the query terminates
+ registerQueryTerminationListener()
+ val clientSchema = WriteClientSchema.of(writeConfig)
+ val client = new SparkRDDWriteClient(context, writeConfig,
getOrStartTimelineService(context, writeConfig))
+ val previousClient = writeClient
+ // track the client as soon as it exists so that a failed micro-batch
still reuses and eventually closes it
+ writeClient = Some(client)
+ writeClientSchema = Some(clientSchema)
+ // the writer re-creates the client when the write schema changes (as
HoodieStreamer does): hand the new client
+ // to the async table services and close the previous one. Its timeline
server is the sink's and keeps running.
+ previousClient.foreach(retireWriteClient(_, client))
+ client
+ }
+
+ /**
+ * Same as HoodieStreamer on a schema change (StreamSync#reInitWriteClient
and
+ * HoodieStreamer#onInitializingWriteClient): the async table services
switch to the new client for their next
+ * runs, and the previous client is closed right away.
+ */
+ private def retireWriteClient(previousClient: SparkRDDWriteClient[_],
newClient: SparkRDDWriteClient[_]): Unit = {
+ if (asyncCompactorService != null) {
+ asyncCompactorService.updateWriteClient(newClient)
+ }
+ if (asyncClusteringService != null) {
+ asyncClusteringService.updateWriteClient(newClient)
+ }
+ closeQuietly(previousClient)
Review Comment:
[P2] Wait for in-flight table services before closing their client
This fixes retention by closing a client that can still be in use.
HoodieSparkCompactor.compact captures the current client in a local variable
and uses it for both compact and commitCompaction; HoodieSparkClusteringClient
similarly captures its client for the run. updateWriteClient only changes the
client for subsequent runs. Meanwhile close() stops heartbeats, closes
transaction/index/table-service resources, and resets the old config's
remote-view settings underneath the active run. A schema change during
compaction/clustering can therefore terminate an otherwise healthy query;
keeping the shared timeline server alive does not preserve these other
resources. I saw the documented HoodieStreamer trade-off, but matching that
behavior introduces the race into this sink and does not make a live-client
close safe. Could we drain the current runs before closing, or use a bounded
per-run lease? A latch-controlled test that pauses an async run, evolves the
schema, and verifies the old c
lient stays open through completion would cover this.
--
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]
