Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5764174261 Thank you, @wombatu-kun ! @yihua happy to run through any additional scenarios if that helps. Otherwise this is ready from my side. Thank you! -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5648860324 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * 1c24f373659ef84bb8ca3f85e684863f26ed54d6 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17057) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5647766331 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * 6a17a936894e41965d0ed2d34482cf281ccc9a2e Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16986) * 1c24f373659ef84bb8ca3f85e684863f26ed54d6 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5647799732 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * 6a17a936894e41965d0ed2d34482cf281ccc9a2e Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16986) * 1c24f373659ef84bb8ca3f85e684863f26ed54d6 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17057) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3997081700
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +482,222 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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 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> 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3981856891
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +482,222 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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 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> 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3981770906
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +482,222 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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 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> 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5622555762 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * 6a17a936894e41965d0ed2d34482cf281ccc9a2e Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16986) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5620991424 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * 85f26459353f4e10f7d217b8bcfb076d9acb250a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16964) * 6a17a936894e41965d0ed2d34482cf281ccc9a2e Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16986) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5620370420 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * 85f26459353f4e10f7d217b8bcfb076d9acb250a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16964) * 6a17a936894e41965d0ed2d34482cf281ccc9a2e UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3977457895
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -389,6 +406,16 @@ public static class Config implements Serializable {
+ " source-fetch -> Transform -> Hudi Write in loop")
public Boolean continuousMode = false;
+@Parameter(names = {"--fail-fast-on-continuous"},
+description = "Only applies in continuous mode. When enabled, the
first table failure immediately tears "
++ "down the remaining table syncs; 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. When disabled "
++ "(default), each table is synced independently and a single
failure does not stop the others. Either "
++ "way, if the run ends at all with a failed table the job exits
with a non-zero status, since continuous "
++ "mode is not meant to end. A table failing while the others keep
running does not end the run, so it "
++ "surfaces through that table's error log and metrics rather than
the exit code.")
+public Boolean failFastOnContinuousMode = false;
Review Comment:
🤖 nit: this `--help` description is quite a wall of text; could you trim it
to the essentials (fail-fast tears down siblings on first failure; default lets
others continue) and keep the inflight-instant / exit-code nuances in the
`syncContinuously()` javadoc, which already covers them?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5612089710 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * 85f26459353f4e10f7d217b8bcfb076d9acb250a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16964) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5611335265 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * c48ee15d2ca0bae7bc3aac85fd3b44d314cc75a5 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16962) * 85f26459353f4e10f7d217b8bcfb076d9acb250a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16964) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5611241261 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * c48ee15d2ca0bae7bc3aac85fd3b44d314cc75a5 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16962) * 85f26459353f4e10f7d217b8bcfb076d9acb250a UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5611151072 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * dc0815ef1d5ed202238acff3312124c6bea2e735 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16961) * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * c48ee15d2ca0bae7bc3aac85fd3b44d314cc75a5 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16962) * 85f26459353f4e10f7d217b8bcfb076d9acb250a UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3974547136
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -389,6 +406,16 @@ public static class Config implements Serializable {
+ " source-fetch -> Transform -> Hudi Write in loop")
public Boolean continuousMode = false;
+@Parameter(names = {"--fail-fast-on-continuous"},
+description = "Only applies in continuous mode. When enabled, the
first table failure immediately tears "
Review Comment:
🤖 nit: this `--help` description is eight lines of caveats; could you trim
it to the one-line behavior ("In continuous mode, stop all table syncs on the
first table failure. Default: false") and move the inflight-instant / exit-code
details into the Javadoc on `syncContinuously()`, which already covers most of
it?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java:
##
@@ -220,6 +220,20 @@ public void shutdownGracefully() {
});
}
+ /**
+ * 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("Forcefully shutting down DeltaStreamer");
Review Comment:
🤖 nit: the log line says "Forcefully shutting down" but the method is
`interruptIngestion()` and the Javadoc stresses it neither closes nor waits —
could you align the message (e.g. "Interrupting ingestion for table {}" with
the target table name) so a reader of the driver log doesn't assume the
streamer was fully shut down?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5610814269 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * dc0815ef1d5ed202238acff3312124c6bea2e735 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16961) * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * c48ee15d2ca0bae7bc3aac85fd3b44d314cc75a5 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16962) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5610640117 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * 7c6c8015bc4b3d65762abd133700a613fcc1bf37 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16960) * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * dc0815ef1d5ed202238acff3312124c6bea2e735 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16961) * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN * c48ee15d2ca0bae7bc3aac85fd3b44d314cc75a5 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5610583166 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) * 7c6c8015bc4b3d65762abd133700a613fcc1bf37 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16960) * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * dc0815ef1d5ed202238acff3312124c6bea2e735 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16961) * 2892085382bf6bcfe2758269373ac19e60506a1d UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5610237472 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) * 7c6c8015bc4b3d65762abd133700a613fcc1bf37 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16960) * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * dc0815ef1d5ed202238acff3312124c6bea2e735 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16961) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5610181533 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) * 7c6c8015bc4b3d65762abd133700a613fcc1bf37 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16960) * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN * dc0815ef1d5ed202238acff3312124c6bea2e735 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3973882509
##
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.
+ *
+ * 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.
+ *
+ * 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 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> 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3973682290
##
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.
+ *
+ * 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.
+ *
+ * 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 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> 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();
Review Comment:
Great catch! Simple yet I had to tread carefully.
1. With fail fast off every table could die and the job would still exit 0.
Added a throw at the end of syncContinuously when failedTables isn't empty
2. Regardless of the flag, continuous mode returning at all is only expected
when a table stops itself via a termination strategy. So added an assertion to
the test. Updated the flag description and javadoc as well.
3. On the partial case like 1 out of 10 failed and the job keeps running,
the check I added is never reachable. The only way to identify this is via
single error log for the failed table and a per table failure metrics.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5609768434 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) * 7c6c8015bc4b3d65762abd133700a613fcc1bf37 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16960) * f3ebaece35594e759a2f318a8c258be5cf1d7778 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5609656036 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) * 7c6c8015bc4b3d65762abd133700a613fcc1bf37 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16960) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3973519798
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java:
##
@@ -220,6 +220,19 @@ public void shutdownGracefully() {
});
}
+ /**
+ * 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.
+ */
+ public void interruptIngestion() {
+ingestionService.ifPresent(ds -> {
+ if (!ds.isShutdown()) {
+log.info("Forcefully shutting down DeltaStreamer");
+ds.shutdown(true);
Review Comment:
It is deliberate, yes! Documented it for now. You are right, the loop
already exits between rounds, but the graceful path is not reachable today.
HoodieAsyncService only offers interrupt or block for 24 hours. Happy to create
a follow up after this PR if you think it's worth adding.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5609423814 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) * 7c6c8015bc4b3d65762abd133700a613fcc1bf37 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5607307545 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5606033676 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 708a73a6a6d828de9a7ffd5e1c06a9bc5364dd86 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16929) * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16957) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5605789812 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 708a73a6a6d828de9a7ffd5e1c06a9bc5364dd86 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16929) * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN * aa1906a29cef83aca36a79b863e02eebd04da73d UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5605697634 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 708a73a6a6d828de9a7ffd5e1c06a9bc5364dd86 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16929) * 787d6594e78af5290b710b1cb406973e86c175bc UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3970991202
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +484,196 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.
+ *
+ * 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 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> 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.
Review Comment:
🤖 nit: the inline comments in this finally block (and the one above
`successTables.add` in `runTableSync`) are quite dense and read like a design
note; it might be worth trimming them to the key point ("finally also covers
Errors the workers don't catch; interrupt siblings then wait for cleanup") and
moving the race-condition reasoning into the method Javadoc.
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -389,6 +406,12 @@ public static class Config implements Serializable {
+ " source-fetch -> Transform -> Hudi Write in loop")
public Boolean continuousMode = false;
+@Parameter(names = {"--fail-fast-on-continuous"},
Review Comment:
🤖 nit: the CLI flag `--fail-fast-on-continuous` and the field
`failFastOnContinuousMode` differ slightly; could you align them (e.g.
`--fail-fast-on-continuous-mode`) so the flag mirrors `--continuous-mode` and
is easier to grep for?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
--
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3971032133
##
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.
+ *
+ * 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.
+ *
+ * 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 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(),
Review Comment:
This is tricky! `getSparkSchedulingConfigs` takes a `HoodieStreamer.Config`,
and all N tables set the same `hoodiedeltasync` pool. We will be looking at all
N ingests as a group vs all N compactions as a group. The PR is already large
and I think this pool modelling needs a separate PR. I'll create an issue for
this upon merge. Meanwhile I am adding a note in docs and also a warning at
start up.
Another observation is the generator also no-ops unless
`spark.scheduler.mode=FAIR` is set at submit time.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3970894156
##
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.
+ *
+ * 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.
+ *
+ * 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 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> 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) {
Review Comment:
Good analysis, you are right on all findings. Took your first option to drop
the throw and the terminated flag. The shutdownExecutor is now void and does
the reporting itself including a log on the interrupt branch, which previously
returned false silently.
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5597152580 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 708a73a6a6d828de9a7ffd5e1c06a9bc5364dd86 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16929) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3964726865
##
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.
+ *
+ * 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;
+
+ // lets awaitUntil be the one that reports a stall.
+ private static final long RELEASE_TIMEOUT_SECONDS = 300;
Review Comment:
🤖 nit: this comment reads like a truncated sentence — could you spell it
out, e.g. "Longer than the test's awaitUntil deadline, so a stall is reported
by the test rather than by this source"?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
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.
+ *
+ * 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
yihua commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3964570449
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java:
##
@@ -220,6 +220,19 @@ public void shutdownGracefully() {
});
}
+ /**
+ * 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.
+ */
+ public void interruptIngestion() {
+ingestionService.ifPresent(ds -> {
+ if (!ds.isShutdown()) {
+log.info("Forcefully shutting down DeltaStreamer");
+ds.shutdown(true);
Review Comment:
Is the forceful interrupt here deliberate over a graceful request?
shutdown(true) interrupts the sibling mid-round, so it is left with an inflight
instant to roll back on restart, whereas the ingestion loop already exits on
isShutdownRequested() between rounds and the executor wait in syncContinuously
would bound how long we let the current commit finish. If forceful is the
intended semantics it might be worth saying so in the --fail-fast-on-continuous
description so operators expect the rollbacks.
##
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.
+ *
+ * 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.
+ *
+ * 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 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> 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();
Review Comment:
With fail fast off, if every table eventually fails, sync() returns normally
here and main() exits 0, so an
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5595639335 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 2761811963e10b276c6d5f4efd1e732a092247d4 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16920) * 708a73a6a6d828de9a7ffd5e1c06a9bc5364dd86 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16929) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3964375148
##
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.
+ *
+ * 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.
+ *
+ * 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 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> 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5595348287 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 2761811963e10b276c6d5f4efd1e732a092247d4 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16920) * 708a73a6a6d828de9a7ffd5e1c06a9bc5364dd86 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3964365065
##
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 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 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:
Great Catch! Added the instanceof assertion. I checked it now and it fails
if unwrapCompletionException peels nothing.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3964357560
##
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.
+ *
+ * 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>, Checkpoint>
fetchNextBatch(Option lastCheckpoint, long sourceLimit) {
+// Only rendezvous
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3964346802
##
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.
+ *
+ * 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.
+ *
+ * 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 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> 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
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.
+ *
+ * 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.
+ *
+ * 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 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> 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5593188184 ## CI report: * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 2761811963e10b276c6d5f4efd1e732a092247d4 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16920) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5592564875 ## CI report: * c32721da12f4cc434a0fd7359037a800c457edee Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16919) * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 2761811963e10b276c6d5f4efd1e732a092247d4 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16920) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
vinothchandar commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5592184693 @yihua as well, to help -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5592238104 ## CI report: * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) * c32721da12f4cc434a0fd7359037a800c457edee Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16919) * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 2761811963e10b276c6d5f4efd1e732a092247d4 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16920) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5591917075 ## CI report: * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) * c32721da12f4cc434a0fd7359037a800c457edee Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16919) * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN * 2761811963e10b276c6d5f4efd1e732a092247d4 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5591638490 ## CI report: * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) * c32721da12f4cc434a0fd7359037a800c457edee Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16919) * b583c4695d05598fd612279cb24c95fe87def2ef UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3962042040
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -389,6 +406,12 @@ public static class Config implements Serializable {
+ " source-fetch -> Transform -> Hudi Write in loop")
public Boolean continuousMode = false;
+@Parameter(names = {"--fail-fast-on-continuous"},
+description = "Only applies in continuous mode. When enabled, the
failure of any single table sync fails the "
++ "whole job. The remaining table syncs are shut down and the
process exits with a non-zero status. When "
++ "disabled (default), each table is synced independently and a
single failure does not stop the others.")
+public Boolean failFastOnContinuousMode = false;
+
@Parameter(names = {"--min-sync-interval-seconds"},
Review Comment:
The flag follows the same --continuous / continuousMode alignment.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3961945090
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java:
##
@@ -220,6 +220,16 @@ public void shutdownGracefully() {
});
}
+ // Interrupts an in-progress ingestion, unlike shutdownGracefully() which
lets the current round finish.
+ public void shutdownForcefully() {
Review Comment:
Done, renamed the method. I like the wording - Neither closes nor waits - I
used the same in javadoc.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3962038658
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +484,194 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.info("Ingestion failed for topics: {}", failedTables);
+ }
+}
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
successTables.add(Helpers.getTableWithDatabase(context));
-streamer.shutdownGracefully();
} catch (Exception e) {
log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
failedTables.add(Helpers.getTableWithDatabase(context));
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, context);
}
}
}
+ }
+
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * 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 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> 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, streamerInstances, shutdownRequested);
+ } else {
+CompletableFuture.allOf(tableFutures.toArray(new
CompletableFuture[0])).join();
+ }
+} 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");
+}
+ }
+
+ /**
+ * 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 streamerInstances, AtomicBoolean shutdownRequested) {
+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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3962023136
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java:
##
@@ -243,6 +247,109 @@ 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 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 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");
+
+assertThrows(HoodieException.class, streamer::sync);
+assertFalse(streamer.getFailedTables().isEmpty());
+// 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 testFailFastOnContinuousAfterASiblingFinished() throws
IOException {
Review Comment:
Added `testFailFastOnContinuousWhenEveryTableFinishes` with fail fast and
both tables terminating via NoNewDataTerminationStrategy. Also asserts sync()
returns and both land in successTables.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5591498696 ## CI report: * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) * c32721da12f4cc434a0fd7359037a800c457edee Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16919) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3962004166
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,159 @@
+/*
+ * 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.
+ *
+ * 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 fails only after a sibling
has finished its sync.
+ public static final String FAIL_AFTER_SIBLING_COMPLETES =
"hoodie.test.continuous.source.fail.after.sibling.completes";
+
+ private static final long BARRIER_TIMEOUT_SECONDS = 60;
+ // Time for a finished sibling's future to complete before the waiting table
fails. Only widens the gap between
+ // the two completions, so a longer settle can never hide a regression.
+ private static final long SIBLING_SETTLE_MILLIS = 2000;
+
+ 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);
+ // Counted down when a table releases its source, which its ingestion
service does as the sync ends.
+ private static volatile CountDownLatch tableCompleted = new
CountDownLatch(1);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ private final boolean failAfterSiblingCompletes;
+ 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.failAfterSiblingCompletes =
props.getBoolean(FAIL_AFTER_SIBLING_COMPLETES, 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);
+tableCompleted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3961973667
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -389,6 +406,12 @@ public static class Config implements Serializable {
+ " source-fetch -> Transform -> Hudi Write in loop")
public Boolean continuousMode = false;
+@Parameter(names = {"--fail-fast-on-continuous"},
+description = "Only applies in continuous mode. When enabled, the
failure of any single table sync fails the "
++ "whole job. The remaining table syncs are shut down and the
process exits with a non-zero status. When "
++ "disabled (default), each table is synced independently and a
single failure does not stop the others.")
+public Boolean failFastOnContinuousMode = false;
+
@Parameter(names = {"--min-sync-interval-seconds"},
Review Comment:
🤖 nit: the CLI flag is `--fail-fast-on-continuous` but the field is
`failFastOnContinuousMode`; could you align them (e.g.
`--fail-fast-on-continuous-mode`) so it's easier to grep between the two?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +484,194 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.info("Ingestion failed for topics: {}", failedTables);
+ }
+}
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
successTables.add(Helpers.getTableWithDatabase(context));
-streamer.shutdownGracefully();
} catch (Exception e) {
log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
failedTables.add(Helpers.getTableWithDatabase(context));
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, context);
}
}
}
+ }
+
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * 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 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> 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, streamerInstances, shutdownRequested);
+ } else {
+CompletableFuture.allOf(tableFutures.toArray(new
CompletableFuture[0])).join();
+ }
+} 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5591429332 ## CI report: * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) * c32721da12f4cc434a0fd7359037a800c457edee UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3961930340
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +484,194 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.info("Ingestion failed for topics: {}", failedTables);
+ }
+}
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
successTables.add(Helpers.getTableWithDatabase(context));
-streamer.shutdownGracefully();
} catch (Exception e) {
log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
failedTables.add(Helpers.getTableWithDatabase(context));
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, context);
}
}
}
+ }
+
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * 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 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> 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, streamerInstances, shutdownRequested);
+ } else {
+CompletableFuture.allOf(tableFutures.toArray(new
CompletableFuture[0])).join();
+ }
+} 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");
+}
+ }
+
+ /**
+ * 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 streamerInstances, AtomicBoolean shutdownRequested) {
+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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
wombatu-kun commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3953809893
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,159 @@
+/*
+ * 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.
+ *
+ * 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 fails only after a sibling
has finished its sync.
+ public static final String FAIL_AFTER_SIBLING_COMPLETES =
"hoodie.test.continuous.source.fail.after.sibling.completes";
+
+ private static final long BARRIER_TIMEOUT_SECONDS = 60;
+ // Time for a finished sibling's future to complete before the waiting table
fails. Only widens the gap between
+ // the two completions, so a longer settle can never hide a regression.
+ private static final long SIBLING_SETTLE_MILLIS = 2000;
+
+ 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);
+ // Counted down when a table releases its source, which its ingestion
service does as the sync ends.
+ private static volatile CountDownLatch tableCompleted = new
CountDownLatch(1);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ private final boolean failAfterSiblingCompletes;
+ 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.failAfterSiblingCompletes =
props.getBoolean(FAIL_AFTER_SIBLING_COMPLETES, 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);
+tableCompleted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
wombatu-kun commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3953806057
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java:
##
@@ -220,6 +220,16 @@ public void shutdownGracefully() {
});
}
+ // Interrupts an in-progress ingestion, unlike shutdownGracefully() which
lets the current round finish.
+ public void shutdownForcefully() {
Review Comment:
Yes - shutdownGracefully closes and blocks, this one neither closes nor
waits, so the shared prefix is what makes it read as a peer. interruptIngestion
says what it does, and shutdownStreamers is its only caller so the rename costs
nothing.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
wombatu-kun commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3953797742
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieMultiTableDeltaStreamer.java:
##
@@ -243,6 +247,109 @@ 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 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 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");
+
+assertThrows(HoodieException.class, streamer::sync);
+assertFalse(streamer.getFailedTables().isEmpty());
+// 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 testFailFastOnContinuousAfterASiblingFinished() throws
IOException {
Review Comment:
Both fail-fast tests end in awaitFailFast's catch, so the branch where every
table terminates normally under fail fast is never exercised. Worth adding a
case with failFastOnContinuousMode on and NoNewDataTerminationStrategy on both
tables, asserting sync() returns and both tables land in successTables.
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +484,194 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.info("Ingestion failed for topics: {}", failedTables);
+ }
+}
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
successTables.add(Helpers.getTableWithDatabase(context));
-streamer.shutdownGracefully();
} catch (Exception e) {
log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
failedTables.add(Helpers.getTableWithDatabase(context));
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, context);
}
}
}
+ }
+
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3953488205
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,159 @@
+/*
+ * 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.
+ *
+ * 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 fails only after a sibling
has finished its sync.
+ public static final String FAIL_AFTER_SIBLING_COMPLETES =
"hoodie.test.continuous.source.fail.after.sibling.completes";
+
+ private static final long BARRIER_TIMEOUT_SECONDS = 60;
+ // Time for a finished sibling's future to complete before the waiting table
fails. Only widens the gap between
+ // the two completions, so a longer settle can never hide a regression.
+ private static final long SIBLING_SETTLE_MILLIS = 2000;
+
+ 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);
+ // Counted down when a table releases its source, which its ingestion
service does as the sync ends.
+ private static volatile CountDownLatch tableCompleted = new
CountDownLatch(1);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ private final boolean failAfterSiblingCompletes;
+ 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.failAfterSiblingCompletes =
props.getBoolean(FAIL_AFTER_SIBLING_COMPLETES, 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);
+tableCompleted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5575115470 ## CI report: * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5574647238 ## CI report: * 9850fc471826f9b301953731e8d81c508c8f50f2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16881) * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5574314040 ## CI report: * f117a834799f58f1b1a441aa65a2de282095c3e8 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16846) * 9850fc471826f9b301953731e8d81c508c8f50f2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16881) * 8aa620d62ff2b470a1e398aae576f755ef781723 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16883) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5573800702 ## CI report: * f117a834799f58f1b1a441aa65a2de282095c3e8 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16846) * 9850fc471826f9b301953731e8d81c508c8f50f2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16881) * 8aa620d62ff2b470a1e398aae576f755ef781723 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5573748310 ## CI report: * f117a834799f58f1b1a441aa65a2de282095c3e8 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16846) * 9850fc471826f9b301953731e8d81c508c8f50f2 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3951593026
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +484,186 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.info("Ingestion failed for topics: {}", failedTables);
+ }
+}
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
successTables.add(Helpers.getTableWithDatabase(context));
-streamer.shutdownGracefully();
} catch (Exception e) {
log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
failedTables.add(Helpers.getTableWithDatabase(context));
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, context);
}
}
}
+ }
+
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * 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 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> tableFutures =
tableExecutionContexts.stream()
+ .map(context -> CompletableFuture.runAsync(() -> {
Review Comment:
Done! Extracted to `runTableSync`
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3951577521
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -461,28 +484,186 @@ private static String resetTarget(Config configuration,
String database, String
/**
* Creates actual HoodieDeltaStreamer objects for every table/topic and does
incremental sync.
+ *
+ * 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.info("Ingestion failed for topics: {}", failedTables);
+ }
+}
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
successTables.add(Helpers.getTableWithDatabase(context));
-streamer.shutdownGracefully();
} catch (Exception e) {
log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
failedTables.add(Helpers.getTableWithDatabase(context));
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, context);
}
}
}
+ }
+
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * 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 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> tableFutures =
tableExecutionContexts.stream()
+ .map(context -> CompletableFuture.runAsync(() -> {
Review Comment:
🤖 nit: have you considered pulling this ~30-line lambda into a named method
like `runTableSync(context, streamerInstances, shutdownRequested)`?
`syncContinuously()` would then read as just setup → submit → await → teardown,
and the per-table registration/fail-fast logic gets its own javadoc.
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3951558177
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java:
##
@@ -220,6 +220,16 @@ public void shutdownGracefully() {
});
}
+ // Interrupts an in-progress ingestion, unlike shutdownGracefully() which
lets the current round finish.
+ public void shutdownForcefully() {
Review Comment:
This is was intentional. `shutdownStreamers()` calls this from the
coordinating thread while the sibling's `sync()` is still running, so closing
here would release the resources out from under a live writer. The thread that
ran `sync()` closes it in its finally, same as every other shutdownGracefully()
caller. Happy to rename this to something like `interruptIngestion()`, what do
you think?
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3951535682
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -479,10 +518,137 @@ 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.
+ *
+ * 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 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> 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));
Review Comment:
This is intended, but you're right that something was missing. I think the
actual gap was that the set gives no way to tell which table caused the
shutdown. I fixed that at the source by passing the table name for log.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3951501942
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -479,10 +518,137 @@ 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.
+ *
+ * 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 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> 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)).collect(Collectors.toList());
+
+ 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.toArray(new
CompletableFuture[0])).join();
+ }
+ log.info("Successful tables: {}, Failed tables: {}", successTables,
failedTables);
Review Comment:
Dropped it from `syncContinuously` and moved the `sync()` summary into
finally.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3951498730
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -479,10 +518,137 @@ 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.
+ *
+ * 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 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> 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();
Review Comment:
Good catch! Both finally blocks now go through a `shutdownQuietly()` helper
that logs instead of throwing, same as shutdownStreamers. Also dropped the
duplicate shutdownGracefully() on the sequential success path.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5557289849 ## CI report: * f117a834799f58f1b1a441aa65a2de282095c3e8 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16846) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5556771700 ## CI report: * ef059fb606bbe19a7907f16ee17b6415d215ca83 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16816) * f117a834799f58f1b1a441aa65a2de282095c3e8 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16846) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5556750373 ## CI report: * ef059fb606bbe19a7907f16ee17b6415d215ca83 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16816) * f117a834799f58f1b1a441aa65a2de282095c3e8 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
wombatu-kun commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3930340977
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -479,10 +518,137 @@ 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.
+ *
+ * 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 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> 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();
Review Comment:
A throw from shutdownGracefully() in this finally replaces the in-flight
exception, and SourceFormatAdapter.close does throw HoodieIOException when the
source fails to close. Wrap it in try/catch-and-log the way shutdownStreamers
already does.
##
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##
@@ -479,10 +518,137 @@ 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.
+ *
+ * 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 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> tableFutures =
tableExecutionContexts.stream()
+ .map(context ->
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5530519881 ## CI report: * c85cbc4a40c669dee7db0758f787f05339b6305a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16810) * ece48094405d6398d9967b2486cb4182e8498d97 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16812) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5533069926 ## CI report: * ef059fb606bbe19a7907f16ee17b6415d215ca83 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16816) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-553178 ## CI report: * ece48094405d6398d9967b2486cb4182e8498d97 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16812) * ef059fb606bbe19a7907f16ee17b6415d215ca83 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5532070469 ## CI report: * ece48094405d6398d9967b2486cb4182e8498d97 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16812) * ef059fb606bbe19a7907f16ee17b6415d215ca83 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16816) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3928622856
##
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.
+ *
+ * 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 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 exceptional completion, so a table that terminates
normally (e.g. via a
+ * {@link PostWriteTerminationStrategy}) does not abort its siblings.
+ */
+ private void awaitFailFast(CompletableFuture[] tableFutures,
List streamerInstances, AtomicBoolean shutdownRequested) {
+final CompletableFuture firstOutcome = new CompletableFuture<>();
+// Trip as soon as any table fails ...
+for (CompletableFuture tableFuture : tableFutures) {
+ tableFuture.whenComplete((result, throwable) -> {
+if (throwable != null) {
+
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3928645933
##
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:
Added `@Timeout(600)` to match the tests in TestHoodieDeltaStreamer.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3928639888
##
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.
+ *
+ * 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 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 exceptional completion, so a table that terminates
normally (e.g. via a
+ * {@link PostWriteTerminationStrategy}) does not abort its siblings.
+ */
+ private void awaitFailFast(CompletableFuture[] tableFutures,
List streamerInstances, AtomicBoolean shutdownRequested) {
+final CompletableFuture firstOutcome = new CompletableFuture<>();
+// Trip as soon as any table fails ...
+for (CompletableFuture tableFuture : tableFutures) {
+ tableFuture.whenComplete((result, throwable) -> {
+if (throwable != null) {
+
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3928603154
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,123 @@
+/*
+ * 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.
+ *
+ * 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";
+
+ 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);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ 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);
+ }
+
+ // Resets the shared barrier to expect one arrival per table, and the
interrupt latch. Call before each sync.
+ public static void resetBarrier(int numTables) {
+startBarrier = new CyclicBarrier(numTables);
+blockedTableInterrupted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option 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) {
+blockUntilInterrupted();
Review Comment:
Renamed the method.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3928598670
##
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.
+ *
+ * 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 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 exceptional completion, so a table that terminates
normally (e.g. via a
+ * {@link PostWriteTerminationStrategy}) does not abort its siblings.
+ */
+ private void awaitFailFast(CompletableFuture[] tableFutures,
List streamerInstances, AtomicBoolean shutdownRequested) {
+final CompletableFuture firstOutcome = new CompletableFuture<>();
Review Comment:
Thanks for your review! Good call, Delegated the calls to `FutureUtils`.
Added a small unwrap helper to isolate the CompletionException.
--
This is an automated message from the
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5531671286 ## CI report: * ece48094405d6398d9967b2486cb4182e8498d97 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16812) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5530434880 ## CI report: * c85cbc4a40c669dee7db0758f787f05339b6305a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16810) * ece48094405d6398d9967b2486cb4182e8498d97 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3927536818
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,123 @@
+/*
+ * 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.
+ *
+ * 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";
+
+ 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);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ 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);
+ }
+
+ // Resets the shared barrier to expect one arrival per table, and the
interrupt latch. Call before each sync.
+ public static void resetBarrier(int numTables) {
+startBarrier = new CyclicBarrier(numTables);
+blockedTableInterrupted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option 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) {
+blockUntilInterrupted();
Review Comment:
🤖 nit: the boolean field `blockUntilInterrupted` and this
`blockUntilInterrupted()` method share the exact same name, which is a bit
confusing to read at the call site. Could you rename the method to something
like `blockUntilFailFastInterrupts()` to distinguish the action from the flag?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
--
This is an automated message
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5529192854 ## CI report: * ab9c1bba165f11d66077ba25fe1a9bea5f5e6f82 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16774) * c85cbc4a40c669dee7db0758f787f05339b6305a UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5529264168 ## CI report: * ab9c1bba165f11d66077ba25fe1a9bea5f5e6f82 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16774) * c85cbc4a40c669dee7db0758f787f05339b6305a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16810) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
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.
+ *
+ * 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 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 exceptional completion, so a table that terminates
normally (e.g. via a
+ * {@link PostWriteTerminationStrategy}) does not abort its siblings.
+ */
+ private void awaitFailFast(CompletableFuture[] tableFutures,
List streamerInstances, AtomicBoolean shutdownRequested) {
+final CompletableFuture 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
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3920983905
##
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.
+ *
+ * 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 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(),
Review Comment:
🤖 When no tables are configured, `getTablesToBeIngested()` returns an empty
list (it returns `new ArrayList<>()` when `TABLES_TO_BE_INGESTED` is unset), so
`tableExecutionContexts` is empty and `Executors.newFixedThreadPool(0, ...)`
throws `IllegalArgumentException` here. The old sequential loop just no-oped on
an empty list — could you short-circuit the empty case before creating the pool?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
--
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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3915522838
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ *
+ * 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";
+
+ 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);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ 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);
+ }
+
+ // Resets the shared barrier to expect one arrival per table, and the
interrupt latch. Call before each sync.
+ public static void resetBarrier(int numTables) {
+startBarrier = new CyclicBarrier(numTables);
+blockedTableInterrupted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option 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) {
+blockUntilInterrupted();
+ }
+}
+return super.fetchNextBatch(lastCheckpoint, sourceLimit);
+ }
+
+ // Blocks until fail fast interrupts this table, records that it observed
the interrupt, then propagates it.
+ private void blockUntilInterrupted() {
+try {
+ new CountDownLatch(1).await();
Review Comment:
Fair point, Extracted it into a local var.
--
This is an automated message from the Apache Git Service.
To
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5513172632 ## CI report: * ab9c1bba165f11d66077ba25fe1a9bea5f5e6f82 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16774) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5511903015 ## CI report: * b41783da68a8257bc12d88e10c0e923ed5f76bf8 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16757) * ab9c1bba165f11d66077ba25fe1a9bea5f5e6f82 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16774) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-bot commented on PR #19811: URL: https://github.com/apache/hudi/pull/19811#issuecomment-5511808684 ## CI report: * b41783da68a8257bc12d88e10c0e923ed5f76bf8 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16757) * ab9c1bba165f11d66077ba25fe1a9bea5f5e6f82 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
smtwilio commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3915522838
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ *
+ * 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";
+
+ 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);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ 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);
+ }
+
+ // Resets the shared barrier to expect one arrival per table, and the
interrupt latch. Call before each sync.
+ public static void resetBarrier(int numTables) {
+startBarrier = new CyclicBarrier(numTables);
+blockedTableInterrupted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option 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) {
+blockUntilInterrupted();
+ }
+}
+return super.fetchNextBatch(lastCheckpoint, sourceLimit);
+ }
+
+ // Blocks until fail fast interrupts this table, records that it observed
the interrupt, then propagates it.
+ private void blockUntilInterrupted() {
+try {
+ new CountDownLatch(1).await();
Review Comment:
Fair point, Fexttacted it into a named local.
##
Re: [PR] feat(utilities): add fail-fast continuous mode for HoodieMultiTableStreamer [hudi]
hudi-agent commented on code in PR #19811:
URL: https://github.com/apache/hudi/pull/19811#discussion_r3912767852
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/ContinuousTestSource.java:
##
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ *
+ * 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";
+
+ 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);
+
+ private final boolean failAfterBarrier;
+ private final boolean blockUntilInterrupted;
+ 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);
+ }
+
+ // Resets the shared barrier to expect one arrival per table, and the
interrupt latch. Call before each sync.
+ public static void resetBarrier(int numTables) {
+startBarrier = new CyclicBarrier(numTables);
+blockedTableInterrupted = new CountDownLatch(1);
+ }
+
+ // Whether a blocking table has already observed the fail-fast interrupt.
+ public static boolean wasBlockedTableInterrupted() {
+return blockedTableInterrupted.getCount() == 0;
+ }
+
+ @Override
+ public Pair>, Checkpoint>
fetchNextBatch(Option 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) {
+blockUntilInterrupted();
+ }
+}
+return super.fetchNextBatch(lastCheckpoint, sourceLimit);
+ }
+
+ // Blocks until fail fast interrupts this table, records that it observed
the interrupt, then propagates it.
+ private void blockUntilInterrupted() {
+try {
+ new CountDownLatch(1).await();
Review Comment:
🤖 nit: blocking on a throwaway `new CountDownLatch(1).await()` that is never
counted down is a bit of an obscure
