You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "nsivabalan (via GitHub)" <gi...@apache.org> on 2026/10/22 16:29:29 UTC
[PR] fix(spark): reuse one embedded timeline server across streaming sink micro-batches [hudi]
nsivabalan opened a new pull request, #20412:
URL: https://github.com/apache/hudi/pull/20412
### Describe the issue this Pull Request addresses
Closes #20400 (same root cause as #12652).
`HoodieStreamingSink` could not keep an embedded timeline server alive across micro-batches. The writer closed the write client after every batch (unless async table services were on), which stopped the server it owned, and the sink then reused that closed client. To paper over this, the sink disabled the timeline server by default (HUDI-3636). Users who turned it on with a `REMOTE_ONLY` view failed on the second micro-batch with `Connection refused`.
This aligns the sink with `HoodieStreamer`: one embedded timeline server per streaming query, reused across every commit.
### Summary and Changelog
- `HoodieStreamingSink` owns an `EmbeddedTimelineService`. It is started on the first micro-batch and passed into the write client, which does not own it (as `StreamSync` does), so closing or rebuilding a client never stops the server.
- The write client is kept open across micro-batches; the sink owns its lifecycle.
- A `StreamingQueryListener` keyed on the query id closes the client and stops the server when the query terminates. A V1 `Sink` has no close hook, and long-lived drivers (e.g. a Spark Connect server running Spark Declarative Pipelines flows, #19997) would otherwise hold them until JVM exit.
- Removed the HUDI-3636 default that set `hoodie.embed.timeline.server=false` for streaming writes. Setting it explicitly still opts out.
- `HoodieSparkSqlWriter`: new `StreamingWriteParams.writeClientFactory`. When set, the writer builds the client through it and does not close it after the write.
- `DataSourceUtils.createHoodieClient`: new overload taking a client factory; the existing method delegates to it.
- Markers stay forced to `DIRECT` in this PR. Switching to timeline-server-based markers is a separate stacked PR.
### Impact
- The timeline server is now on by default for Spark structured streaming writes into Hudi, giving them the cached file-system view.
- On query termination, in-flight async compaction/clustering is force-stopped rather than left running until JVM exit (a graceful stop would block Spark's listener bus thread).
- Known follow-up (pre-existing, also true for `HoodieStreamer`): the metadata table writer starts and stops its own short-lived timeline server per commit.
### Risk Level
medium. Changes the write client and timeline server lifecycle for every streaming write. Verified with `TestStructuredStreaming` (all tests pass), including a new COW/MOR test that runs 3 micro-batches with `REMOTE_ONLY` view and no backup view (the #12652 scenario).
### Documentation Update
none
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): reuse one embedded timeline server across streaming sink micro-batches [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #20412:
URL: https://github.com/apache/hudi/pull/20412#discussion_r4373968980
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -533,7 +568,7 @@ class HoodieSparkSqlWriterInternal {
// scalastyle:on
val writeConfig = client.getConfig
- instantTime = client.startCommit(commitActionType)
+ instantTime = recordStartedInstant(client, client.startCommit(commitActionType))
Review Comment:
[P2] Handle failures inside startCommit after the heartbeat starts
The outer cleanup now covers failures after startCommit returns, but argument evaluation means recordStartedInstant is never called if startCommit itself throws. BaseHoodieWriteClient.startCommit starts the LAZY-policy heartbeat before createNewInstant/createRequestedCommitWithReplaceMetadata, and its finally only ends the transaction. If writing the requested instant fails (or succeeds on storage but reports an error), startedInstant remains None and releaseStartedInstant does nothing. With ignoreFailedBatch=true, the live client is retained, so that heartbeat keeps running; a persisted requested instant remains protected from lazy rollback. Could startCommit clean up its own heartbeat on exceptional exit, or otherwise expose the allocated instant before that fallible storage operation? Please add failure injection after heartbeat startup and before startCommit returns.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): reuse one embedded timeline server across streaming sink micro-batches [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20412:
URL: https://github.com/apache/hudi/pull/20412#issuecomment-6356591490
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20412?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 79.65%. Comparing base ([`60caf5f`](https://app.codecov.io/gh/apache/hudi/commit/60caf5f423989ff6d07136407bc8b0c20ee04c65?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`43fa11b`](https://app.codecov.io/gh/apache/hudi/commit/43fa11b85532b1eaaa448d808a274a1d0d3994c6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 84 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20412 +/- ##
============================================
- Coverage 80.49% 79.65% -0.84%
- Complexity 34953 35509 +556
============================================
Files 2549 2551 +2
Lines 143060 146947 +3887
Branches 17392 18142 +750
============================================
+ Hits 115151 117050 +1899
- Misses 20004 21792 +1788
- Partials 7905 8105 +200
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (-3.00%)` | :arrow_down: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.04% <ø> (-0.50%)` | :arrow_down: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.96% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.83% <ø> (-0.03%)` | :arrow_down: |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.18% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.78% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.60% <ø> (-0.44%)` | :arrow_down: |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.23% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.69% <ø> (+0.19%)` | :arrow_up: |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.15% <ø> (-0.15%)` | :arrow_down: |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20412/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20412/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20412/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.94% <ø> (-0.04%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20412?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...src/main/java/org/apache/hudi/DataSourceUtils.java](https://app.codecov.io/gh/apache/hudi/pull/20412?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2FDataSourceUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmstY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9odWRpL0RhdGFTb3VyY2VVdGlscy5qYXZh) | `71.07% <ø> (-0.24%)` | :arrow_down: |
| [...n/scala/org/apache/hudi/HoodieSparkSqlWriter.scala](https://app.codecov.io/gh/apache/hudi/pull/20412?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark-common%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fhudi%2FHoodieSparkSqlWriter.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmstY29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvaHVkaS9Ib29kaWVTcGFya1NxbFdyaXRlci5zY2FsYQ==) | `78.54% <ø> (-0.45%)` | :arrow_down: |
| [...in/scala/org/apache/hudi/HoodieStreamingSink.scala](https://app.codecov.io/gh/apache/hudi/pull/20412?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark-common%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fhudi%2FHoodieStreamingSink.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmstY29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvaHVkaS9Ib29kaWVTdHJlYW1pbmdTaW5rLnNjYWxh) | `68.45% <ø> (-1.25%)` | :arrow_down: |
... and [52 files with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20412/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): reuse one embedded timeline server across streaming sink micro-batches [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #20412:
URL: https://github.com/apache/hudi/pull/20412#discussion_r4374113808
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -533,7 +568,7 @@ class HoodieSparkSqlWriterInternal {
// scalastyle:on
val writeConfig = client.getConfig
- instantTime = client.startCommit(commitActionType)
+ instantTime = recordStartedInstant(client, client.startCommit(commitActionType))
Review Comment:
Addressed in 2c4f041d334695f3bdff640bbf7c8f1f16be67d1. `startCommit` now catches failures while the allocated instant is still known and stops only that instant's heartbeat, preserving the original exception (cleanup errors are suppressed). Four regression cases inject failures before/after persisting commit and replace-commit instants, assert another live heartbeat is unaffected, and reuse the same client for the next commit. The fix has also been carried through #20413 and #20415. On the final stack, all 147 related Java tests passed with JDK 17 and UTC; Checkstyle and RAT passed. The full Spark integration suite was not rerun.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): reuse one embedded timeline server across streaming sink micro-batches [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): reuse one embedded timeline server across streaming sink micro-batches [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): reuse one embedded timeline server across streaming sink micro-batches [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 merged PR #20412:
URL: https://github.com/apache/hudi/pull/20412
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org