hudi-agent commented on code in PR #19696:
URL: https://github.com/apache/hudi/pull/19696#discussion_r3825971709
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -479,11 +512,81 @@ 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<>();
+ final ExecutorService executor =
Executors.newFixedThreadPool(tableExecutionContexts.size(),
+ new CustomizedThreadFactory("multi-table-streamer", true));
+ 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);
+ streamer.sync();
+ 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);
+
+ // In fail-fast mode getConditionalFuture returns anyOf(...), which
completes exceptionally as soon as the first
+ // table fails; otherwise it returns allOf(...) and only completes once
every table sync has terminated.
+ Throwable failure = getConditionalFuture(tableFutures)
+ .handle((unused, throwable) -> throwable)
+ .join();
+ log.info("Successful tables: {}, Failed tables: {}", successTables,
failedTables);
+ if (failure != null && failFastOnContinuousMode) {
+ log.error("MultiTableDeltaStreamer failed with an exception, shutting
down remaining sources as fail fast is enabled!", failure);
+ shutdownStreamers(streamerInstances);
+ throw new HoodieException("Fail fast is enabled and a table sync
failed in continuous mode.", failure);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private void shutdownStreamers(List<HoodieStreamer> streamerInstances) {
+ for (HoodieStreamer streamer : streamerInstances) {
+ try {
+ if (!streamer.getIngestionService().isShutdown()) {
+ streamer.getIngestionService().shutdown(true);
+ }
+ } catch (Exception e) {
+ log.warn("Failed to shut down a streamer instance during fail-fast
handling.", e);
+ }
+ }
+ }
+
+ /**
+ * Returns a future that completes when the first table sync fails
(fail-fast) or when all table syncs
+ * complete (default).
+ */
+ private CompletableFuture<?> getConditionalFuture(final
CompletableFuture<?>[] tableFutures) {
+ if (failFastOnContinuousMode) {
+ log.info("Fail fast enabled in continuous mode. The whole job fails on
any single table failure.");
+ return CompletableFuture.anyOf(tableFutures);
Review Comment:
🤖 `CompletableFuture.anyOf` completes on the first future to *finish* —
success or failure — not the first failure. In continuous mode a table with a
`PostWriteTerminationStrategy` (like the `NoNewDataTerminationStrategy` your
new tests use) terminates normally, so if a healthy table finishes first,
`anyOf` completes normally, `failure` stays null, the fail-fast throw is
skipped, and the `finally`'s `executor.shutdownNow()` still interrupts the
other tables mid-sync. That also makes
`testFailFastOnContinuousThrowsWhenATableFails` depend on table 2 failing
before table 1 completes. Could you trigger fail-fast only on the first
*exceptional* completion instead — e.g. attach a `whenComplete`/`exceptionally`
that trips a signal on throwable rather than relying on `anyOf`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -479,11 +512,81 @@ 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<>();
+ final ExecutorService executor =
Executors.newFixedThreadPool(tableExecutionContexts.size(),
+ new CustomizedThreadFactory("multi-table-streamer", true));
+ 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);
+ streamer.sync();
+ 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);
+
+ // In fail-fast mode getConditionalFuture returns anyOf(...), which
completes exceptionally as soon as the first
+ // table fails; otherwise it returns allOf(...) and only completes once
every table sync has terminated.
+ Throwable failure = getConditionalFuture(tableFutures)
+ .handle((unused, throwable) -> throwable)
+ .join();
+ log.info("Successful tables: {}, Failed tables: {}", successTables,
failedTables);
+ if (failure != null && failFastOnContinuousMode) {
+ log.error("MultiTableDeltaStreamer failed with an exception, shutting
down remaining sources as fail fast is enabled!", failure);
+ shutdownStreamers(streamerInstances);
Review Comment:
🤖 When fail-fast trips, `shutdownStreamers` only shuts down streamers
already added to `streamerInstances`. A sibling thread still inside `new
HoodieStreamer(...)` (or in the window before `streamerInstances.add(...)` at
the top of the task) won't be in the list, so its ingestion can start right
after this shutdown runs. And `executor.shutdownNow()` in the `finally` only
interrupts the worker thread — `StreamSyncService.startService()` uses its own
non-daemon single-thread executor for the ingest loop, which isn't stopped by
interrupting the worker. Could this leave a streamer running after fail-fast?
It might be safer to set a shutdown flag that every task re-checks after
registering, or to shut down by draining `streamerInstances` until empty.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java:
##########
@@ -243,6 +245,75 @@ public void testMultiTableExecutionWithParquetSource()
throws IOException {
}
}
+ @Test
+ public void testFailFastOnContinuousDefaultsToFalse() {
+ HoodieMultiTableDeltaStreamer.Config cfg = new
HoodieMultiTableDeltaStreamer.Config();
+ assertFalse(cfg.failFastOnContinuousMode);
+ }
+
+ @Test
+ public void testMultiTableContinuousModeSyncsAllTablesInParallel() throws
IOException {
+ // ingest test data to 2 parquet source paths
+ String parquetSourceRoot1 = basePath + "/parquetContSrc1/";
+ prepareParquetDFSFiles(10, parquetSourceRoot1);
+ String parquetSourceRoot2 = basePath + "/parquetContSrc2/";
+ prepareParquetDFSFiles(5, parquetSourceRoot2);
+
+ String parquetPropsFile = populateCommonPropsAndWriteToFile();
+
+ HoodieMultiTableDeltaStreamer.Config cfg =
TestHelpers.getConfig(parquetPropsFile, basePath + "/config",
ParquetDFSSource.class.getName(), false, false,
+ false, "multi_table_parquet_continuous", null);
+ // Continuous mode blocks per table, so tables must be synced concurrently.
+ cfg.continuousMode = true;
+
+ HoodieMultiTableDeltaStreamer streamer = new
HoodieMultiTableDeltaStreamer(cfg, jsc);
Review Comment:
🤖 nit: `Arrays.asList(new String[] {...})` — the intermediate `new String[]`
is redundant since `Arrays.asList` is already varargs. Could you simplify to
`Arrays.asList(parquetSourceRoot1, parquetSourceRoot2)`? Same applies on line
299.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
Review Comment:
🤖 **Line 613:** nit: `getConditionalFuture` is a bit opaque — "conditional"
doesn't say what the condition is or what kind of future is returned. Could you
rename it to something like `buildSyncCompletionFuture`? The Javadoc explains
the intent well, but a clearer name would make the call-site in
`syncContinuously` self-explanatory without jumping to the definition.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
