This is an automated email from the ASF dual-hosted git repository.
yihua 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 f2e0b661441e feat(utilities): add fail-fast continuous mode for
HoodieMultiTableStreamer (#19811)
f2e0b661441e is described below
commit f2e0b661441e6fada9f54d40e0055c4ebc81059e
Author: Srikanth Mamidala <[email protected]>
AuthorDate: Tue Oct 27 14:13:09 2026 -0500
feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer
(#19811)
In continuous mode each table's sync() blocks until shutdown, so the
existing sequential loop would only ever sync the first table. Sync the
tables concurrently (one thread per table) when continuous mode is on,
while keeping the sequential path for run-once mode.
Add a --fail-fast-on-continuous flag: when enabled, the first table
failure shuts down the sibling streamers and throws a HoodieException so
the job exits non-zero; when disabled (default), each table is synced
independently and a single failure does not affect the others.
Use thread-safe collections for the success/failed table tracking now
that syncs run in parallel.
---
.../streamer/HoodieMultiTableStreamer.java | 237 ++++++++++++++++++++-
.../hudi/utilities/streamer/HoodieStreamer.java | 14 ++
.../TestHoodieMultiTableDeltaStreamer.java | 165 ++++++++++++++
.../utilities/sources/ContinuousTestSource.java | 156 ++++++++++++++
4 files changed, 562 insertions(+), 10 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java
index fc8de0d22792..39138973fba1 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java
@@ -23,6 +23,8 @@ import org.apache.hudi.SparkAdapterSupport$;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.util.CustomizedThreadFactory;
+import org.apache.hudi.common.util.FutureUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
@@ -51,10 +53,18 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
import static
org.apache.hudi.utilities.config.HoodieSchemaProviderConfig.SCHEMA_REGISTRY_BASE_URL;
@@ -79,12 +89,19 @@ public class HoodieMultiTableStreamer {
private transient JavaSparkContext jssc;
private final Set<String> successTables;
private final Set<String> failedTables;
+ @Getter(AccessLevel.NONE)
+ private final boolean failFastOnContinuousMode;
+ @Getter(AccessLevel.NONE)
+ private final boolean continuousMode;
public HoodieMultiTableStreamer(Config config, JavaSparkContext jssc) throws
IOException {
this.tableExecutionContexts = new ArrayList<>();
- this.successTables = new HashSet<>();
- this.failedTables = new HashSet<>();
+ // In continuous mode tables are synced concurrently, so use thread-safe
sets.
+ this.successTables = ConcurrentHashMap.newKeySet();
+ this.failedTables = ConcurrentHashMap.newKeySet();
this.jssc = jssc;
+ this.failFastOnContinuousMode = config.failFastOnContinuousMode;
+ this.continuousMode = config.continuousMode;
String commonPropsFile = config.propsFilePath;
String configFolder = config.configFolder;
ValidationUtils.checkArgument(!config.filterDupes || config.operation !=
WriteOperationType.UPSERT,
@@ -389,6 +406,10 @@ public class HoodieMultiTableStreamer {
+ " source-fetch -> Transform -> Hudi Write in loop")
public Boolean continuousMode = false;
+ @Parameter(names = {"--fail-fast-on-continuous"},
+ description = "In continuous mode, stop all table syncs on the first
table failure. Default: false")
+ public Boolean failFastOnContinuousMode = false;
+
@Parameter(names = {"--min-sync-interval-seconds"},
description = "the min sync interval of each sync in continuous mode")
public Integer minSyncIntervalSeconds = 0;
@@ -461,28 +482,222 @@ public class HoodieMultiTableStreamer {
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * <p>In continuous mode each table's sync blocks until it is shut down, so
the tables are synced concurrently.
+ * Otherwise the tables are synced sequentially, one after another.
*/
public void sync() {
+ try {
+ if (continuousMode) {
+ syncContinuously();
+ } else {
+ syncSequentially();
+ }
+ } finally {
+ log.info("Ingestion was successful for topics: {}", successTables);
+ if (!failedTables.isEmpty()) {
+ log.error("Ingestion failed for topics: {}", failedTables);
+ }
+ }
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
+ String table = Helpers.getTableWithDatabase(context);
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
- successTables.add(Helpers.getTableWithDatabase(context));
- streamer.shutdownGracefully();
+ successTables.add(table);
} catch (Exception e) {
- log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
- failedTables.add(Helpers.getTableWithDatabase(context));
+ log.error("error while running MultiTableDeltaStreamer for table: {}",
table, e);
+ failedTables.add(table);
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, table);
}
}
}
+ }
- log.info("Ingestion was successful for topics: {}", successTables);
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * <p>When {@code --fail-fast-on-continuous} is enabled, the first table
failure tears the sibling streamers down
+ * at once. They are interrupted mid-round rather than allowed to finish it,
so a table can be left with an inflight
+ * instant that is rolled back on the next run. Otherwise every table is
synced independently and a single failure
+ * does not affect the others.
+ *
+ * <p>Either way, a {@link HoodieException} is thrown if the run ends with a
failed table, so the job exits with a
+ * non-zero status. Note that the run only ends once every table has
stopped: a table failing while the others keep
+ * running does not end it, and surfaces through that table's error log and
metrics rather than the exit code.
+ *
+ * <p>Teardown runs in a {@code finally} rather than a catch so that it also
covers an {@link Error}, which the
+ * workers do not catch, and it is a no-op once a table has shut its own
ingestion service down. The siblings are
+ * interrupted first, then waited on, so this does not return while a table
is still writing and {@code main()}
+ * stops the shared Spark context under it.
+ */
+ private void syncContinuously() {
+ if (tableExecutionContexts.isEmpty()) {
+ return;
+ }
+ warnIfSchedulerIsNotFair();
+ // Streamer instances are registered from worker threads, so a thread-safe
list is required.
+ final List<HoodieStreamer> streamerInstances = new
CopyOnWriteArrayList<>();
+ // Set once fail fast trips, so tasks that register their streamer
afterwards stop before starting the sync.
+ final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
+ final ExecutorService executor =
Executors.newFixedThreadPool(tableExecutionContexts.size(),
+ new CustomizedThreadFactory("multi-table-streamer", true));
+ try {
+ final List<CompletableFuture<Void>> tableFutures =
tableExecutionContexts.stream()
+ .map(context -> CompletableFuture.runAsync(
+ () -> runTableSync(context, streamerInstances,
shutdownRequested), executor))
+ .collect(Collectors.toList());
+
+ if (failFastOnContinuousMode) {
+ log.info("Fail fast enabled in continuous mode. The whole job fails on
any single table failure");
+ awaitFailFast(tableFutures);
+ } else {
+ CompletableFuture.allOf(tableFutures.toArray(new
CompletableFuture[0])).join();
+ }
+ } finally {
+ // FutureUtils.allOf only cancels the futures, so the siblings are still
ingesting on an abnormal exit.
+ shutdownRequested.set(true);
+ interruptAllIngestion(streamerInstances);
+ // Logs rather than throws. The failure path is already propagating its
own exception.
+ shutdownExecutor(executor);
+ }
+ // Reached only when nothing was rethrown above. Continuous mode is not
meant to end, so returning here with a
+ // failed table would exit 0 and tell an orchestrator that nothing is
wrong while nothing is ingesting.
if (!failedTables.isEmpty()) {
- log.info("Ingestion failed for topics: {}", failedTables);
+ throw new HoodieException("Continuous mode ended with failed tables: " +
failedTables);
+ }
+ }
+
+ /**
+ * Warns when several tables share a SparkContext that schedules FIFO, where
one table's job holds the cluster
+ * until it finishes and the others wait behind it. FAIR has to be set at
submit time to interleave them.
+ */
+ private void warnIfSchedulerIsNotFair() {
+ if (tableExecutionContexts.size() < 2) {
+ return;
+ }
+ String schedulerMode =
jssc.getConf().get(SchedulerConfGenerator.SPARK_SCHEDULER_MODE_KEY, "FIFO");
+ if
(!SchedulerConfGenerator.SPARK_SCHEDULER_FAIR_MODE.equalsIgnoreCase(schedulerMode))
{
+ log.warn("Syncing {} tables concurrently on a SparkContext with {}={}.
One table's job will hold the cluster "
+ + "until it completes while the others queue behind it; set {}=FAIR
at submit time to interleave them.",
+ tableExecutionContexts.size(),
SchedulerConfGenerator.SPARK_SCHEDULER_MODE_KEY, schedulerMode,
+ SchedulerConfGenerator.SPARK_SCHEDULER_MODE_KEY);
+ }
+ }
+
+ /**
+ * Syncs one table on the calling worker thread. Rethrows only under fail
fast; otherwise the failure is recorded
+ * in {@link #failedTables} and the sibling tables carry on.
+ */
+ private void runTableSync(TableExecutionContext context,
List<HoodieStreamer> streamerInstances, AtomicBoolean shutdownRequested) {
+ String table = Helpers.getTableWithDatabase(context);
+ // The tables now log concurrently into one driver log, so name the worker
after the table it is syncing.
+ Thread.currentThread().setName("multi-table-streamer-" + table);
+ HoodieStreamer streamer = null;
+ try {
+ streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
+ streamerInstances.add(streamer);
+ // Register before checking the flag so a concurrent
interruptAllIngestion() always sees this streamer.
+ if (shutdownRequested.get()) {
+ return;
+ }
+ streamer.sync();
+ // sync() also returns when fail fast stopped this table, so do not call
that a success. The stop can arrive
+ // as a flag rather than an interrupt if it lands while the table is
starting up; the ingestion loop reads the
+ // flag between rounds, so the table finishes any round already in
flight and then stops.
+ if (!shutdownRequested.get()) {
+ successTables.add(table);
+ }
+ } catch (Exception e) {
+ log.error("error while running MultiTableDeltaStreamer for table: {}",
table, e);
+ failedTables.add(table);
+ if (failFastOnContinuousMode) {
+ // Name the table so the thrown exception identifies the culprit, not
the siblings torn down after it.
+ throw new HoodieException("Table sync failed in continuous mode for
table: " + table, e);
+ }
+ } finally {
+ if (streamer != null) {
+ shutdownQuietly(streamer, table);
+ }
+ }
+ }
+
+ /**
+ * Waits until either every table sync finishes successfully or the first
one fails. On the first failure, the
+ * remaining streamers are shut down and a {@link HoodieException} is
thrown. {@link FutureUtils#allOf} only trips
+ * on an <em>exceptional</em> completion, so a table that terminates
normally (e.g. via a
+ * {@link PostWriteTerminationStrategy}) does not abort its siblings.
+ */
+ private static void awaitFailFast(List<CompletableFuture<Void>>
tableFutures) {
+ try {
+ FutureUtils.allOf(tableFutures).join();
+ } catch (CompletionException e) {
+ Throwable cause = unwrapCompletionException(e);
+ // An Error is rethrown as is rather than boxed, so the JVM-level
failure reaches the caller unchanged.
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ log.error("error while running MultiTableDeltaStreamer, shutting down
remaining tables as fail fast is enabled", cause);
+ throw new HoodieException("Fail fast is enabled and a table sync failed
in continuous mode.", cause);
+ }
+ }
+
+ /**
+ * Releases a streamer's resources, logging rather than propagating a
failure to do so. Closing can throw, and this
+ * runs in a {@code finally} on the failure path where escaping would mask
the table failure and, in
+ * {@link #syncSequentially()}, abort the tables not synced yet.
+ */
+ private static void shutdownQuietly(HoodieStreamer streamer, String table) {
+ try {
+ streamer.shutdownGracefully();
+ } catch (Exception e) {
+ log.warn("error while shutting down the streamer for table: {}", table,
e);
+ }
+ }
+
+ // A worker failure reaches the waiter wrapped in CompletionException, and
FutureUtils.allOf re-wraps it, so the
+ // real cause can sit under more than one layer.
+ private static Throwable unwrapCompletionException(CompletionException e) {
+ Throwable cause = e;
+ while (cause instanceof CompletionException && cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ return cause;
+ }
+
+ /**
+ * Waits for the per-table workers to unwind, bounded by {@link
Constants#SHUTDOWN_TIMEOUT_SECONDS} so a stuck table
+ * cannot hang the job. The workers are not interrupted: their ingestion
services have already been, and a worker
+ * only waits on its own service, so interrupting it would just make it
abandon that wait and close its StreamSync
+ * while the ingestion thread is still writing. A table that does not stop
is reported rather than thrown, since the
+ * caller is either already propagating a failure or has seen every worker
return.
+ */
+ private static void shutdownExecutor(ExecutorService executor) {
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(Constants.SHUTDOWN_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ log.error("table ingestion workers did not terminate; ingestion may
still be running");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.error("interrupted while waiting for the table ingestion workers to
terminate; ingestion may still be running");
+ }
+ }
+
+ private static void interruptAllIngestion(List<HoodieStreamer>
streamerInstances) {
+ for (HoodieStreamer streamer : streamerInstances) {
+ try {
+ streamer.interruptIngestion();
+ } catch (Exception e) {
+ log.warn("error while interrupting the ingestion of a streamer
instance for the table: {}", streamer.getConfig().targetTableName, e);
+ }
}
}
@@ -497,5 +712,7 @@ public class HoodieMultiTableStreamer {
private static final String DELIMITER = ".";
private static final String UNDERSCORE = "_";
private static final String COMMA_SEPARATOR = ",";
+ // How long the executor shutdown waits for the running table syncs to
terminate.
+ private static final long SHUTDOWN_TIMEOUT_SECONDS = 60;
}
}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
index a9ff817e0fa9..d062e11b6bc2 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
@@ -220,6 +220,20 @@ public class HoodieStreamer implements Serializable {
});
}
+ /**
+ * Interrupts an in-progress ingestion. Unlike {@link #shutdownGracefully()}
this neither closes nor waits: the sync
+ * is still running, and the thread owning it releases the resources once it
returns. The interrupt lands mid-round,
+ * so the table can be left with an inflight instant that is rolled back on
the next run.
+ */
+ public void interruptIngestion() {
+ ingestionService.ifPresent(ds -> {
+ if (!ds.isShutdown()) {
+ log.info("Interrupting ingestion for table {}", cfg.targetTableName);
+ ds.shutdown(true);
+ }
+ });
+ }
+
/**
* Main method to start syncing.
*/
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java
index b1e15b494c7e..4d4b5d9298f8 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java
@@ -28,22 +28,32 @@ import
org.apache.hudi.utilities.config.HoodieSchemaProviderConfig;
import org.apache.hudi.utilities.config.HoodieStreamerConfig;
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
import org.apache.hudi.utilities.schema.SchemaRegistryProvider;
+import org.apache.hudi.utilities.sources.ContinuousTestSource;
import org.apache.hudi.utilities.sources.JsonKafkaSource;
import org.apache.hudi.utilities.sources.ParquetDFSSource;
import org.apache.hudi.utilities.sources.TestDataSource;
+import org.apache.hudi.utilities.streamer.NoNewDataTerminationStrategy;
import org.apache.hudi.utilities.streamer.TableExecutionContext;
import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
import java.util.stream.Collectors;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -243,6 +253,120 @@ public class TestHoodieMultiTableDeltaStreamer extends
HoodieDeltaStreamerTestBa
}
}
+ @Test
+ public void testFailFastOnContinuousDefaultsToFalse() {
+ HoodieMultiTableDeltaStreamer.Config cfg = new
HoodieMultiTableDeltaStreamer.Config();
+ assertFalse(cfg.failFastOnContinuousMode);
+ }
+
+ @Timeout(600)
+ @Test
+ public void testMultiTableContinuousModeSyncsAllTablesInParallel() throws
IOException {
+ HoodieMultiTableDeltaStreamer streamer =
setupContinuousStreamer("parquetContinuous", false);
+ List<TableExecutionContext> contexts =
streamer.getTableExecutionContexts();
+ // Let each table stop on its own once it has ingested its data, so the
test does not run forever.
+ setTerminationStrategy(contexts);
+
+ streamer.sync();
+
+ assertEquals(2, streamer.getSuccessTables().size());
+ assertTrue(streamer.getFailedTables().isEmpty());
+ assertRecordCount(10, contexts.get(0).getConfig().targetBasePath,
sqlContext);
+ assertRecordCount(5, contexts.get(1).getConfig().targetBasePath,
sqlContext);
+ }
+
+ @Timeout(600)
+ @Test
+ public void testFailFastOnContinuousThrowsWhenATableFails() throws
IOException {
+ HoodieMultiTableDeltaStreamer streamer =
setupContinuousStreamer("parquetFailFast", true);
+ List<TableExecutionContext> contexts =
streamer.getTableExecutionContexts();
+ // Table 1 blocks after the barrier, so only fail fast interrupting it can
end its sync. Table 2 fails after the
+ // barrier. This proves fail fast tears down a sibling that is still
actively running, not one that stopped itself.
+
contexts.get(0).getProperties().setProperty(ContinuousTestSource.BLOCK_UNTIL_INTERRUPTED,
"true");
+
contexts.get(1).getProperties().setProperty(ContinuousTestSource.FAIL_AFTER_BARRIER,
"true");
+
+ HoodieException thrown = assertThrows(HoodieException.class,
streamer::sync);
+ assertFalse(streamer.getFailedTables().isEmpty());
+ // Both tables end up in failedTables, so the exception is what identifies
the one that actually failed.
+ // Assert to catch if unwrapCompletionException peeled nothing.
+ assertTrue(thrown.getCause() instanceof HoodieException, "expected the
cause to be unwrapped to a HoodieException");
+
assertTrue(thrown.getCause().getMessage().contains(tableWithDatabase(contexts.get(1))),
+ "expected the cause to name the failing table, got: " +
thrown.getCause().getMessage());
+ // sync() returns only after the blocked sibling was interrupted, so the
latch must already be counted down.
+ assertTrue(ContinuousTestSource.wasBlockedTableInterrupted());
+ }
+
+ @Timeout(600)
+ @Test
+ public void testFailFastOnContinuousWhenEveryTableFinishes() throws
IOException {
+ HoodieMultiTableDeltaStreamer streamer =
setupContinuousStreamer("parquetFailFastAllFinish", true);
+ List<TableExecutionContext> contexts =
streamer.getTableExecutionContexts();
+ // With fail fast on and no table failing, the wait has to complete
normally rather than trip on the first
+ // table to reach its termination strategy.
+ setTerminationStrategy(contexts);
+
+ streamer.sync();
+
+ assertEquals(2, streamer.getSuccessTables().size());
+ assertTrue(streamer.getFailedTables().isEmpty());
+ assertRecordCount(10, contexts.get(0).getConfig().targetBasePath,
sqlContext);
+ assertRecordCount(5, contexts.get(1).getConfig().targetBasePath,
sqlContext);
+ }
+
+ @Timeout(600)
+ @Test
+ public void testFailFastOnContinuousAfterASiblingFinished() throws Exception
{
+ HoodieMultiTableDeltaStreamer streamer =
setupContinuousStreamer("parquetFailFastAfterSibling", true);
+ List<TableExecutionContext> contexts =
streamer.getTableExecutionContexts();
+ // Table 1 terminates normally, then table 2 fails. This is what separates
allOf from anyOf: anyOf resolves on
+ // table 1's normal completion, so table 2's failure would never surface
and sync() would return cleanly.
+ contexts.get(0).getConfig().postWriteTerminationStrategyClass =
NoNewDataTerminationStrategy.class.getName();
+
contexts.get(1).getProperties().setProperty(ContinuousTestSource.FAIL_WHEN_RELEASED,
"true");
+
+ ExecutorService syncExecutor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> sync = syncExecutor.submit(streamer::sync);
+ // successTables only gains a table once its sync has returned, so this
gates table 2's failure on table 1
+ // having genuinely finished rather than on any intermediate step of its
ingestion.
+ awaitUntil(() -> !streamer.getSuccessTables().isEmpty(), "table 1 never
finished its sync");
+ ContinuousTestSource.releaseFailingTable();
+
+ ExecutionException thrown = assertThrows(ExecutionException.class,
sync::get);
+ assertTrue(thrown.getCause() instanceof HoodieException, "expected
sync() to fail with a HoodieException");
+ // It has to be fail fast that threw, not the end-of-sync check on
failedTables: that check fires whichever
+ // way the wait resolved, so only this message distinguishes allOf from
anyOf.
+ assertTrue(thrown.getCause().getMessage().contains("Fail fast is
enabled"),
+ "expected fail fast to surface table 2's failure, got: " +
thrown.getCause().getMessage());
+ } finally {
+ syncExecutor.shutdownNow();
+ }
+
+ // Only the table that actually failed is recorded; the one that finished
first was not torn down with it.
+ assertEquals(1, streamer.getSuccessTables().size());
+ assertEquals(1, streamer.getFailedTables().size());
+ assertRecordCount(10, contexts.get(0).getConfig().targetBasePath,
sqlContext);
+ }
+
+ @Timeout(600)
+ @Test
+ public void testContinuousModeDefaultDoesNotStopSiblingsWhenATableFails()
throws IOException {
+ HoodieMultiTableDeltaStreamer streamer =
setupContinuousStreamer("parquetNoFailFast", false);
+ List<TableExecutionContext> contexts =
streamer.getTableExecutionContexts();
+ setTerminationStrategy(contexts);
+ // Table 2 fails after the barrier while table 1 ingests normally. With
fail fast off, the failure of one table
+ // must not stop the other.
+
contexts.get(1).getProperties().setProperty(ContinuousTestSource.FAIL_AFTER_BARRIER,
"true");
+
+ // The job still fails, since continuous mode is not meant to end with a
failed table, but only after the
+ // healthy table has run to completion on its own.
+ assertThrows(HoodieException.class, streamer::sync);
+
+ assertEquals(1, streamer.getSuccessTables().size());
+ assertEquals(1, streamer.getFailedTables().size());
+ // The healthy table still ingested all of its records.
+ assertRecordCount(10, contexts.get(0).getConfig().targetBasePath,
sqlContext);
+ }
+
@Test
public void testTableLevelProperties() throws IOException {
HoodieMultiTableDeltaStreamer.Config cfg =
TestHelpers.getConfig(PROPS_FILENAME_TEST_SOURCE1, basePath + "/config",
TestDataSource.class.getName(), false, false, null);
@@ -300,4 +424,45 @@ public class TestHoodieMultiTableDeltaStreamer extends
HoodieDeltaStreamerTestBa
assertRecordCount(table1ExpectedRecords, targetBasePath1, sqlContext);
assertRecordCount(table2ExpectedRecords, targetBasePath2, sqlContext);
}
+
+ /**
+ * Builds a two-table continuous-mode streamer backed by {@link
ContinuousTestSource} (10 and 5 records). The source
+ * makes every table wait at a shared barrier before producing data, so a
sequential implementation would block the
+ * first table forever and time out; only truly concurrent syncs let all
tables pass the barrier.
+ */
+ private HoodieMultiTableDeltaStreamer setupContinuousStreamer(String
namePrefix, boolean failFast) throws IOException {
+ String sourceRoot1 = basePath + "/" + namePrefix + "Src1/";
+ String sourceRoot2 = basePath + "/" + namePrefix + "Src2/";
+ prepareParquetDFSFiles(10, sourceRoot1);
+ prepareParquetDFSFiles(5, sourceRoot2);
+
+ HoodieMultiTableDeltaStreamer.Config cfg =
TestHelpers.getConfig(populateCommonPropsAndWriteToFile(), basePath + "/config",
+ ContinuousTestSource.class.getName(), false, false, false,
"multi_table_" + namePrefix, null);
+ cfg.continuousMode = true;
+ cfg.failFastOnContinuousMode = failFast;
+
+ HoodieMultiTableDeltaStreamer streamer = new
HoodieMultiTableDeltaStreamer(cfg, jsc);
+ List<TableExecutionContext> contexts =
streamer.getTableExecutionContexts();
+ ingestPerParquetSourceProps(contexts, Arrays.asList(sourceRoot1,
sourceRoot2));
+ ContinuousTestSource.resetBarrier(contexts.size());
+ return streamer;
+ }
+
+ private static String tableWithDatabase(TableExecutionContext context) {
+ return context.getDatabase() + "." + context.getTableName();
+ }
+
+ private void setTerminationStrategy(List<TableExecutionContext>
executionContexts) {
+ for (TableExecutionContext context : executionContexts) {
+ context.getConfig().postWriteTerminationStrategyClass =
NoNewDataTerminationStrategy.class.getName();
+ }
+ }
+
+ private static void awaitUntil(BooleanSupplier condition, String message)
throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.MINUTES.toNanos(2);
+ while (!condition.getAsBoolean()) {
+ assertTrue(System.nanoTime() < deadline, message);
+ Thread.sleep(100);
+ }
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java
new file mode 100644
index 000000000000..8b687688749a
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java
@@ -0,0 +1,156 @@
+/*
+ * 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.utilities.sources;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.table.checkpoint.Checkpoint;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A parquet test source that proves continuous-mode multi-table syncs run in
parallel.
+ *
+ * <p>Every table waits at a shared barrier before producing data, so a
sequential implementation would block the first
+ * table forever and time out. Only concurrent syncs let all tables pass the
barrier.
+ */
+public class ContinuousTestSource extends ParquetDFSSource {
+
+ // When set on a table's properties, that table fails right after passing
the barrier, i.e. once all tables started.
+ public static final String FAIL_AFTER_BARRIER =
"hoodie.test.continuous.source.fail.after.barrier";
+
+ // When set on a table's properties, that table blocks after the barrier
until fail fast interrupts it.
+ public static final String BLOCK_UNTIL_INTERRUPTED =
"hoodie.test.continuous.source.block.until.interrupted";
+
+ // When set on a table's properties, that table blocks after the barrier and
fails once releaseFailingTable()
+ // is called, letting a test choose exactly when the failure happens
relative to the other tables.
+ public static final String FAIL_WHEN_RELEASED =
"hoodie.test.continuous.source.fail.when.released";
+
+ private static final long BARRIER_TIMEOUT_SECONDS = 60;
+
+ // Longer than the test's awaitUntil deadline, so a stall is reported by the
test rather than by this source
+ private static final long RELEASE_TIMEOUT_SECONDS = 300;
+
+ private static volatile CyclicBarrier startBarrier = new CyclicBarrier(1);
+ // Counted down by a blocking table once it observes the fail-fast
interrupt, so a test can assert it was torn down.
+ private static volatile CountDownLatch blockedTableInterrupted = new
CountDownLatch(1);
+ // Released by the test to make a blocked table fail at a moment of the
test's choosing.
+ private static volatile CountDownLatch failRelease = new CountDownLatch(1);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ private final boolean failWhenReleased;
+ private final AtomicBoolean barrierPassed = new AtomicBoolean(false);
+
+ public ContinuousTestSource(TypedProperties props, JavaSparkContext
sparkContext, SparkSession sparkSession,
+ SchemaProvider schemaProvider) {
+ super(props, sparkContext, sparkSession, schemaProvider);
+ this.failAfterBarrier = props.getBoolean(FAIL_AFTER_BARRIER, false);
+ this.blockUntilInterrupted = props.getBoolean(BLOCK_UNTIL_INTERRUPTED,
false);
+ this.failWhenReleased = props.getBoolean(FAIL_WHEN_RELEASED, false);
+ }
+
+ // Resets the shared barrier and latch used to coordinate tables. Call
before each sync.
+ public static void resetBarrier(int numTables) {
+ startBarrier = new CyclicBarrier(numTables);
+ blockedTableInterrupted = new CountDownLatch(1);
+ failRelease = new CountDownLatch(1);
+ }
+
+ // Makes the table configured with FAIL_WHEN_RELEASED fail now.
+ public static void releaseFailingTable() {
+ failRelease.countDown();
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+ return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair<Option<Dataset<Row>>, Checkpoint>
fetchNextBatch(Option<Checkpoint> lastCheckpoint, long sourceLimit) {
+ // Only rendezvous once, on the first fetch, so that later empty polls do
not block termination.
+ if (barrierPassed.compareAndSet(false, true)) {
+ awaitBarrier();
+ if (failAfterBarrier) {
+ throw new HoodieException("Simulated table sync failure after all
tables started");
+ }
+ if (blockUntilInterrupted) {
+ blockUntilFailFastInterrupts();
+ }
+ if (failWhenReleased) {
+ awaitRelease();
+ throw new HoodieException("Simulated table sync failure, released by
the test");
+ }
+ }
+ return super.fetchNextBatch(lastCheckpoint, sourceLimit);
+ }
+
+ // Blocks until the test decides this table should fail.
+ private void awaitRelease() {
+ try {
+ if (!failRelease.await(RELEASE_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ throw new HoodieException("Timed out waiting for the test to release
the failing table");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new HoodieException("Interrupted while waiting for the test to
release the failing table", e);
+ }
+ }
+
+ // Blocks until fail fast interrupts this table, records that it observed
the interrupt, then propagates it.
+ private void blockUntilFailFastInterrupts() {
+ // Nothing counts this down. Only an interrupt can end.
+ CountDownLatch blockIndefinitely = new CountDownLatch(1);
+ try {
+ blockIndefinitely.await();
+ } catch (InterruptedException e) {
+ blockedTableInterrupted.countDown();
+ Thread.currentThread().interrupt();
+ throw new HoodieException("Table sync was interrupted by fail fast", e);
+ }
+ }
+
+ private void awaitBarrier() {
+ try {
+ startBarrier.await(BARRIER_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ throw new HoodieException("Timed out waiting for all tables to start
syncing; continuous-mode syncs are not "
+ + "running in parallel", e);
+ } catch (BrokenBarrierException e) {
+ throw new HoodieException("Barrier was broken while waiting for all
tables to start syncing", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new HoodieException("Interrupted while waiting for all tables to
start syncing", e);
+ }
+ }
+}
