This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new e4c9512bee94 feat(spark): stop forcing direct markers in the streaming
sink (#20413)
e4c9512bee94 is described below
commit e4c9512bee94096dd2bcc141ba34436676fe3a0d
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Mon Oct 26 03:18:04 2026 -0700
feat(spark): stop forcing direct markers in the streaming sink (#20413)
* feat(spark): stop forcing direct markers in the streaming sink
The sink forced DIRECT markers because the embedded timeline server used to
be
stopped after the first micro-batch. The sink now keeps one timeline server
for
the whole query, so streaming writes can use the configured marker type
(timeline-server-based by default; WriteMarkersFactory still falls back to
direct markers when the timeline server is disabled).
* Address review: update marker type docs and test the streaming marker path
- Drop the MarkerType.TIMELINE_SERVER_BASED note that structured streaming
always uses direct markers.
- Assert that a micro-batch failing mid-write leaves timeline-server-based
markers, so a reintroduced override is caught.
* Address review: describe the remote-view exception for timeline-server
markers
---
.../org/apache/hudi/common/table/marker/MarkerType.java | 6 ++++--
.../main/scala/org/apache/hudi/HoodieStreamingSink.scala | 7 +++----
.../apache/hudi/functional/TestStructuredStreaming.scala | 13 +++++++++----
3 files changed, 16 insertions(+), 10 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerType.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerType.java
index b0c67a0ea248..dac006d8284e 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerType.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerType.java
@@ -34,7 +34,9 @@ public enum MarkerType {
@EnumFieldDescription("Marker operations are all handled at the timeline
service which serves as a proxy. "
+ "New marker entries are batch processed and stored in a limited number
of underlying files for efficiency. "
- + "If HDFS is used or timeline server is disabled, DIRECT markers are used
as fallback even if this is configured. "
- + "This configuration does not take effect for Spark structured streaming;
DIRECT markers are always used.")
+ + "If HDFS is used, or the embedded timeline server is disabled and the
file system view is not remote, DIRECT markers "
+ + "are used as fallback even if this is configured. With a remote file
system view (REMOTE_FIRST or REMOTE_ONLY) and the "
+ + "embedded timeline server disabled, markers are created through the
configured remote timeline server, which must be "
+ + "reachable; configure DIRECT markers otherwise.")
TIMELINE_SERVER_BASED
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
index 7939cd301852..e4f8f223d4e1 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
@@ -25,7 +25,6 @@ import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.client.embedded.{EmbeddedTimelineServerHelper,
EmbeddedTimelineService}
import org.apache.hudi.common.model.HoodieCommitMetadata
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
-import org.apache.hudi.common.table.marker.MarkerType
import org.apache.hudi.common.table.timeline.HoodieInstant
import org.apache.hudi.common.util.{ClusteringUtils, CommitUtils,
CompactionUtils, Option => HOption}
import org.apache.hudi.common.util.ValidationUtils.checkArgument
@@ -122,10 +121,10 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
- // Override to use direct markers.
- var updatedOptions = options.updated(HoodieWriteConfig.MARKERS_TYPE.key(),
MarkerType.DIRECT.name())
+ // Markers follow hoodie.write.markers.type: the sink keeps one timeline
server running across micro-batches,
+ // so timeline-server-based markers (the default) work for streaming
writes too.
// we need auto adjustment enabled for streaming sink since async table
services are feasible within the same JVM.
- updatedOptions =
updatedOptions.updated(HoodieWriteConfig.AUTO_ADJUST_LOCK_CONFIGS.key, "true")
+ var updatedOptions =
options.updated(HoodieWriteConfig.AUTO_ADJUST_LOCK_CONFIGS.key, "true")
updatedOptions =
updatedOptions.updated(HoodieSparkSqlWriter.SPARK_STREAMING_BATCH_ID,
batchId.toString)
retry(retryCnt, retryIntervalMs)(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala
index 125245adad48..e4fe503622b2 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala
@@ -25,12 +25,13 @@ import org.apache.hudi.common.config.{HoodieStorageConfig,
TypedProperties}
import org.apache.hudi.common.engine.RecordContext
import org.apache.hudi.common.model.{FileSlice,
HoodieFailedWritesCleaningPolicy, HoodieTableType, WriteConcurrencyMode}
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
+import org.apache.hudi.common.table.marker.MarkerType
import org.apache.hudi.common.table.read.BufferedRecord
import org.apache.hudi.common.table.timeline.HoodieTimeline
import org.apache.hudi.common.table.view.{FileSystemViewStorageConfig,
FileSystemViewStorageType}
import org.apache.hudi.common.testutils.{HoodieTestDataGenerator,
HoodieTestTable, HoodieTestUtils}
import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
-import org.apache.hudi.common.util.{CollectionUtils, CommitUtils}
+import org.apache.hudi.common.util.{CollectionUtils, CommitUtils, MarkerUtils,
Option => HOption}
import org.apache.hudi.config.{HoodieCleanConfig, HoodieClusteringConfig,
HoodieCompactionConfig, HoodieLockConfig, HoodieWriteConfig}
import org.apache.hudi.core.transaction.lock.InProcessLockProvider
import org.apache.hudi.exception.{HoodieException, TableNotFoundException}
@@ -593,7 +594,8 @@ class TestStructuredStreaming extends
HoodieSparkClientTestBase {
/**
* 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.
+ * looks alive for as long as the query runs and is never rolled back. The
failed write's markers also show that
+ * the sink uses timeline-server-based markers.
*/
@Test
def testStructuredStreamingReleasesHeartbeatOfFailedBatch(): Unit = {
@@ -622,11 +624,14 @@ class TestStructuredStreaming extends
HoodieSparkClientTestBase {
} finally {
FailingSparkRecordMerger.shouldFail = false
}
- val failedInstant = HoodieTestUtils.createMetaClient(storage,
destPath).getActiveTimeline
- .getCommitsTimeline.filterInflightsAndRequested().lastInstant()
+ val metaClient = HoodieTestUtils.createMetaClient(storage, destPath)
+ val failedInstant =
metaClient.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 sink does not force direct markers: the failed write's markers
went through the sink's timeline server
+ assertEquals(HOption.of(MarkerType.TIMELINE_SERVER_BASED),
+ MarkerUtils.readMarkerType(storage,
metaClient.getMarkerFolderPath(failedInstant.get().requestedTime())))
// the query recovers on the next micro-batch
spark.read.json(spark.sparkContext.parallelize(
