wombatu-kun commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3922950296
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -479,10 +516,143 @@ public void sync() {
}
}
}
+ }
- log.info("Ingestion was successful for topics: {}", successTables);
- if (!failedTables.isEmpty()) {
- log.info("Ingestion failed for topics: {}", failedTables);
+ /**
+ * 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() {
+ // 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 CompletableFuture<?>[] tableFutures =
tableExecutionContexts.stream()
+ .map(context -> CompletableFuture.runAsync(() -> {
+ 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(Helpers.getTableWithDatabase(context));
+ }
+ } catch (Exception e) {
+ log.error("error while running MultiTableDeltaStreamer for
table: {}", context.getTableName(), e);
+ failedTables.add(Helpers.getTableWithDatabase(context));
+ if (failFastOnContinuousMode) {
+ throw new CompletionException(e);
+ }
+ } finally {
+ if (streamer != null) {
+ streamer.shutdownGracefully();
+ }
+ }
+ }, executor)).toArray(CompletableFuture[]::new);
+
+ if (failFastOnContinuousMode) {
+ log.info("Fail fast enabled in continuous mode. The whole job fails on
any single table failure");
+ awaitFailFast(tableFutures, streamerInstances, shutdownRequested);
+ } else {
+ CompletableFuture.allOf(tableFutures).join();
+ }
+ log.info("Successful tables: {}, Failed tables: {}", successTables,
failedTables);
+ } finally {
+ // 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");
+ }
+ }
+
+ /**
+ * 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. Unlike {@code anyOf(...)}, this 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 void awaitFailFast(CompletableFuture<?>[] tableFutures,
List<HoodieStreamer> streamerInstances, AtomicBoolean shutdownRequested) {
+ final CompletableFuture<Void> firstOutcome = new CompletableFuture<>();
Review Comment:
hudi-common's FutureUtils.allOf already provides this exact contract,
completing normally once every future succeeds and exceptionally on the first
exceptional completion, and CloudObjectsSelectorCommon uses it in this module
over the same fixed pool and CustomizedThreadFactory shape. Could awaitFailFast
delegate the future plumbing to it and keep only the
shutdownRequested/shutdownStreamers teardown, unwrapping the extra
CompletionException layer the way unwrapExistsCheckFailure does?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java:
##########
@@ -243,6 +246,89 @@ public void testMultiTableExecutionWithParquetSource()
throws IOException {
}
}
+ @Test
+ public void testFailFastOnContinuousDefaultsToFalse() {
+ HoodieMultiTableDeltaStreamer.Config cfg = new
HoodieMultiTableDeltaStreamer.Config();
+ assertFalse(cfg.failFastOnContinuousMode);
+ }
+
+ @Test
+ public void testMultiTableContinuousModeSyncsAllTablesInParallel() throws
IOException {
Review Comment:
None of the three new continuous tests carry @Timeout, while every
continuous-mode test in hudi-utilities does, and a regression in the fail-fast
teardown parks the worker in shutdownGracefully's 24-hour awaitTermination
instead of failing. Could you add @Timeout(600) to match the continuous tests
in TestHoodieDeltaStreamer?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -479,10 +516,143 @@ public void sync() {
}
}
}
+ }
- log.info("Ingestion was successful for topics: {}", successTables);
- if (!failedTables.isEmpty()) {
- log.info("Ingestion failed for topics: {}", failedTables);
+ /**
+ * 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() {
+ // 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 CompletableFuture<?>[] tableFutures =
tableExecutionContexts.stream()
+ .map(context -> CompletableFuture.runAsync(() -> {
+ 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(Helpers.getTableWithDatabase(context));
+ }
+ } catch (Exception e) {
+ log.error("error while running MultiTableDeltaStreamer for
table: {}", context.getTableName(), e);
+ failedTables.add(Helpers.getTableWithDatabase(context));
+ if (failFastOnContinuousMode) {
+ throw new CompletionException(e);
+ }
+ } finally {
+ if (streamer != null) {
+ streamer.shutdownGracefully();
+ }
+ }
+ }, executor)).toArray(CompletableFuture[]::new);
+
+ if (failFastOnContinuousMode) {
+ log.info("Fail fast enabled in continuous mode. The whole job fails on
any single table failure");
+ awaitFailFast(tableFutures, streamerInstances, shutdownRequested);
+ } else {
+ CompletableFuture.allOf(tableFutures).join();
+ }
+ log.info("Successful tables: {}, Failed tables: {}", successTables,
failedTables);
+ } finally {
+ // 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");
+ }
+ }
+
+ /**
+ * 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. Unlike {@code anyOf(...)}, this 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 void awaitFailFast(CompletableFuture<?>[] tableFutures,
List<HoodieStreamer> streamerInstances, AtomicBoolean shutdownRequested) {
+ final CompletableFuture<Void> firstOutcome = new CompletableFuture<>();
+ // Trip as soon as any table fails ...
+ for (CompletableFuture<?> tableFuture : tableFutures) {
+ tableFuture.whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ firstOutcome.completeExceptionally(throwable);
+ }
+ });
+ }
+ // ... or complete normally once every table has finished without failure.
+ CompletableFuture.allOf(tableFutures).whenComplete((result, throwable) -> {
Review Comment:
No test pairs --fail-fast-on-continuous with a table that terminates
normally, so replacing this whole block with
CompletableFuture.anyOf(tableFutures).join() still passes all three new tests.
Could you add a fail-fast case where one table stops via
NoNewDataTerminationStrategy while a sibling keeps running?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -479,10 +516,143 @@ public void sync() {
}
}
}
+ }
- log.info("Ingestion was successful for topics: {}", successTables);
- if (!failedTables.isEmpty()) {
- log.info("Ingestion failed for topics: {}", failedTables);
+ /**
+ * 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() {
+ // 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 CompletableFuture<?>[] tableFutures =
tableExecutionContexts.stream()
+ .map(context -> CompletableFuture.runAsync(() -> {
+ 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(Helpers.getTableWithDatabase(context));
+ }
+ } catch (Exception e) {
+ log.error("error while running MultiTableDeltaStreamer for
table: {}", context.getTableName(), e);
+ failedTables.add(Helpers.getTableWithDatabase(context));
+ if (failFastOnContinuousMode) {
+ throw new CompletionException(e);
+ }
+ } finally {
+ if (streamer != null) {
+ streamer.shutdownGracefully();
+ }
+ }
+ }, executor)).toArray(CompletableFuture[]::new);
+
+ if (failFastOnContinuousMode) {
+ log.info("Fail fast enabled in continuous mode. The whole job fails on
any single table failure");
+ awaitFailFast(tableFutures, streamerInstances, shutdownRequested);
+ } else {
+ CompletableFuture.allOf(tableFutures).join();
+ }
+ log.info("Successful tables: {}, Failed tables: {}", successTables,
failedTables);
+ } finally {
+ // 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");
+ }
+ }
+
+ /**
+ * 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. Unlike {@code anyOf(...)}, this 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 void awaitFailFast(CompletableFuture<?>[] tableFutures,
List<HoodieStreamer> streamerInstances, AtomicBoolean shutdownRequested) {
+ final CompletableFuture<Void> firstOutcome = new CompletableFuture<>();
+ // Trip as soon as any table fails ...
+ for (CompletableFuture<?> tableFuture : tableFutures) {
+ tableFuture.whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ firstOutcome.completeExceptionally(throwable);
+ }
+ });
+ }
+ // ... or complete normally once every table has finished without failure.
+ CompletableFuture.allOf(tableFutures).whenComplete((result, throwable) -> {
+ if (throwable == null) {
+ firstOutcome.complete(null);
+ }
+ });
+
+ try {
+ firstOutcome.join();
+ } catch (CompletionException e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ log.error("error while running MultiTableDeltaStreamer, shutting down
remaining tables as fail fast is enabled", cause);
+ shutdownRequested.set(true);
+ // shutdownStreamers only interrupts; the executor teardown in
syncContinuously() waits for the siblings to stop.
+ shutdownStreamers(streamerInstances);
+ throw new HoodieException("Fail fast is enabled and a table sync failed
in continuous mode.", 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 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;
+ }
+ }
+
+ private void shutdownStreamers(List<HoodieStreamer> streamerInstances) {
+ for (HoodieStreamer streamer : streamerInstances) {
+ try {
+ if (!streamer.getIngestionService().isShutdown()) {
+ streamer.getIngestionService().shutdown(true);
Review Comment:
This is the first src/main caller of HoodieStreamer.getIngestionService(),
which is annotated @VisibleForTesting and every other caller of which is a
test. Would a forced-shutdown method on HoodieStreamer, mirroring
shutdownGracefully(), be a better fit than reaching through that accessor?
--
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]
