danny0405 commented on code in PR #20415:
URL: https://github.com/apache/hudi/pull/20415#discussion_r4356019247
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -87,7 +87,24 @@ object HoodieSparkSqlWriter {
case class StreamingWriteParams(hoodieTableConfigOpt:
Option[HoodieTableConfig] = Option.empty,
asyncCompactionTriggerFn:
Option[SparkRDDWriteClient[_] => Unit] = Option.empty,
asyncClusteringTriggerFn:
Option[SparkRDDWriteClient[_] => Unit] = Option.empty,
- extraPreCommitFn:
Option[BiConsumer[HoodieTableMetaClient, HoodieCommitMetadata]] = Option.empty)
+ extraPreCommitFn:
Option[BiConsumer[HoodieTableMetaClient, HoodieCommitMetadata]] = Option.empty,
+ // When set, the caller constructs the write
client from the resolved write config and owns
+ // its lifecycle: the writer does not close
it after the write.
+ writeClientFactory:
Option[(HoodieSparkEngineContext, HoodieWriteConfig) => SparkRDDWriteClient[_]]
= Option.empty,
+ // Schema the caller-owned write client was
created with. The client is reused only while
+ // it matches the schema of the write,
otherwise the factory creates a new one.
+ writeClientSchema: Option[WriteClientSchema]
= Option.empty)
Review Comment:
Non-blocking abstraction suggestion: can we carry the caller-owned client
and its creation-schema snapshot in one handle, or expose an owner with
acquire(schema)/release semantics? Currently the client is a separate write
argument, while its factory and matching schema are two independent optional
fields, and the sink updates the client and snapshot separately. A factory with
a missing snapshot silently causes recreation on every write. Bundling this
state would encode the invariant directly and give the async retirement logic
one place to manage lifetime, instead of spreading ownership across the writer
and sink.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala:
##########
@@ -288,20 +296,158 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
}
+ /**
+ * 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[_] = {
+ 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 release the previous one. Its timeline
server is the sink's and keeps running.
+ previousClient.foreach(retireWriteClient(_, client))
+ registerQueryTerminationListener()
+ client
+ }
+
+ private def retireWriteClient(previousClient: SparkRDDWriteClient[_],
newClient: SparkRDDWriteClient[_]): Unit = {
+ if (asyncCompactorService != null || asyncClusteringService != null) {
+ // a compaction or clustering may still be running on the previous
client: close it once the services stop
+ if (asyncCompactorService != null) {
+ asyncCompactorService.updateWriteClient(newClient)
+ }
+ if (asyncClusteringService != null) {
+ asyncClusteringService.updateWriteClient(newClient)
+ }
+ retiredWriteClients = previousClient :: retiredWriteClients
Review Comment:
[P2] Could we release retired clients when the async operations using them
finish, rather than only when the entire service/query stops? These services
remain non-null while idle, so every schema change appends another open client
even when no compaction/clustering is in flight. A long-running stream with
repeated schema evolution therefore accumulates clients and their
index/table-service/transaction resources for its whole lifetime. A
per-operation client lease/reference count, or a handoff at a service safe
point, would bound retention to clients actually in use. Please cover multiple
schema changes with an active async service and assert that superseded clients
are eventually closed before query termination.
--
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]
