wombatu-kun commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3964151632
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -461,28 +484,204 @@ private static String resetTarget(Config configuration,
String database, String
/**
* 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);
}
}
}
+ }
+
+ /**
+ * 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 fails the whole job. The sibling
+ * streamers are shut down and a {@link HoodieException} is thrown so the
caller can exit with a non-zero status.
+ * Otherwise, every table is synced independently and a single failure does
not affect the others.
+ */
+ private void syncContinuously() {
+ if (tableExecutionContexts.isEmpty()) {
+ return;
+ }
+ // 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));
+ boolean terminated = false;
+ 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 {
+ // On an abnormal exit the siblings are still ingesting, since
FutureUtils.allOf only cancels their futures.
+ // Stopping them here rather than in a catch covers every such exit,
including an Error, which the workers do
+ // not catch; it is a no-op on the success path because each table has
already shut its ingestion service down.
+ shutdownRequested.set(true);
+ shutdownStreamers(streamerInstances);
+ // Wait for every worker thread to finish (including its finally
cleanup) before returning, so sync() does not
+ // return while a table is still writing and main() then stops the
shared Spark context under it.
+ terminated = shutdownExecutor(executor);
+ }
+ // If the workers never terminated, ingestion may still be running. Fail
loudly instead of returning as if the
+ // cleanup succeeded, so the caller does not silently proceed to Spark
teardown with live writers.
+ if (!terminated) {
+ throw new HoodieException("Timed out shutting down table ingestion
workers in continuous mode");
+ }
+ }
+
+ /**
+ * 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 shutdownStreamers()
always sees this streamer.
+ if (shutdownRequested.get()) {
+ return;
+ }
+ streamer.sync();
+ // A streamer registered just before fail fast tripped can reach here
without ever ingesting.
+ // shutdown() call will be a no-op because its ingestion service hadn't
started yet.
Review Comment:
shutdown(true) is not a no-op on an unstarted service: HoodieAsyncService
sets shutdownRequested before the executor null check, and that flag is exactly
why startService()'s loop runs zero rounds and sync() returns without
ingesting. Reword so the comment says the interrupt already marked the service
shut down.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -461,28 +484,204 @@ private static String resetTarget(Config configuration,
String database, String
/**
* 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);
}
}
}
+ }
+
+ /**
+ * 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 fails the whole job. The sibling
+ * streamers are shut down and a {@link HoodieException} is thrown so the
caller can exit with a non-zero status.
+ * Otherwise, every table is synced independently and a single failure does
not affect the others.
+ */
+ private void syncContinuously() {
+ if (tableExecutionContexts.isEmpty()) {
+ return;
+ }
+ // 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));
+ boolean terminated = false;
+ 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 {
+ // On an abnormal exit the siblings are still ingesting, since
FutureUtils.allOf only cancels their futures.
+ // Stopping them here rather than in a catch covers every such exit,
including an Error, which the workers do
+ // not catch; it is a no-op on the success path because each table has
already shut its ingestion service down.
+ shutdownRequested.set(true);
+ shutdownStreamers(streamerInstances);
+ // Wait for every worker thread to finish (including its finally
cleanup) before returning, so sync() does not
+ // return while a table is still writing and main() then stops the
shared Spark context under it.
+ terminated = shutdownExecutor(executor);
+ }
+ // If the workers never terminated, ingestion may still be running. Fail
loudly instead of returning as if the
+ // cleanup succeeded, so the caller does not silently proceed to Spark
teardown with live writers.
+ if (!terminated) {
+ throw new HoodieException("Timed out shutting down table ingestion
workers in continuous mode");
+ }
+ }
+
+ /**
+ * 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 shutdownStreamers()
always sees this streamer.
+ if (shutdownRequested.get()) {
+ return;
+ }
+ streamer.sync();
+ // A streamer registered just before fail fast tripped can reach here
without ever ingesting.
+ // shutdown() call will be a no-op because its ingestion service hadn't
started yet.
+ // Don't count that as a success.
+ 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;
+ }
+
+ /**
+ * Two-phase shutdown of the per-table executor: wait for the running syncs
to finish, then force-cancel any that
+ * ignore interruption. Bounded by {@link
Constants#SHUTDOWN_TIMEOUT_SECONDS} so a stuck table cannot hang the job.
+ *
+ * @return true if all workers terminated, false if any were still running
when the timeout elapsed.
+ */
+ private static boolean shutdownExecutor(ExecutorService executor) {
+ executor.shutdown();
+ try {
+ if (executor.awaitTermination(Constants.SHUTDOWN_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ return true;
+ }
+ executor.shutdownNow();
Review Comment:
shutdownNow() makes a worker terminate by interrupting it out of
waitForShutdown(), so awaitTermination can report success - and the worker's
finally can close its StreamSync - while that table's ingestion thread is still
running. Is the second phase worth keeping, given this wait exists so that
sync() does not return while a table is still writing?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -461,28 +484,204 @@ private static String resetTarget(Config configuration,
String database, String
/**
* 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);
}
}
}
+ }
+
+ /**
+ * 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 fails the whole job. The sibling
+ * streamers are shut down and a {@link HoodieException} is thrown so the
caller can exit with a non-zero status.
+ * Otherwise, every table is synced independently and a single failure does
not affect the others.
+ */
+ private void syncContinuously() {
+ if (tableExecutionContexts.isEmpty()) {
+ return;
+ }
+ // 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));
+ boolean terminated = false;
+ 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 {
+ // On an abnormal exit the siblings are still ingesting, since
FutureUtils.allOf only cancels their futures.
+ // Stopping them here rather than in a catch covers every such exit,
including an Error, which the workers do
+ // not catch; it is a no-op on the success path because each table has
already shut its ingestion service down.
+ shutdownRequested.set(true);
+ shutdownStreamers(streamerInstances);
+ // Wait for every worker thread to finish (including its finally
cleanup) before returning, so sync() does not
+ // return while a table is still writing and main() then stops the
shared Spark context under it.
+ terminated = shutdownExecutor(executor);
+ }
+ // If the workers never terminated, ingestion may still be running. Fail
loudly instead of returning as if the
+ // cleanup succeeded, so the caller does not silently proceed to Spark
teardown with live writers.
+ if (!terminated) {
+ throw new HoodieException("Timed out shutting down table ingestion
workers in continuous mode");
+ }
+ }
+
+ /**
+ * 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 shutdownStreamers()
always sees this streamer.
+ if (shutdownRequested.get()) {
+ return;
+ }
+ streamer.sync();
+ // A streamer registered just before fail fast tripped can reach here
without ever ingesting.
+ // shutdown() call will be a no-op because its ingestion service hadn't
started yet.
+ // Don't count that as a success.
+ 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;
+ }
+
+ /**
+ * Two-phase shutdown of the per-table executor: wait for the running syncs
to finish, then force-cancel any that
+ * ignore interruption. Bounded by {@link
Constants#SHUTDOWN_TIMEOUT_SECONDS} so a stuck table cannot hang the job.
+ *
+ * @return true if all workers terminated, false if any were still running
when the timeout elapsed.
+ */
+ private static boolean shutdownExecutor(ExecutorService executor) {
+ executor.shutdown();
+ try {
+ if (executor.awaitTermination(Constants.SHUTDOWN_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ return true;
+ }
+ executor.shutdownNow();
+ if (executor.awaitTermination(Constants.SHUTDOWN_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ return true;
+ }
+ log.error("executor service did not terminate after shutdown");
+ return false;
+ } catch (InterruptedException e) {
+ executor.shutdownNow();
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
- log.info("Ingestion was successful for topics: {}", successTables);
- if (!failedTables.isEmpty()) {
- log.info("Ingestion failed for topics: {}", failedTables);
+ private static void shutdownStreamers(List<HoodieStreamer>
streamerInstances) {
Review Comment:
shutdownStreamers only calls interruptIngestion, which neither closes nor
waits, and its warning says "during fail fast handling" although the finally
calls it on every exit, fail fast off and success included.
interruptAllIngestion plus a neutral message would match what it does.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * 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;
+
+ 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(BARRIER_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
Review Comment:
awaitRelease reuses BARRIER_TIMEOUT_SECONDS, so the failing table gives up
after 60s even though it is waiting for a whole table sync rather than a
rendezvous, while awaitUntil allows two minutes. Give the release its own
longer constant so awaitUntil's deadline is the binding one.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java:
##########
@@ -243,6 +253,113 @@ public void testMultiTableExecutionWithParquetSource()
throws IOException {
}
}
+ @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.
+
assertTrue(thrown.getCause().getMessage().contains(tableWithDatabase(contexts.get(1))),
Review Comment:
This passes even if unwrapCompletionException peels nothing, because
CompletionException.getMessage() is its cause's toString(). Add
assertTrue(thrown.getCause() instanceof HoodieException) alongside it, the way
testFailFastOnContinuousAfterASiblingFinished does, so the unwrapping is pinned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
