danny0405 commented on code in PR #20412:
URL: https://github.com/apache/hudi/pull/20412#discussion_r4356018802
##########
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:
[P2] Could we put per-instant cleanup in a finally that encloses the whole
operation after startCommit? This helper is only reached by the normal-write
doWriteOperation catch and the later commit/post-operation finally.
DELETE/DELETE_PREPPED and DELETE_PARTITION can throw from
doDeleteOperation/doDeletePartitionsOperation before reaching either, and
record-RDD creation can also fail before the normal-write catch. With LAZY
cleaning and streaming.ignore.failed.batch=true, the sink keeps this client
alive and converts the failure to Success, so reset never stops that instant's
heartbeat; the failed instant continues to look live. Please include a failing
delete or record-construction case in the heartbeat test.
##########
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:
[P2] This setting prevents the test from exercising the deadlock it
describes. In addBatch, ignoreFailedBatch converts the injected Failure to
Success before retry sees it; retry therefore never reaches its
exhausted-retries reset(false). query.stop() now runs cleanup on a separate
thread, and this test does not wait for that cleanup either. Could we test the
actual exhausted-retry/reset path with the process-exit behavior isolated, or
directly exercise reset while a controlled async service completes its shutdown
callback, and assert that both calls finish within the timeout?
##########
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:
[P2] The timeline service can be started before the client constructor
throws, but the termination listener is only registered after construction
succeeds. For example, a client-init callback failure with
streaming.ignore.failed.batch=true leaves embeddedTimelineService populated,
writeClient empty, and no termination listener; stopping the query then never
closes the server. Could we register cleanup before acquiring the server, or
close the newly acquired server on construction failure? An exception-safe
resource owner would also make the client/server lifetime easier to reason
about.
--
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]
