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 9a43cc1fd322 fix(spark): reuse one embedded timeline server across
streaming sink micro-batches (#20412)
9a43cc1fd322 is described below
commit 9a43cc1fd322786d94b758298a35a4468367670e
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Mon Oct 26 00:33:11 2026 -0700
fix(spark): reuse one embedded timeline server across streaming sink
micro-batches (#20412)
* fix(spark): reuse one embedded timeline server across streaming sink
micro-batches
HoodieStreamingSink now owns a single embedded timeline server per streaming
query, hands it to the write client (non-owning, as StreamSync does), keeps
the
client open across micro-batches, and releases both when the query
terminates
via a StreamingQueryListener. Drops the HUDI-3636 override that disabled the
timeline server for streaming writes.
Fixes #20400
* Address review: release failed-write heartbeats and harden sink cleanup
- Stop the heartbeat of a failed write's instant when the caller owns the
write client, so the instant can be rolled back under the lazy
failed-writes policy while the client is reused.
- Match the termination listener on the run id too, and release resources on
a separate thread with a graceful shutdown of async table services.
- Proceed without the embedded timeline server if it fails to start, as a
client starting its own server does.
* Fix streaming sink hang between reset and async service shutdown callbacks
reset() waited for the async compaction/clustering services while holding
the
sink lock, and the services' shutdown callbacks called reset() and blocked
on
that lock, so the service executor never terminated. This hung the query
when
a micro-batch exhausted its retries with an async service running (also on
master), and the graceful release on query termination.
reset() now detaches a service before stopping it; a callback does nothing
for
a detached service and resets the sink on a separate thread when the service
stopped on its own.
* Address review: release failed-instant heartbeats on all write paths
- Record the instant started by each write (upsert/insert, delete, delete
partition, bulk insert as row) and stop its heartbeat on any failure when
the caller owns the write client.
- Register the sink's termination listener before acquiring the timeline
server, so it is released even if the client fails to build.
- Test the reset/async-callback deadlock by driving reset directly with an
async service running, and add a failing delete case to the heartbeat
tests.
* fix(client): stop the heartbeat when commit startup fails
---------
Co-authored-by: danny0405 <[email protected]>
---
.../apache/hudi/client/BaseHoodieWriteClient.java | 13 +-
.../hudi/client/TestBaseHoodieWriteClient.java | 50 +++++
.../main/java/org/apache/hudi/DataSourceUtils.java | 14 +-
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 101 ++++++++---
.../org/apache/hudi/HoodieStreamingSink.scala | 161 +++++++++++++---
.../hudi/functional/TestStructuredStreaming.scala | 202 ++++++++++++++++++++-
6 files changed, 487 insertions(+), 54 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index 004f8a5dce32..d93108c35b1f 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -1152,7 +1152,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
HoodieTimeline.COMMIT_ACTION, () ->
tableServiceClient.rollbackFailedWrites(metaClient));
txnManager.beginStateChange(Option.empty(),
lastCompletedTxnAndMetadata.map(Pair::getLeft));
- String instantTime;
+ String instantTime = null;
HoodieInstant instant = null;
try {
instantTime = providedInstantTime.orElseGet(() ->
createNewInstantTime(false));
@@ -1173,6 +1173,17 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
instant = metaClient.createNewInstant(State.REQUESTED, actionType,
instantTime);
metaClient.getActiveTimeline().createNewInstant(instant);
}
+ } catch (RuntimeException | Error failure) {
+ // The caller does not know the allocated instant until startCommit
returns. In particular, a
+ // failed timeline write must not leave its heartbeat running on a
client reused by streaming.
+ if (instantTime != null && config.getFailedWritesCleanPolicy().isLazy())
{
+ try {
+ heartbeatClient.stop(instantTime);
+ } catch (RuntimeException cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ }
+ throw failure;
} finally {
txnManager.endStateChange(Option.ofNullable(instant));
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java
index 2c7fac462e68..107ad1c2b3e1 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java
@@ -20,9 +20,11 @@ package org.apache.hudi.client;
import org.apache.hudi.callback.common.WriteStatusValidator;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
+import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
import org.apache.hudi.client.transaction.TransactionManager;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTimelineTimeZone;
import org.apache.hudi.common.model.MetaFieldsMode;
@@ -42,6 +44,7 @@ import
org.apache.hudi.common.table.view.FileSystemViewStorageType;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieLockConfig;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -65,6 +68,7 @@ import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.InOrder;
import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
import java.io.IOException;
import java.time.Instant;
@@ -82,6 +86,7 @@ import java.util.stream.Stream;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
@@ -645,6 +650,51 @@ class TestBaseHoodieWriteClient extends
HoodieCommonTestHarness {
inOrder.verify(transactionManager).endStateChange(Option.of(expectedInstant));
}
+ @ParameterizedTest
+ @CsvSource({"commit,false", "commit,true", "replacecommit,false",
"replacecommit,true"})
+ void testFailedStartCommitStopsOnlyItsHeartbeat(String action, boolean
persistBeforeFailure) throws IOException {
+ initMetaClient();
+ HoodieWriteConfig writeConfig =
HoodieWriteConfig.newBuilder().withPath(basePath)
+ .withEmbeddedTimelineServerEnabled(false)
+
.withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+
.withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY).build())
+ .build();
+ HoodieTableMetaClient failingMetaClient = Mockito.spy(metaClient);
+ HoodieActiveTimeline timeline =
Mockito.spy(metaClient.getActiveTimeline());
+ Mockito.doReturn(timeline).when(failingMetaClient).getActiveTimeline();
+ String failedInstant = "20260101000000001";
+ String nextInstant = "20260101000000002";
+ HoodieException failure = new HoodieException("Injected timeline write
failure");
+ Answer<Object> failTimelineWrite = invocation -> {
+
assertTrue(HoodieHeartbeatClient.heartbeatExists(metaClient.getStorage(),
basePath, failedInstant));
+ if (persistBeforeFailure) {
+ invocation.callRealMethod();
+ }
+ throw failure;
+ };
+ if (HoodieTimeline.REPLACE_COMMIT_ACTION.equals(action)) {
+
Mockito.doAnswer(failTimelineWrite).when(timeline).createRequestedCommitWithReplaceMetadata(failedInstant,
action);
+ } else {
+
Mockito.doAnswer(failTimelineWrite).when(timeline).createNewInstant(Mockito.any(HoodieInstant.class));
+ }
+
+ try (TestWriteClient client = new TestWriteClient(writeConfig,
mock(HoodieTable.class), Option.empty(),
+ mock(BaseHoodieTableServiceClient.class),
mock(TransactionManager.class), mock(TimeGenerator.class))) {
+ // A reused client can have another live instant; failure cleanup must
not stop all heartbeats.
+ client.getHeartbeatClient().start(nextInstant);
+ assertSame(failure, assertThrows(HoodieException.class,
+ () -> client.startCommit(Option.of(failedInstant), action,
failingMetaClient)));
+
assertFalse(HoodieHeartbeatClient.heartbeatExists(metaClient.getStorage(),
basePath, failedInstant));
+
assertTrue(HoodieHeartbeatClient.heartbeatExists(metaClient.getStorage(),
basePath, nextInstant));
+ assertEquals(persistBeforeFailure,
metaClient.reloadActiveTimeline().containsInstant(failedInstant));
+
+ assertEquals(nextInstant, client.startCommit(Option.of(nextInstant),
action, metaClient));
+
assertTrue(metaClient.reloadActiveTimeline().containsInstant(nextInstant));
+ client.releaseResources(nextInstant);
+ }
+ }
+
private static class TestWriteClient extends BaseHoodieWriteClient<String,
String, String, String> {
private final HoodieTable<String, String, String, String> table;
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
index 7ec921717258..e5e977941618 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
@@ -63,6 +63,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiFunction;
import java.util.stream.Collectors;
import scala.Tuple2;
@@ -186,6 +187,17 @@ public class DataSourceUtils {
public static SparkRDDWriteClient createHoodieClient(JavaSparkContext jssc,
String schemaStr, String basePath,
String tblName,
Map<String, String> parameters) {
+ return createHoodieClient(jssc, schemaStr, basePath, tblName, parameters,
+ (context, writeConfig) -> new SparkRDDWriteClient<>(context,
writeConfig));
+ }
+
+ /**
+ * Creates a write client, letting the caller construct it from the resolved
write config. Callers that own
+ * resources shared across clients (e.g. an embedded timeline server) use
this to hand them to the client.
+ */
+ public static SparkRDDWriteClient createHoodieClient(JavaSparkContext jssc,
String schemaStr, String basePath,
+ String tblName,
Map<String, String> parameters,
+
BiFunction<HoodieSparkEngineContext, HoodieWriteConfig, SparkRDDWriteClient>
clientFactory) {
HoodieSparkEngineContext context = new HoodieSparkEngineContext(jssc);
HoodieWriteConfig writeConfig = createHoodieConfig(schemaStr, basePath,
tblName, parameters);
// The write options carry the table config of an existing table, so only
a single-field complex keygen table
@@ -198,7 +210,7 @@ public class DataSourceUtils {
KeyGenUtils.recordComplexKeygenEncodingIfMissing(metaClient,
writeConfig);
}
}
- return new SparkRDDWriteClient<>(context, writeConfig);
+ return clientFactory.apply(context, writeConfig);
}
public static HoodieWriteResult doWriteOperation(SparkRDDWriteClient client,
JavaRDD<HoodieRecord> hoodieRecords,
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index 7f9ed3818eb7..d3dd41c29998 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -76,7 +76,7 @@ import org.apache.spark.sql.types.StructType
import org.slf4j.LoggerFactory
import java.util.concurrent.atomic.AtomicBoolean
-import java.util.function.BiConsumer
+import java.util.function.{BiConsumer, BiFunction}
import scala.collection.JavaConverters._
import scala.collection.mutable
@@ -87,7 +87,10 @@ 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)
/**
* Controls whether incoming batch's schema's nullability constraints should
be canonicalized
@@ -165,6 +168,9 @@ class HoodieSparkSqlWriterInternal {
private var tableExists: Boolean = false
private var asyncCompactionTriggerFnDefined: Boolean = false
private var asyncClusteringTriggerFnDefined: Boolean = false
+ private var writeClientOwnedByCaller: Boolean = false
+ // instant started by the current write and the client that started it, to
stop its heartbeat if the write fails
+ private var startedInstant: Option[(SparkRDDWriteClient[_], String)] = None
def write(sqlContext: SQLContext,
mode: SaveMode,
@@ -183,6 +189,7 @@ class HoodieSparkSqlWriterInternal {
while (counter <= maxRetry && !succeeded) {
try {
+ startedInstant = None
toReturn = writeInternal(sqlContext, mode, optParams, sourceDf,
streamingWritesParamsOpt, hoodieWriteClient,
schemaFromCatalog = schemaFromCatalog)
if (counter > 0) {
@@ -191,6 +198,7 @@ class HoodieSparkSqlWriterInternal {
succeeded = true
} catch {
case e: HoodieWriteConflictException =>
+ releaseStartedInstant()
val writeConcurrencyMode =
optParams.getOrElse(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(),
HoodieWriteConfig.WRITE_CONCURRENCY_MODE.defaultValue())
if (WriteConcurrencyMode.supportsMultiWriter(writeConcurrencyMode)
&& counter < maxRetry) {
counter += 1
@@ -198,6 +206,9 @@ class HoodieSparkSqlWriterInternal {
} else {
throw e
}
+ case e: Throwable =>
+ releaseStartedInstant()
+ throw e
}
}
toReturn
@@ -205,6 +216,29 @@ class HoodieSparkSqlWriterInternal {
retryWrite()
}
+ /**
+ * Stops the heartbeat of the instant a failed write started, when the
caller owns the write client: the client
+ * lives on, so its heartbeat would otherwise keep the failed instant
looking alive and block its rollback under
+ * the lazy failed-writes clean policy. A client the writer owns is closed
instead, which stops all its heartbeats.
+ */
+ private def releaseStartedInstant(): Unit = {
+ if (writeClientOwnedByCaller) {
+ startedInstant.foreach { case (client, instant) =>
+ try {
+ client.releaseResources(instant)
+ } catch {
+ case e: Exception => log.warn(s"Failed to release the resources of
instant $instant", e)
+ }
+ }
+ }
+ startedInstant = None
+ }
+
+ private def recordStartedInstant(client: SparkRDDWriteClient[_], instant:
String): String = {
+ startedInstant = Some((client, instant))
+ instant
+ }
+
private def writeInternal(sqlContext: SQLContext,
mode: SaveMode,
optParams: Map[String, String],
@@ -236,6 +270,7 @@ class HoodieSparkSqlWriterInternal {
asyncCompactionTriggerFnDefined =
streamingWritesParamsOpt.map(_.asyncCompactionTriggerFn.isDefined).orElse(Some(false)).get
asyncClusteringTriggerFnDefined =
streamingWritesParamsOpt.map(_.asyncClusteringTriggerFn.isDefined).orElse(Some(false)).get
+ writeClientOwnedByCaller =
streamingWritesParamsOpt.exists(_.writeClientFactory.isDefined)
// re-use table configs and inject defaults.
var (parameters, hoodieConfig) = mergeParamsAndGetHoodieConfig(optParams,
tableConfig, mode, streamingWritesParamsOpt.isDefined)
val databaseName =
hoodieConfig.getStringOrDefault(HoodieTableConfig.DATABASE_NAME, "")
@@ -409,9 +444,9 @@ class HoodieSparkSqlWriterInternal {
// Create a HoodieWriteClient & issue the delete.
val internalSchemaOpt =
HoodieSchemaUtils.getLatestTableInternalSchema(hoodieConfig, tableMetaClient)
- val client =
hoodieWriteClient.getOrElse(DataSourceUtils.createHoodieClient(jsc,
+ val client = hoodieWriteClient.getOrElse(createWriteClient(jsc,
null, path, tblName,
- addSchemaEvolutionParameters(parameters,
internalSchemaOpt).asJava))
+ addSchemaEvolutionParameters(parameters,
internalSchemaOpt).asJava, streamingWritesParamsOpt))
.asInstanceOf[SparkRDDWriteClient[_]]
if (isAsyncCompactionEnabled(client, tableConfig, parameters,
jsc.hadoopConfiguration())) {
@@ -421,7 +456,7 @@ class HoodieSparkSqlWriterInternal {
streamingWritesParamsOpt.map(_.asyncClusteringTriggerFn.get.apply(client))
}
- instantTime = client.startCommit(commitActionType)
+ instantTime = recordStartedInstant(client,
client.startCommit(commitActionType))
// Ingestion setup may have recorded the encoding, and commit
start may have upgraded the table.
tableMetaClient.reloadTableConfig()
val deleteKeyGenProps =
KeyGenUtils.withComplexKeyGenEncoding(TypedProperties.copy(hoodieConfig.getProps),
tableMetaClient.getTableConfig)
@@ -464,12 +499,12 @@ class HoodieSparkSqlWriterInternal {
// Issue the delete.
val schemaStr = new
TableSchemaResolver(tableMetaClient).getTableSchema(false).toString
- val client =
hoodieWriteClient.getOrElse(DataSourceUtils.createHoodieClient(jsc,
+ val client = hoodieWriteClient.getOrElse(createWriteClient(jsc,
schemaStr, path, tblName,
- parameters.asJava))
+ parameters.asJava, streamingWritesParamsOpt))
.asInstanceOf[SparkRDDWriteClient[_]]
// Issue delete partitions
- instantTime = client.startDeletePartitionCommit(tableMetaClient)
+ instantTime = recordStartedInstant(client,
client.startDeletePartitionCommit(tableMetaClient))
val writeResult =
DataSourceUtils.doDeletePartitionsOperation(client, partitionsToDelete,
instantTime)
(writeResult, client)
@@ -514,7 +549,7 @@ class HoodieSparkSqlWriterInternal {
val client = hoodieWriteClient.getOrElse {
val finalOpts = addSchemaEvolutionParameters(parameters,
internalSchemaOpt, Some(writerSchema))
// TODO(HUDI-4772) proper writer-schema has to be specified here
- DataSourceUtils.createHoodieClient(jsc,
processedDataSchema.toString, path, tblName, finalOpts.asJava)
+ createWriteClient(jsc, processedDataSchema.toString, path,
tblName, finalOpts.asJava, streamingWritesParamsOpt)
}
if (isAsyncCompactionEnabled(client, tableConfig, parameters,
jsc.hadoopConfiguration())) {
@@ -533,7 +568,7 @@ class HoodieSparkSqlWriterInternal {
// scalastyle:on
val writeConfig = client.getConfig
- instantTime = client.startCommit(commitActionType)
+ instantTime = recordStartedInstant(client,
client.startCommit(commitActionType))
// Ingestion setup may have recorded the encoding, and commit
start may have upgraded the table.
tableMetaClient.reloadTableConfig()
tableConfig = tableMetaClient.getTableConfig
@@ -561,7 +596,7 @@ class HoodieSparkSqlWriterInternal {
} catch {
case e: HoodieException =>
// close the write client in all cases
- closeWriteClient(client, tableConfig, parameters,
jsc.hadoopConfiguration())
+ closeWriteClient(client, instantTime, tableConfig, parameters,
jsc.hadoopConfiguration())
throw e
}
}
@@ -576,17 +611,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)
+ }
+ } else {
+ val asyncCompactionEnabled = isAsyncCompactionEnabled(writeClient,
tableConfig, parameters, configuration)
+ val asyncClusteringEnabled = isAsyncClusteringEnabled(writeClient,
parameters)
+ if (!asyncCompactionEnabled && !asyncClusteringEnabled) {
+ writeClient.close()
+ }
}
}
@@ -855,7 +914,7 @@ class HoodieSparkSqlWriterInternal {
.orNull
val commitActionType =
CommitUtils.getCommitActionType(overwriteOperationType,
writeConfig.getTableType)
- val instantTime = writeClient.startCommit(commitActionType)
+ val instantTime = recordStartedInstant(writeClient,
writeClient.startCommit(commitActionType))
val executor = mode match {
case _ if overwriteOperationType == null =>
// Don't need to overwrite
@@ -876,7 +935,7 @@ class HoodieSparkSqlWriterInternal {
} catch {
case e: HoodieException =>
// close the write client in all cases
- closeWriteClient(writeClient, tableConfig, parameters,
jsc.hadoopConfiguration())
+ closeWriteClient(writeClient, instantTime, tableConfig, parameters,
jsc.hadoopConfiguration())
throw e
}
@@ -886,7 +945,7 @@ class HoodieSparkSqlWriterInternal {
TableInstantInfo(basePath, instantTime, executor.getCommitActionType,
executor.getWriteOperationType), Option.empty)
(writeSuccessful, HOption.ofNullable(instantTime), compactionInstant,
clusteringInstant, writeClient, tableConfig)
} finally {
- closeWriteClient(writeClient, tableConfig, parameters,
jsc.hadoopConfiguration())
+ closeWriteClient(writeClient, instantTime, tableConfig, parameters,
jsc.hadoopConfiguration())
}
}
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 ce16d82c33a3..7939cd301852 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
@@ -22,11 +22,12 @@ import org.apache.hudi.HoodieStreamingSink.{QUERY_ID_KEY,
SINK_CHECKPOINT_KEY}
import org.apache.hudi.async.{AsyncClusteringService, AsyncCompactService,
SparkStreamingAsyncClusteringService, SparkStreamingAsyncCompactService}
import org.apache.hudi.client.SparkRDDWriteClient
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}
+import org.apache.hudi.common.util.{ClusteringUtils, CommitUtils,
CompactionUtils, Option => HOption}
import org.apache.hudi.common.util.ValidationUtils.checkArgument
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.exception.{HoodieCorruptedDataException,
HoodieException, TableNotFoundException}
@@ -35,9 +36,11 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.spark.api.java.JavaSparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SQLContext}
import org.apache.spark.sql.execution.streaming.Sink
-import org.apache.spark.sql.streaming.OutputMode
+import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryListener}
+import
org.apache.spark.sql.streaming.StreamingQueryListener.{QueryProgressEvent,
QueryStartedEvent, QueryTerminatedEvent}
import org.slf4j.LoggerFactory
+import java.io.IOException
import java.lang
import java.util.function.{BiConsumer, Function}
@@ -79,8 +82,8 @@ class HoodieStreamingSink(sqlContext: SQLContext,
private val disableCompaction =
options.getOrElse(STREAMING_DISABLE_COMPACTION.key,
STREAMING_DISABLE_COMPACTION.defaultValue).toBoolean
- private var isAsyncCompactorServiceShutdownAbnormally = false
- private var isAsyncClusteringServiceShutdownAbnormally = false
+ @volatile private var isAsyncCompactorServiceShutdownAbnormally = false
+ @volatile private var isAsyncClusteringServiceShutdownAbnormally = false
private val mode =
if (outputMode == OutputMode.Append()) {
@@ -89,9 +92,15 @@ class HoodieStreamingSink(sqlContext: SQLContext,
SaveMode.Overwrite
}
- private var asyncCompactorService: AsyncCompactService = _
- private var asyncClusteringService: AsyncClusteringService = _
+ // volatile: read without the sink lock by the services' shutdown callbacks
+ @volatile private var asyncCompactorService: AsyncCompactService = _
+ @volatile private var asyncClusteringService: AsyncClusteringService = _
private var writeClient: Option[SparkRDDWriteClient[_]] = Option.empty
+ // Owned by the sink and handed to every write client it creates, so that
one timeline server is reused across
+ // micro-batches (same as HoodieStreamer). Stopped on reset or when the
streaming query terminates.
+ private var embeddedTimelineService: Option[EmbeddedTimelineService] =
Option.empty
+ private var timelineServiceBasePath: String = _
+ private var queryTerminationListener: Option[StreamingQueryListener] =
Option.empty
private var hoodieTableConfig: Option[HoodieTableConfig] = Option.empty
override def addBatch(batchId: Long, data: DataFrame): Unit =
this.synchronized {
@@ -113,18 +122,11 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
- // Override to use direct markers. In Structured streaming, timeline
server is closed after
- // first micro-batch and subsequent micro-batches do not have timeline
server running.
- // Thus, we can't use timeline-server-based markers.
+ // Override to use direct markers.
var updatedOptions = options.updated(HoodieWriteConfig.MARKERS_TYPE.key(),
MarkerType.DIRECT.name())
// 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")
updatedOptions =
updatedOptions.updated(HoodieSparkSqlWriter.SPARK_STREAMING_BATCH_ID,
batchId.toString)
- if
(!options.contains(HoodieWriteConfig.EMBEDDED_TIMELINE_SERVER_ENABLE.key())) {
- // if user does not explicitly override, we are disabling timeline
server for streaming sink.
- // refer to HUDI-3636 for more details
- updatedOptions =
updatedOptions.updated(HoodieWriteConfig.EMBEDDED_TIMELINE_SERVER_ENABLE.key(),
" false")
- }
retry(retryCnt, retryIntervalMs)(
Try(
@@ -136,7 +138,8 @@ class HoodieStreamingSink(sqlContext: SQLContext,
val identifier =
options.getOrElse(STREAMING_CHECKPOINT_IDENTIFIER.key(),
STREAMING_CHECKPOINT_IDENTIFIER.defaultValue())
newCommitMetadata.addMetadata(SINK_CHECKPOINT_KEY,
CommitUtils.getCheckpointValueAsString(identifier, String.valueOf(batchId)))
}
- }))), writeClient)
+ }),
+ writeClientFactory = Some((context, writeConfig) =>
createWriteClient(context, writeConfig)))), writeClient)
)
match {
case Success((true, commitOps, compactionInstantOps,
clusteringInstant, client, tableConfig)) =>
@@ -233,14 +236,14 @@ class HoodieStreamingSink(sqlContext: SQLContext,
protected def triggerAsyncCompactor(client: SparkRDDWriteClient[_]): Unit = {
if (null == asyncCompactorService) {
log.info("Triggering Async compaction !!")
- asyncCompactorService = new SparkStreamingAsyncCompactService(new
HoodieSparkEngineContext(new JavaSparkContext(sqlContext.sparkContext)),
+ val compactorService = new SparkStreamingAsyncCompactService(new
HoodieSparkEngineContext(new JavaSparkContext(sqlContext.sparkContext)),
client)
- asyncCompactorService.start(new Function[java.lang.Boolean,
java.lang.Boolean] {
+ asyncCompactorService = compactorService
+ compactorService.start(new Function[java.lang.Boolean,
java.lang.Boolean] {
override def apply(errored: lang.Boolean): lang.Boolean = {
log.info(s"Async Compactor shutdown. Errored ? $errored")
isAsyncCompactorServiceShutdownAbnormally = errored
- reset(false)
- log.info("Done resetting write client.")
+ resetOnAsyncServiceStop(asyncCompactorService eq compactorService)
true
}
})
@@ -263,13 +266,14 @@ class HoodieStreamingSink(sqlContext: SQLContext,
protected def triggerAsyncClustering(client: SparkRDDWriteClient[_]): Unit =
{
if (null == asyncClusteringService) {
log.info("Triggering async clustering!")
- asyncClusteringService = new SparkStreamingAsyncClusteringService(new
HoodieSparkEngineContext(new JavaSparkContext(sqlContext.sparkContext)),
+ val clusteringService = new SparkStreamingAsyncClusteringService(new
HoodieSparkEngineContext(new JavaSparkContext(sqlContext.sparkContext)),
client)
- asyncClusteringService.start(new Function[java.lang.Boolean,
java.lang.Boolean] {
+ asyncClusteringService = clusteringService
+ clusteringService.start(new Function[java.lang.Boolean,
java.lang.Boolean] {
override def apply(errored: lang.Boolean): lang.Boolean = {
log.info(s"Async clustering service shutdown. Errored ? $errored")
isAsyncClusteringServiceShutdownAbnormally = errored
- reset(false)
+ resetOnAsyncServiceStop(asyncClusteringService eq clusteringService)
true
}
})
@@ -288,21 +292,128 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
}
- private def reset(force: Boolean): Unit = this.synchronized {
+ /**
+ * 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[_] = {
+ // register the release first, so that a timeline server started for a
client that then fails to build is
+ // still stopped when the query terminates
+ registerQueryTerminationListener()
+ val client = new SparkRDDWriteClient(context, writeConfig,
getOrStartTimelineService(context, writeConfig))
+ // track the client as soon as it exists so that a failed micro-batch
still reuses and eventually closes it
+ writeClient = Some(client)
+ client
+ }
+
+ private def getOrStartTimelineService(context: HoodieSparkEngineContext,
+ writeConfig: HoodieWriteConfig):
HOption[EmbeddedTimelineService] = {
+ if (!writeConfig.isEmbeddedTimelineServerEnabled) {
+ HOption.empty()
+ } else {
+ embeddedTimelineService match {
+ case Some(service) =>
+
EmbeddedTimelineServerHelper.updateWriteConfigWithTimelineServer(service,
writeConfig)
+ case None =>
+ try {
+ embeddedTimelineService =
Some(EmbeddedTimelineServerHelper.createEmbeddedTimelineService(context,
writeConfig))
+ timelineServiceBasePath = writeConfig.getBasePath
+ } catch {
+ case e: IOException =>
+ // same as a client that starts its own server: proceed as if
the embedded server is disabled, and
+ // try again for the next write client
+ log.warn("Unable to start the embedded timeline server.
Proceeding as if it is disabled", e)
+
writeConfig.setValue(HoodieWriteConfig.EMBEDDED_TIMELINE_SERVER_ENABLE, "false")
+ }
+ }
+ embeddedTimelineService.map(service =>
HOption.of(service)).getOrElse(HOption.empty())
+ }
+ }
+
+ /**
+ * A V1 sink has no close hook, so release the write client and timeline
server when the streaming query that
+ * owns this sink terminates. Otherwise they would live until the JVM exits,
which matters for long-lived drivers
+ * (e.g. a Spark Connect server) that run many streaming queries.
+ */
+ private def registerQueryTerminationListener(): Unit = {
+ if (queryTerminationListener.isEmpty) {
+ val queryId = sqlContext.sparkContext.getLocalProperty(QUERY_ID_KEY)
+ // the query id is kept across restarts from the same checkpoint, so
also match the run id: a late termination
+ // event of a previous run must not release the resources of this run
+ val runId = Option(queryId).flatMap(id =>
Option(sqlContext.sparkSession.streams.get(id))).map(_.runId.toString)
+ val listener = new StreamingQueryListener {
+ override def onQueryStarted(event: QueryStartedEvent): Unit = {}
+
+ override def onQueryProgress(event: QueryProgressEvent): Unit = {}
+
+ override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {
+ if (event.id.toString == queryId && runId.forall(_ ==
event.runId.toString)) {
+ // release on a separate thread: the listener bus is shared by all
queries of the session, and a graceful
+ // shutdown waits for in-flight async compaction / clustering
before closing the client they use
+ val cleanup = new Thread(new Runnable {
+ override def run(): Unit = {
+ log.info(s"Streaming query $queryId terminated, closing the
write client and timeline server")
+ reset(false)
+ }
+ }, s"hudi-streaming-sink-cleanup-$queryId")
+ cleanup.setDaemon(true)
+ cleanup.start()
+ }
+ }
+ }
+ sqlContext.sparkSession.streams.addListener(listener)
+ queryTerminationListener = Some(listener)
+ }
+ }
+
+ /**
+ * Called on an async table service's own thread when the service stops.
[[reset]] detaches a service before
+ * shutting it down and waiting for it, so a service that is no longer
attached was stopped by the sink and
+ * needs nothing more. A service that stopped on its own (e.g. it failed)
resets the sink on a separate thread:
+ * resetting here would wait on the sink lock, and for this service's own
executor to terminate.
+ */
+ private def resetOnAsyncServiceStop(stillAttached: Boolean): Unit = {
+ if (stillAttached) {
+ val resetThread = new Thread(new Runnable {
+ override def run(): Unit = {
+ reset(false)
+ log.info("Done resetting write client.")
+ }
+ }, "hudi-streaming-sink-reset")
+ resetThread.setDaemon(true)
+ resetThread.start()
+ }
+ }
+
+ protected def reset(force: Boolean): Unit = this.synchronized {
+ // detach each service before shutting it down, see resetOnAsyncServiceStop
if (asyncCompactorService != null) {
- asyncCompactorService.shutdown(force)
+ val compactorService = asyncCompactorService
asyncCompactorService = null
+ compactorService.shutdown(force)
}
if (asyncClusteringService != null) {
- asyncClusteringService.shutdown(force)
+ val clusteringService = asyncClusteringService
asyncClusteringService = null
+ clusteringService.shutdown(force)
}
if (writeClient.isDefined) {
writeClient.get.close()
writeClient = Option.empty
}
+
+ // stop the timeline server only after the client using it is closed
+ if (embeddedTimelineService.isDefined) {
+ embeddedTimelineService.get.stopForBasePath(timelineServiceBasePath)
+ embeddedTimelineService = Option.empty
+ }
+
+ if (queryTerminationListener.isDefined) {
+
sqlContext.sparkSession.streams.removeListener(queryTerminationListener.get)
+ queryTerminationListener = Option.empty
+ }
}
private def canSkipBatch(incomingBatchId: Long, operationType: String):
Boolean = {
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 ec0c5ff2ac09..125245adad48 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
@@ -17,27 +17,32 @@
package org.apache.hudi.functional
-import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions,
HoodieDataSourceHelpers}
+import org.apache.hudi.{DataSourceReadOptions, DataSourceUtils,
DataSourceWriteOptions, DefaultSparkRecordMerger, HoodieDataSourceHelpers,
HoodieStreamingSink}
import
org.apache.hudi.DataSourceWriteOptions.{STREAMING_CHECKPOINT_IDENTIFIER,
UPSERT_OPERATION_OPT_VAL}
import org.apache.hudi.HoodieStreamingSink.SINK_CHECKPOINT_KEY
-import org.apache.hudi.common.config.HoodieStorageConfig
-import org.apache.hudi.common.model.{FileSlice, HoodieTableType,
WriteConcurrencyMode}
+import org.apache.hudi.client.SparkRDDWriteClient
+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.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.config.{HoodieClusteringConfig, HoodieCompactionConfig,
HoodieLockConfig, HoodieWriteConfig}
+import org.apache.hudi.config.{HoodieCleanConfig, HoodieClusteringConfig,
HoodieCompactionConfig, HoodieLockConfig, HoodieWriteConfig}
import org.apache.hudi.core.transaction.lock.InProcessLockProvider
-import org.apache.hudi.exception.TableNotFoundException
+import org.apache.hudi.exception.{HoodieException, TableNotFoundException}
import org.apache.hudi.storage.{HoodieStorage, StoragePath}
import org.apache.hudi.testutils.{DataSourceTestUtils,
HoodieSparkClientTestBase, HoodieSparkDeleteRecordMerger}
import org.apache.spark.sql._
+import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, Trigger}
import org.apache.spark.sql.types.StructType
import org.junit.jupiter.api.{BeforeEach, Test}
-import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.{CsvSource, EnumSource, ValueSource}
import org.slf4j.LoggerFactory
@@ -98,6 +103,18 @@ class TestStructuredStreaming extends
HoodieSparkClientTestBase {
(sourcePath, destPath)
}
+ /**
+ * Lazy failed-writes cleaning, where failed instants are rolled back once
their heartbeat expires; failed
+ * micro-batches are ignored so that the query keeps running.
+ */
+ private val lazyFailedWritesCleaningOpts = Map(
+ 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")
+
def getOptsWithTableType(tableType: HoodieTableType): Map[String, String] = {
commonOpts + (DataSourceWriteOptions.TABLE_TYPE.key -> tableType.name())
}
@@ -508,6 +525,144 @@ 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())
+ }
+
+ /**
+ * Same as [[testStructuredStreamingReleasesHeartbeatOfFailedBatch]] for a
delete stream, which fails inside the
+ * delete operation: a record without a record key fails the key generation
after the commit started.
+ */
+ @Test
+ def testStructuredStreamingReleasesHeartbeatOfFailedDelete(): Unit = {
+ val (sourcePath, destPath) = initStreamingSourceAndDestPath("source",
"dest")
+ val opts = getOptsWithTableType(HoodieTableType.COPY_ON_WRITE) ++
lazyFailedWritesCleaningOpts
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(
+ recordsToStrings(dataGen.generateInserts("000", 10)).asScala.toList, 1))
+
inputDF.write.format("org.apache.hudi").options(opts).mode(SaveMode.Overwrite).save(destPath)
+
+ inputDF.limit(1).withColumn("_row_key",
lit(null).cast("string")).write.mode(SaveMode.Append).json(sourcePath)
+ val query = initWritingStreamingQuery(inputDF.schema, sourcePath, destPath,
+ opts + (DataSourceWriteOptions.OPERATION.key ->
DataSourceWriteOptions.DELETE_OPERATION_OPT_VAL))
+ try {
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+
+ 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 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
+ ) ++ lazyFailedWritesCleaningOpts
+
+ 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())
+ }
+
+ /**
+ * When a micro-batch exhausts its retries, the sink resets while holding
its lock: it shuts down the async table
+ * services and waits for them. The services' shutdown callbacks must not
block on the sink in turn, otherwise the
+ * reset never finishes and the query hangs. The reset is driven directly
here, with async clustering running,
+ * since a micro-batch that exhausts its retries exits the JVM.
+ */
+ @Test
+ def testSinkResetWithAsyncServiceRunningDoesNotHang(): Unit = {
+ val (_, destPath) = initStreamingSourceAndDestPath("source", "dest")
+ val opts = getOptsWithTableType(HoodieTableType.COPY_ON_WRITE)
+
spark.read.json(spark.sparkContext.parallelize(recordsToStrings(dataGen.generateInserts("000",
10)).asScala.toList, 1))
+
.write.format("org.apache.hudi").options(opts).mode(SaveMode.Overwrite).save(destPath)
+
+ val sink = new ResettableStreamingSink(sqlContext, opts + ("path" ->
destPath))
+ val client = DataSourceUtils.createHoodieClient(jsc,
HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA, destPath,
+ commonOpts(HoodieWriteConfig.TBL_NAME.key), opts.asJava)
+ try {
+ sink.startAsyncClustering(client)
+ Await.result(Future(sink.resetHoldingSinkLock()), Duration(2, "minutes"))
+ } finally {
+ client.close()
+ }
+ }
+
/**
* Injects a failing micro-batch into [[HoodieStreamingSink]] by pointing
the record key at a
* non-existent column so that every Hudi write throws. With
STREAMING_IGNORE_FAILED_BATCH
@@ -614,3 +769,38 @@ class TestStructuredStreaming extends
HoodieSparkClientTestBase {
}
}
}
+
+/**
+ * Event-time merger that fails every merge while
[[FailingSparkRecordMerger.shouldFail]] is set, to fail a write
+ * after its commit started.
+ */
+class FailingSparkRecordMerger extends DefaultSparkRecordMerger {
+ override def getMergingStrategy: String =
FailingSparkRecordMerger.STRATEGY_ID
+
+ override def merge[T](older: BufferedRecord[T], newer: BufferedRecord[T],
recordContext: RecordContext[T],
+ props: TypedProperties): BufferedRecord[T] = {
+ if (FailingSparkRecordMerger.shouldFail) {
+ throw new HoodieException("Injected merge failure")
+ }
+ super.merge(older, newer, recordContext, props)
+ }
+}
+
+object FailingSparkRecordMerger {
+ val STRATEGY_ID = "6b3a2b4e-2f4f-4f0c-9a55-3c3f7d0b8e21"
+ @volatile var shouldFail: Boolean = false
+}
+
+/**
+ * Streaming sink that exposes its async clustering service and reset to tests.
+ */
+class ResettableStreamingSink(sqlContext: SQLContext, options: Map[String,
String])
+ extends HoodieStreamingSink(sqlContext, options, Seq.empty,
OutputMode.Append) {
+
+ def startAsyncClustering(client: SparkRDDWriteClient[_]): Unit =
triggerAsyncClustering(client)
+
+ // a micro-batch that exhausts its retries resets the sink from addBatch,
which holds the sink lock
+ def resetHoldingSinkLock(): Unit = this.synchronized {
+ reset(false)
+ }
+}
