danny0405 commented on code in PR #20415:
URL: https://github.com/apache/hudi/pull/20415#discussion_r4374113972
##########
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:
Addressed in f03644b0b62. Async compaction/clustering now use the same
operation monitor for a complete run and `updateWriteClient`, so handoff waits
for an active run (including commit) to finish. The sink closes the old client
only after both handoffs return. This bounds client lifetime without retaining
retired clients until query termination. Four latch-controlled cases cover
compaction/clustering with successful and failed active runs, verify no
premature close, and check replacement use on the next run and prompt handoff
while idle. All 147 related Java tests passed on the final stack with JDK 17
and UTC; Checkstyle and RAT passed. The deliberate trade-off is that a
schema-changing micro-batch waits for any active table-service run. The broader
client/schema-owner API refactor remains a separate non-blocking follow-up.
--
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]
