nsivabalan commented on code in PR #20415:
URL: https://github.com/apache/hudi/pull/20415#discussion_r4363169862
##########
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:
Thanks — agreed that holding replaced clients until the services stop
doesn't bound well over a long-running stream with repeated schema changes.
I've changed the sink to do exactly what `HoodieStreamer` does on a schema
change (`StreamSync.reInitWriteClient` +
`HoodieStreamer.onInitializingWriteClient`): hand the new client to the async
compaction/clustering services and close the previous client right away, so
nothing is retained.
The trade-off is the one `HoodieStreamer` already has: a compaction or
clustering that is mid-run on the previous client when the schema changes
continues on a closed client. If it fails, the async service stops with an
error and the writer stops: the sink fails its next micro-batch ("Async
Compactor shutdown unexpectedly"), and `HoodieStreamer` throws once it next
schedules a compaction/clustering. On restart, the pending instant is enqueued
again and re-run with the new client, so nothing is lost; the cost is a
restart, only when a schema change lands while a table service is running. A
per-operation lease (the sink can supply its own compactor/clusterer via
`createCompactor`/`createClusteringClient`) would avoid that for both writers,
which I'd rather do as a follow-up covering `HoodieStreamer` too than diverge
the sink from it here.
##########
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:
Good point on the invariant. In this PR the sink is the only caller that
sets `writeClientFactory`, and it always sets `writeClientSchema` with it (both
are updated together in `createWriteClient`), so a factory without a snapshot
doesn't happen today. I'd like to keep this PR scoped to the schema-drift fix
and fold the client, its schema snapshot and its lifetime into one owner as
part of the lease follow-up mentioned on the other thread, where that owner
would also manage the async services' clients.
--
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]
