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 53d0b14389bd fix(spark): re-create the streaming sink write client
when the writer schema changes (#20415)
53d0b14389bd is described below
commit 53d0b14389bddd3d753c38d7635c09c21043491d
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Mon Oct 26 19:05:05 2026 -0700
fix(spark): re-create the streaming sink write client when the writer
schema changes (#20415)
* fix(spark): re-create the streaming sink write client when the writer
schema changes
The streaming sink reuses one write client across micro-batches, so the
client's write config keeps the schema of the micro-batch that created it.
When the writer schema changes under a running query (the table schema
evolves and set-null-for-missing-columns / reconcile is on, a type is
widened, or a delete stream sees a new table schema), later writes and their
commit metadata would use the stale schema.
Like HoodieStreamer, re-create the write client when the schema it was
created
with no longer matches the write's schema, for upserts, deletes and delete
partitions. The comparison uses a snapshot taken at client creation rather
than the live write config (compaction updates its schema) and ignores the
internal schema version id. As in HoodieStreamer, the sink hands the new
client to the async table services and closes the old one right away; the
sink-owned timeline server keeps running across the swap.
* fix(client): hand off async write clients between completed runs
---------
Co-authored-by: danny0405 <[email protected]>
---
.../apache/hudi/async/AsyncClusteringService.java | 12 +-
.../org/apache/hudi/async/AsyncCompactService.java | 13 +-
.../async/TestAsyncTableServiceClientHandoff.java | 152 +++++++++++++++++++++
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 85 ++++++++++--
.../org/apache/hudi/HoodieStreamingSink.scala | 38 +++++-
.../hudi/functional/TestStructuredStreaming.scala | 87 +++++++++++-
6 files changed, 362 insertions(+), 25 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
index eebfa9303eb9..236a6c10e964 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
@@ -76,7 +76,9 @@ public abstract class AsyncClusteringService extends
HoodieAsyncTableService {
final String instant = fetchNextAsyncServiceInstant();
if (null != instant) {
log.info("Starting clustering for instant {}", instant);
- clusteringClient.cluster(instant);
+ synchronized (clusteringClient) {
+ clusteringClient.cluster(instant);
+ }
log.info("Finished clustering for instant {}", instant);
}
}
@@ -98,9 +100,11 @@ public abstract class AsyncClusteringService extends
HoodieAsyncTableService {
}
/**
- * Update the write client to be used for clustering.
+ * Switches clients between complete clustering runs. Once this returns, no
run uses the old client.
*/
- public synchronized void updateWriteClient(BaseHoodieWriteClient
writeClient) {
- this.clusteringClient.updateWriteClient(writeClient);
+ public void updateWriteClient(BaseHoodieWriteClient writeClient) {
+ synchronized (clusteringClient) {
+ clusteringClient.updateWriteClient(writeClient);
+ }
}
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
index 9a5106c02151..6fb6514e56ec 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
@@ -78,7 +78,9 @@ public abstract class AsyncCompactService extends
HoodieAsyncTableService {
if (null != instantTime) {
log.info("Starting Compaction for instant {}", instantTime);
- compactor.compact(instantTime);
+ synchronized (compactor) {
+ compactor.compact(instantTime);
+ }
log.info("Finished Compaction for instant {}", instantTime);
}
}
@@ -108,7 +110,12 @@ public abstract class AsyncCompactService extends
HoodieAsyncTableService {
return false;
}
- public synchronized void updateWriteClient(BaseHoodieWriteClient
writeClient) {
- this.compactor.updateWriteClient(writeClient);
+ /**
+ * Switches clients between complete compaction runs. Once this returns, no
run uses the old client.
+ */
+ public void updateWriteClient(BaseHoodieWriteClient writeClient) {
+ synchronized (compactor) {
+ compactor.updateWriteClient(writeClient);
+ }
}
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceClientHandoff.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceClientHandoff.java
new file mode 100644
index 000000000000..3d8a8543ae7c
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceClientHandoff.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.async;
+
+import org.apache.hudi.client.BaseHoodieWriteClient;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.mockito.stubbing.Answer;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+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.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestAsyncTableServiceClientHandoff {
+
+ @ParameterizedTest
+ @CsvSource({"true,false", "false,false", "true,true", "false,true"})
+ void testHandoffWaitsForActiveRun(boolean compact, boolean failRun) throws
Exception {
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ BaseHoodieWriteClient oldClient = mock(BaseHoodieWriteClient.class);
+ BaseHoodieWriteClient newClient = mock(BaseHoodieWriteClient.class);
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ when(oldClient.getConfig()).thenReturn(config);
+ when(config.areTableServicesEnabled()).thenReturn(true);
+ AtomicReference<BaseHoodieWriteClient> currentClient = new
AtomicReference<>(oldClient);
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch finishRun = new CountDownLatch(1);
+ CountDownLatch nextRun = new CountDownLatch(1);
+ CountDownLatch stopped = new CountDownLatch(1);
+ Answer<Void> run = invocation -> {
+ BaseHoodieWriteClient capturedClient = currentClient.get();
+ if ("001".equals(invocation.getArgument(0))) {
+ entered.countDown();
+ assertTrue(finishRun.await(10, TimeUnit.SECONDS));
+ assertSame(oldClient, capturedClient);
+ verify(oldClient, never()).close();
+ if (failRun) {
+ throw new IOException("Injected async operation failure");
+ }
+ } else {
+ assertSame(newClient, capturedClient);
+ nextRun.countDown();
+ }
+ return null;
+ };
+ Answer<Void> update = invocation -> {
+ currentClient.set(invocation.getArgument(0));
+ return null;
+ };
+ HoodieAsyncService service;
+ Consumer<BaseHoodieWriteClient> handoff;
+ if (compact) {
+ BaseCompactor compactor = mock(BaseCompactor.class);
+ doAnswer(run).when(compactor).compact(anyString());
+ doAnswer(update).when(compactor).updateWriteClient(any());
+ AsyncCompactService compaction = new AsyncCompactService(context,
oldClient, true) {
+ @Override
+ protected BaseCompactor createCompactor(BaseHoodieWriteClient client) {
+ return compactor;
+ }
+ };
+ service = compaction;
+ handoff = compaction::updateWriteClient;
+ } else {
+ BaseClusterer clusterer = mock(BaseClusterer.class);
+ doAnswer(run).when(clusterer).cluster(anyString());
+ doAnswer(update).when(clusterer).updateWriteClient(any());
+ AsyncClusteringService clustering = new AsyncClusteringService(context,
oldClient, true) {
+ @Override
+ protected BaseClusterer createClusteringClient(BaseHoodieWriteClient
client) {
+ return clusterer;
+ }
+ };
+ service = clustering;
+ handoff = clustering::updateWriteClient;
+ }
+
+ ExecutorService swapExecutor = Executors.newSingleThreadExecutor();
+ try {
+ service.start(errored -> {
+ stopped.countDown();
+ return true;
+ });
+ service.enqueuePendingAsyncServiceInstant("001");
+ assertTrue(entered.await(10, TimeUnit.SECONDS));
+ CountDownLatch swapping = new CountDownLatch(1);
+ Future<?> swap = swapExecutor.submit(() -> {
+ swapping.countDown();
+ handoff.accept(newClient);
+ oldClient.close();
+ });
+ assertTrue(swapping.await(10, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> swap.get(200,
TimeUnit.MILLISECONDS));
+ verify(oldClient, never()).close();
+
+ finishRun.countDown();
+ swap.get(10, TimeUnit.SECONDS);
+ verify(oldClient).close();
+ if (failRun) {
+ assertTrue(stopped.await(10, TimeUnit.SECONDS));
+ assertTrue(service.hasError());
+ } else {
+ service.enqueuePendingAsyncServiceInstant("002");
+ assertTrue(nextRun.await(10, TimeUnit.SECONDS));
+ // An idle service must hand back the client promptly too, without
waiting for another run.
+ swapExecutor.submit(() -> handoff.accept(oldClient)).get(10,
TimeUnit.SECONDS);
+ }
+ } finally {
+ finishRun.countDown();
+ service.shutdown(true);
+ swapExecutor.shutdownNow();
+ assertTrue(swapExecutor.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+}
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 d3dd41c29998..9d1c2a6168d6 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
@@ -23,7 +23,7 @@ import
org.apache.hudi.DataSourceUtils.SparkDataSourceWriteStatusValidator
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.HoodieConversionUtils.{toProperties, toScalaOption}
import
org.apache.hudi.HoodieSchemaConversionUtils.{convertStructTypeToHoodieSchema,
getRecordNameAndNamespace}
-import org.apache.hudi.HoodieSparkSqlWriter.StreamingWriteParams
+import org.apache.hudi.HoodieSparkSqlWriter.{StreamingWriteParams,
WriteClientSchema}
import org.apache.hudi.HoodieSparkSqlWriterInternal.{handleInsertDuplicates,
shouldDropDuplicatesForInserts, shouldFailWhenDuplicatesFound}
import org.apache.hudi.HoodieWriterUtils._
import org.apache.hudi.client.{HoodieWriteResult, SparkRDDWriteClient}
@@ -90,7 +90,21 @@ object HoodieSparkSqlWriter {
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)
+ writeClientFactory:
Option[(HoodieSparkEngineContext, HoodieWriteConfig) => SparkRDDWriteClient[_]]
= Option.empty,
+ // Schema the caller-owned write client was
created with. The client is reused only while
+ // it matches the schema of the write,
otherwise the factory creates a new one.
+ writeClientSchema: Option[WriteClientSchema]
= Option.empty)
+
+ /**
+ * Writer schema and internal schema of a write config, captured when a
write client is created. The write config
+ * itself is not a stable record of them, since table services running on
the client update its schema.
+ */
+ case class WriteClientSchema(schema: String, internalSchema: String)
+
+ object WriteClientSchema {
+ def of(writeConfig: HoodieWriteConfig): WriteClientSchema =
+ WriteClientSchema(StringUtils.nullToEmpty(writeConfig.getSchema),
StringUtils.nullToEmpty(writeConfig.getInternalSchema))
+ }
/**
* Controls whether incoming batch's schema's nullability constraints should
be canonicalized
@@ -444,9 +458,16 @@ class HoodieSparkSqlWriterInternal {
// Create a HoodieWriteClient & issue the delete.
val internalSchemaOpt =
HoodieSchemaUtils.getLatestTableInternalSchema(hoodieConfig, tableMetaClient)
- val client = hoodieWriteClient.getOrElse(createWriteClient(jsc,
- null, path, tblName,
- addSchemaEvolutionParameters(parameters,
internalSchemaOpt).asJava, streamingWritesParamsOpt))
+ val deleteOpts = addSchemaEvolutionParameters(parameters,
internalSchemaOpt)
+ // A client that is reused across writes gets the latest table
schema upfront, so that a later change of the
+ // table schema is detected. Otherwise the client sets it on its
first delete.
+ val deleteSchemaStr = if (writeClientOwnedByCaller) {
+ new
TableSchemaResolver(tableMetaClient).getTableSchemaIfPresent(false).map[String](s
=> s.toString).orElse(null)
+ } else {
+ null
+ }
+ val client = reusableWriteClient(hoodieWriteClient,
streamingWritesParamsOpt, deleteSchemaStr, deleteOpts)
+ .getOrElse(createWriteClient(jsc, deleteSchemaStr, path,
tblName, deleteOpts.asJava, streamingWritesParamsOpt))
.asInstanceOf[SparkRDDWriteClient[_]]
if (isAsyncCompactionEnabled(client, tableConfig, parameters,
jsc.hadoopConfiguration())) {
@@ -499,9 +520,8 @@ class HoodieSparkSqlWriterInternal {
// Issue the delete.
val schemaStr = new
TableSchemaResolver(tableMetaClient).getTableSchema(false).toString
- val client = hoodieWriteClient.getOrElse(createWriteClient(jsc,
- schemaStr, path, tblName,
- parameters.asJava, streamingWritesParamsOpt))
+ val client = reusableWriteClient(hoodieWriteClient,
streamingWritesParamsOpt, schemaStr, parameters)
+ .getOrElse(createWriteClient(jsc, schemaStr, path, tblName,
parameters.asJava, streamingWritesParamsOpt))
.asInstanceOf[SparkRDDWriteClient[_]]
// Issue delete partitions
instantTime = recordStartedInstant(client,
client.startDeletePartitionCommit(tableMetaClient))
@@ -546,11 +566,12 @@ class HoodieSparkSqlWriterInternal {
}
// Create a HoodieWriteClient & issue the write.
- val client = hoodieWriteClient.getOrElse {
- val finalOpts = addSchemaEvolutionParameters(parameters,
internalSchemaOpt, Some(writerSchema))
- // TODO(HUDI-4772) proper writer-schema has to be specified here
- createWriteClient(jsc, processedDataSchema.toString, path,
tblName, finalOpts.asJava, streamingWritesParamsOpt)
- }
+ val finalOpts = addSchemaEvolutionParameters(parameters,
internalSchemaOpt, Some(writerSchema))
+ val client = reusableWriteClient(hoodieWriteClient,
streamingWritesParamsOpt, processedDataSchema.toString, finalOpts)
+ .getOrElse {
+ // TODO(HUDI-4772) proper writer-schema has to be specified
here
+ createWriteClient(jsc, processedDataSchema.toString, path,
tblName, finalOpts.asJava, streamingWritesParamsOpt)
+ }
if (isAsyncCompactionEnabled(client, tableConfig, parameters,
jsc.hadoopConfiguration())) {
streamingWritesParamsOpt.map(_.asyncCompactionTriggerFn.get.apply(client))
@@ -631,6 +652,44 @@ class HoodieSparkSqlWriterInternal {
}
}
+ /**
+ * Returns the write client passed in by the caller if it can serve a write
with the given schema. A caller-owned
+ * client (streaming sink) is reused across writes, but carries the schema
of the write that created it: once the
+ * schema changed (e.g. the table schema evolved under a streaming query),
the caller's factory creates a new
+ * client instead, as HoodieStreamer does on a schema change. Otherwise the
write and its commit metadata would use
+ * the stale schema.
+ */
+ private def reusableWriteClient(hoodieWriteClient:
Option[SparkRDDWriteClient[_]],
+ streamingWritesParamsOpt:
Option[StreamingWriteParams],
+ schemaStr: String,
+ writeOpts: Map[String, String]):
Option[SparkRDDWriteClient[_]] = {
+ if (!writeClientOwnedByCaller) {
+ hoodieWriteClient
+ } else {
+ hoodieWriteClient.filter { _ =>
+ val requiredSchema =
WriteClientSchema(StringUtils.nullToEmpty(schemaStr),
+
StringUtils.nullToEmpty(writeOpts.getOrElse(HoodieWriteConfig.INTERNAL_SCHEMA_STRING.key,
null)))
+ val clientSchema =
streamingWritesParamsOpt.flatMap(_.writeClientSchema)
+ val reusable = clientSchema.exists(current => current.schema ==
requiredSchema.schema
+ && isSameInternalSchema(current.internalSchema,
requiredSchema.internalSchema))
+ if (!reusable) {
+ log.info(s"Writer schema changed (schema hash
${clientSchema.map(_.schema.hashCode)} -> "
+ + s"${requiredSchema.schema.hashCode}), re-creating the write
client")
+ }
+ reusable
+ }
+ }
+ }
+
+ private def isSameInternalSchema(current: String, required: String): Boolean
= {
+ // the version id of an internal schema is set to the instant that saved
it, so it changes on every commit
+ current == required || {
+ val currentSchema = SerDeHelper.fromJson(current)
+ val requiredSchema = SerDeHelper.fromJson(required)
+ currentSchema.isPresent && requiredSchema.isPresent &&
currentSchema.get.equalsIgnoringVersion(requiredSchema.get)
+ }
+ }
+
private def closeWriteClient(writeClient: SparkRDDWriteClient[_],
instantTime: String, tableConfig: HoodieTableConfig,
parameters: Map[String, String], configuration:
Configuration): Unit = {
if (writeClientOwnedByCaller) {
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 e4f8f223d4e1..237c800fba60 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
@@ -17,7 +17,7 @@
package org.apache.hudi
import org.apache.hudi.DataSourceWriteOptions._
-import org.apache.hudi.HoodieSparkSqlWriter.StreamingWriteParams
+import org.apache.hudi.HoodieSparkSqlWriter.{StreamingWriteParams,
WriteClientSchema}
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
@@ -95,6 +95,8 @@ class HoodieStreamingSink(sqlContext: SQLContext,
@volatile private var asyncCompactorService: AsyncCompactService = _
@volatile private var asyncClusteringService: AsyncClusteringService = _
private var writeClient: Option[SparkRDDWriteClient[_]] = Option.empty
+ // schema the current write client was created with; the writer re-creates
the client when it changes
+ private var writeClientSchema: Option[WriteClientSchema] = 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
@@ -138,7 +140,8 @@ class HoodieStreamingSink(sqlContext: SQLContext,
newCommitMetadata.addMetadata(SINK_CHECKPOINT_KEY,
CommitUtils.getCheckpointValueAsString(identifier, String.valueOf(batchId)))
}
}),
- writeClientFactory = Some((context, writeConfig) =>
createWriteClient(context, writeConfig)))), writeClient)
+ writeClientFactory = Some((context, writeConfig) =>
createWriteClient(context, writeConfig)),
+ writeClientSchema = writeClientSchema)), writeClient)
)
match {
case Success((true, commitOps, compactionInstantOps,
clusteringInstant, client, tableConfig)) =>
@@ -299,12 +302,40 @@ class HoodieStreamingSink(sqlContext: SQLContext,
// 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 clientSchema = WriteClientSchema.of(writeConfig)
val client = new SparkRDDWriteClient(context, writeConfig,
getOrStartTimelineService(context, writeConfig))
+ val previousClient = writeClient
// track the client as soon as it exists so that a failed micro-batch
still reuses and eventually closes it
writeClient = Some(client)
+ writeClientSchema = Some(clientSchema)
+ // the writer re-creates the client when the write schema changes (as
HoodieStreamer does): hand the new client
+ // to the async table services and close the previous one. Its timeline
server is the sink's and keeps running.
+ previousClient.foreach(retireWriteClient(_, client))
client
}
+ /**
+ * Each handoff waits for the service's current run before switching to the
new client. Close the previous
+ * client only after both services have handed it back, without retaining
clients until query termination.
+ */
+ private def retireWriteClient(previousClient: SparkRDDWriteClient[_],
newClient: SparkRDDWriteClient[_]): Unit = {
+ if (asyncCompactorService != null) {
+ asyncCompactorService.updateWriteClient(newClient)
+ }
+ if (asyncClusteringService != null) {
+ asyncClusteringService.updateWriteClient(newClient)
+ }
+ closeQuietly(previousClient)
+ }
+
+ private def closeQuietly(client: SparkRDDWriteClient[_]): Unit = {
+ try {
+ client.close()
+ } catch {
+ case e: Exception => log.warn("Failed to close a write client of the
streaming sink", e)
+ }
+ }
+
private def getOrStartTimelineService(context: HoodieSparkEngineContext,
writeConfig: HoodieWriteConfig):
HOption[EmbeddedTimelineService] = {
if (!writeConfig.isEmbeddedTimelineServerEnabled) {
@@ -401,9 +432,10 @@ class HoodieStreamingSink(sqlContext: SQLContext,
if (writeClient.isDefined) {
writeClient.get.close()
writeClient = Option.empty
+ writeClientSchema = Option.empty
}
- // stop the timeline server only after the client using it is closed
+ // stop the timeline server only after the clients using it are closed
if (embeddedTimelineService.isDefined) {
embeddedTimelineService.get.stopForBasePath(timelineServiceBasePath)
embeddedTimelineService = Option.empty
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 e4fe503622b2..e4c95355e917 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
@@ -24,7 +24,7 @@ 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.{HoodieTableConfig, HoodieTableMetaClient,
TableSchemaResolver}
import org.apache.hudi.common.table.marker.MarkerType
import org.apache.hudi.common.table.read.BufferedRecord
import org.apache.hudi.common.table.timeline.HoodieTimeline
@@ -39,7 +39,7 @@ 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.functions.{col, lit}
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, Trigger}
import org.apache.spark.sql.types.StructType
import org.junit.jupiter.api.{BeforeEach, Test}
@@ -563,6 +563,89 @@ class TestStructuredStreaming extends
HoodieSparkClientTestBase {
assertEquals(3,
HoodieDataSourceHelpers.allCompletedCommitsCompactions(storage,
destPath).countInstants())
}
+ /**
+ * The sink reuses its write client across micro-batches, so it must
re-create the client when the writer schema
+ * changes under a running query, as HoodieStreamer does. Here a batch
writer adds a column between two
+ * micro-batches; the next micro-batch fills it with nulls and must commit
with the evolved schema.
+ */
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieTableType])
+ def testStructuredStreamingWithTableSchemaEvolvedMidQuery(tableType:
HoodieTableType): Unit = {
+ val (sourcePath, destPath) = initStreamingSourceAndDestPath("source",
"dest")
+ val opts = getOptsWithTableType(tableType) ++ Map(
+ DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.key -> "true")
+
+ 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)
+ val evolvedDF = spark.read.json(spark.sparkContext.parallelize(
+ recordsToStrings(dataGen.generateUniqueUpdates("001",
10)).asScala.toList, 2))
+ .withColumn("new_col", lit("evolved"))
+ val streamedDF = spark.read.json(spark.sparkContext.parallelize(
+ recordsToStrings(dataGen.generateUniqueUpdates("002",
50)).asScala.toList, 2))
+ try {
+ query.processAllAvailable()
+ // evolve the table schema outside of the streaming query
+ evolvedDF.write.format("org.apache.hudi").options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append).save(destPath)
+ streamedDF.coalesce(1).write.mode(SaveMode.Append).json(sourcePath)
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+
+ val metaClient = HoodieTestUtils.createMetaClient(storage, destPath)
+ assertEquals(3,
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().countInstants())
+ val tableSchema = new TableSchemaResolver(metaClient).getTableSchema(false)
+ assertTrue(tableSchema.getField("new_col").isPresent, s"new_col missing
from table schema: $tableSchema")
+ val tableDF = spark.read.format("org.apache.hudi").load(destPath)
+ assertEquals(100, tableDF.count())
+ // rows last written by the batch writer keep their value, rows the stream
updated afterwards are null
+ val evolvedOnlyKeys =
evolvedDF.select("_row_key").except(streamedDF.select("_row_key")).count()
+ assertEquals(evolvedOnlyKeys, tableDF.filter(col("new_col") ===
"evolved").count())
+ }
+
+ /**
+ * Same as [[testStructuredStreamingWithTableSchemaEvolvedMidQuery]] for a
delete stream: delete commits record
+ * the schema of the write client, so a stale client would drop the new
column from the table schema.
+ */
+ @Test
+ def testStructuredStreamingDeletesWithTableSchemaEvolvedMidQuery(): Unit = {
+ val (sourcePath, destPath) = initStreamingSourceAndDestPath("source",
"dest")
+ val opts = getOptsWithTableType(HoodieTableType.COPY_ON_WRITE)
+ val records = recordsToStrings(dataGen.generateInserts("000",
100)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+
inputDF.write.format("org.apache.hudi").options(opts).mode(SaveMode.Overwrite).save(destPath)
+
+ def recordsDF(from: Int, until: Int): DataFrame =
+
spark.read.schema(inputDF.schema).json(spark.sparkContext.parallelize(records.slice(from,
until), 1))
+
+ recordsDF(0, 10).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()
+ // evolve the table schema outside of the streaming query
+ recordsDF(20, 30).withColumn("new_col",
lit("evolved")).write.format("org.apache.hudi").options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append).save(destPath)
+ recordsDF(10, 20).write.mode(SaveMode.Append).json(sourcePath)
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+
+ val metaClient = HoodieTestUtils.createMetaClient(storage, destPath)
+ assertEquals(4,
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().countInstants())
+ val tableSchema = new TableSchemaResolver(metaClient).getTableSchema(false)
+ assertTrue(tableSchema.getField("new_col").isPresent, s"new_col missing
from table schema: $tableSchema")
+ val tableDF = spark.read.format("org.apache.hudi").load(destPath)
+ assertEquals(80, tableDF.count())
+ assertEquals(10, tableDF.filter(col("new_col") === "evolved").count())
+ }
+
/**
* 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.
