nsivabalan commented on code in PR #20412:
URL: https://github.com/apache/hudi/pull/20412#discussion_r4362937592
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -576,17 +581,41 @@ class HoodieSparkSqlWriterInternal {
(writeSuccessful, common.util.Option.ofNullable(instantTime),
compactionInstant, clusteringInstant, writeClient, tableConfig)
} finally {
- closeWriteClient(writeClient, tableConfig, parameters,
jsc.hadoopConfiguration())
+ closeWriteClient(writeClient, instantTime, tableConfig, parameters,
jsc.hadoopConfiguration())
}
}
}
- private def closeWriteClient(writeClient: SparkRDDWriteClient[_],
tableConfig: HoodieTableConfig, parameters: Map[String, String], configuration:
Configuration): Unit = {
- // close the write client in all cases
- val asyncCompactionEnabled = isAsyncCompactionEnabled(writeClient,
tableConfig, parameters, configuration)
- val asyncClusteringEnabled = isAsyncClusteringEnabled(writeClient,
parameters)
- if (!asyncCompactionEnabled && !asyncClusteringEnabled) {
- writeClient.close()
+ private def createWriteClient(jsc: JavaSparkContext, schemaStr: String,
path: String, tblName: String,
+ parameters: java.util.Map[String, String],
+ streamingWritesParamsOpt:
Option[StreamingWriteParams]): SparkRDDWriteClient[_] = {
+ streamingWritesParamsOpt.flatMap(_.writeClientFactory) match {
+ case Some(factory) =>
+ DataSourceUtils.createHoodieClient(jsc, schemaStr, path, tblName,
parameters,
+ new BiFunction[HoodieSparkEngineContext, HoodieWriteConfig,
SparkRDDWriteClient[_]] {
+ override def apply(context: HoodieSparkEngineContext, writeConfig:
HoodieWriteConfig): SparkRDDWriteClient[_] =
+ factory(context, writeConfig)
+ })
+ case None =>
+ DataSourceUtils.createHoodieClient(jsc, schemaStr, path, tblName,
parameters)
+ }
+ }
+
+ private def closeWriteClient(writeClient: SparkRDDWriteClient[_],
instantTime: String, tableConfig: HoodieTableConfig,
+ parameters: Map[String, String], configuration:
Configuration): Unit = {
+ if (writeClientOwnedByCaller) {
+ // a caller-owned client (streaming sink) is reused across writes and
closed by its owner. Stop the heartbeat
+ // of this write's instant in case it failed before commit (a no-op
after a successful commit), so that the
+ // failed instant can be rolled back while the client lives on.
+ if (instantTime != null) {
+ writeClient.releaseResources(instantTime)
Review Comment:
Good catch, thanks. The writer now records the instant (and client) at every
commit start — upsert/insert, DELETE, DELETE_PARTITION and bulk insert as row —
and `write()` stops that instant's heartbeat on any exception out of the write,
when the caller owns the client. The release in `closeWriteClient` stays for
writes that fail without throwing (e.g. write-status errors). Added
`testStructuredStreamingReleasesHeartbeatOfFailedDelete`: a delete stream whose
record has no record key fails inside the delete operation after the commit
started, and the test asserts the inflight instant has no heartbeat.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala:
##########
@@ -288,21 +292,126 @@ 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 client = new SparkRDDWriteClient(context, writeConfig,
getOrStartTimelineService(context, writeConfig))
Review Comment:
Agreed. `createWriteClient` now registers the termination listener before
acquiring the timeline server, so a server started for a client whose
construction then fails is still stopped when the query terminates (it is
sink-owned, and the next micro-batch reuses it).
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala:
##########
@@ -508,6 +511,142 @@ class TestStructuredStreaming extends
HoodieSparkClientTestBase {
assertEquals(25, metaClient.getActiveTimeline.countInstants())
}
+ /**
+ * The sink owns one embedded timeline server for the lifetime of the
streaming query and reuses it across
+ * micro-batches. With a REMOTE_ONLY file system view and no backup view,
every micro-batch after the first used
+ * to fail because the timeline server was stopped along with the first
micro-batch's write client
+ * (https://github.com/apache/hudi/issues/12652). Failed batches are ignored
here so that a regression shows up
+ * as missing commits rather than the sink exiting the JVM.
+ */
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieTableType])
+ def testStructuredStreamingReusesTimelineServerAcrossBatches(tableType:
HoodieTableType): Unit = {
+ val (sourcePath, destPath) = initStreamingSourceAndDestPath("source",
"dest")
+ val opts = getOptsWithTableType(tableType) ++ Map(
+ HoodieWriteConfig.EMBEDDED_TIMELINE_SERVER_ENABLE.key -> "true",
+ FileSystemViewStorageConfig.VIEW_TYPE.key ->
FileSystemViewStorageType.REMOTE_ONLY.name,
+ FileSystemViewStorageConfig.REMOTE_BACKUP_VIEW_ENABLE.key -> "false",
+ DataSourceWriteOptions.STREAMING_IGNORE_FAILED_BATCH.key -> "true",
+ DataSourceWriteOptions.STREAMING_RETRY_CNT.key -> "1")
+
+ val records1 = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF1 = spark.read.json(spark.sparkContext.parallelize(records1, 2))
+ inputDF1.coalesce(1).write.mode(SaveMode.Append).json(sourcePath)
+ val query = initWritingStreamingQuery(inputDF1.schema, sourcePath,
destPath, opts)
+ try {
+ query.processAllAvailable()
+ for (i <- 1 to 2) {
+ val records =
recordsToStrings(dataGen.generateUpdates("%03d".format(i), 50)).asScala.toList
+ spark.read.json(spark.sparkContext.parallelize(records, 2)).coalesce(1)
+ .write.mode(SaveMode.Append).json(sourcePath)
+ query.processAllAvailable()
+ }
+ } finally {
+ query.stop()
+ }
+
+ assertEquals(3,
HoodieDataSourceHelpers.allCompletedCommitsCompactions(storage,
destPath).countInstants())
+ }
+
+ /**
+ * The sink keeps its write client open across micro-batches, so a
micro-batch that fails after its commit started
+ * must still stop that instant's heartbeat. Otherwise, with the lazy
failed-writes clean policy, the failed instant
+ * looks alive for as long as the query runs and is never rolled back.
+ */
+ @Test
+ def testStructuredStreamingReleasesHeartbeatOfFailedBatch(): Unit = {
+ val (sourcePath, destPath) = initStreamingSourceAndDestPath("source",
"dest")
+ val opts = getOptsWithTableType(HoodieTableType.COPY_ON_WRITE) ++ Map(
+ DataSourceWriteOptions.OPERATION.key -> UPSERT_OPERATION_OPT_VAL,
+ DataSourceWriteOptions.RECORD_MERGE_MODE.key -> "CUSTOM",
+ DataSourceWriteOptions.RECORD_MERGE_STRATEGY_ID.key ->
FailingSparkRecordMerger.STRATEGY_ID,
+ DataSourceWriteOptions.RECORD_MERGE_IMPL_CLASSES.key ->
classOf[FailingSparkRecordMerger].getName,
+ HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key ->
WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.name,
+ HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key ->
HoodieFailedWritesCleaningPolicy.LAZY.name,
+ HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key ->
classOf[InProcessLockProvider].getName,
+ STREAMING_CHECKPOINT_IDENTIFIER.key -> "streaming_writer",
+ DataSourceWriteOptions.STREAMING_IGNORE_FAILED_BATCH.key -> "true",
+ DataSourceWriteOptions.STREAMING_RETRY_CNT.key -> "1")
+
+ val records1 = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF1 = spark.read.json(spark.sparkContext.parallelize(records1, 2))
+ inputDF1.coalesce(1).write.mode(SaveMode.Append).json(sourcePath)
+ val query = initWritingStreamingQuery(inputDF1.schema, sourcePath,
destPath, opts)
+ try {
+ query.processAllAvailable()
+
+ // the updates of the next micro-batch fail while merging, after the
commit started
+ FailingSparkRecordMerger.shouldFail = true
+ try {
+ spark.read.json(spark.sparkContext.parallelize(
+ recordsToStrings(dataGen.generateUniqueUpdates("001",
50)).asScala.toList, 2)).coalesce(1)
+ .write.mode(SaveMode.Append).json(sourcePath)
+ query.processAllAvailable()
+ } finally {
+ FailingSparkRecordMerger.shouldFail = false
+ }
+ val failedInstant = HoodieTestUtils.createMetaClient(storage,
destPath).getActiveTimeline
+ .getCommitsTimeline.filterInflightsAndRequested().lastInstant()
+ assertTrue(failedInstant.isPresent, "the failed micro-batch must leave
an inflight instant")
+ assertFalse(storage.exists(new
StoragePath(HoodieTableMetaClient.getHeartbeatFolderPath(destPath),
+ failedInstant.get().requestedTime())), "the heartbeat of the failed
instant must be stopped")
+
+ // the query recovers on the next micro-batch
+ spark.read.json(spark.sparkContext.parallelize(
+ recordsToStrings(dataGen.generateUniqueUpdates("002",
50)).asScala.toList, 2)).coalesce(1)
+ .write.mode(SaveMode.Append).json(sourcePath)
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+ assertEquals(2,
HoodieDataSourceHelpers.allCompletedCommitsCompactions(storage,
destPath).countInstants())
+ }
+
+ /**
+ * A micro-batch that exhausts its retries resets the sink, which shuts down
the async table services and waits for
+ * them. The services' shutdown callbacks must not block on the sink in
turn, otherwise the query hangs.
+ */
+ @Test
+ def testStructuredStreamingFailedBatchWithAsyncClusteringDoesNotHang(): Unit
= {
+ val (sourcePath, destPath) = initStreamingSourceAndDestPath("source",
"dest")
+ val opts = getClusteringOpts(HoodieTableType.COPY_ON_WRITE, "false",
"true", "100", 100) ++ Map(
+ DataSourceWriteOptions.OPERATION.key -> UPSERT_OPERATION_OPT_VAL,
+ DataSourceWriteOptions.RECORD_MERGE_MODE.key -> "CUSTOM",
+ DataSourceWriteOptions.RECORD_MERGE_STRATEGY_ID.key ->
FailingSparkRecordMerger.STRATEGY_ID,
+ DataSourceWriteOptions.RECORD_MERGE_IMPL_CLASSES.key ->
classOf[FailingSparkRecordMerger].getName,
+ DataSourceWriteOptions.STREAMING_IGNORE_FAILED_BATCH.key -> "true",
Review Comment:
You're right — with `ignoreFailedBatch` the failure becomes a `Success`
before `retry` sees it, so that test never reached the exhausted-retries reset.
Replaced it with `testSinkResetWithAsyncServiceRunningDoesNotHang`, which
drives the reset directly through a test subclass of the sink: it starts async
clustering, then calls `reset(false)` while holding the sink lock (as
`addBatch` does after exhausting its retries), and asserts it finishes within
the timeout. `reset` is now `protected`, like the trigger methods, for that.
--
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]
