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.
+ *
+ * <p>In continuous mode each table's sync blocks until it is shut down, so
the tables are synced concurrently.
+ * Otherwise the tables are synced sequentially, one after another.
*/
public void sync() {
+ try {
+ if (continuousMode) {
+ syncContinuously();
+ } else {
+ syncSequentially();
+ }
+ } finally {
+ log.info("Ingestion was successful for topics: {}", successTables);
+ if (!failedTables.isEmpty()) {
+ log.error("Ingestion failed for topics: {}", failedTables);
+ }
+ }
+ }
+
+ private void syncSequentially() {
for (TableExecutionContext context : tableExecutionContexts) {
+ String table = Helpers.getTableWithDatabase(context);
HoodieStreamer streamer = null;
try {
streamer = new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties()));
streamer.sync();
- successTables.add(Helpers.getTableWithDatabase(context));
- streamer.shutdownGracefully();
+ successTables.add(table);
} catch (Exception e) {
- log.error("error while running MultiTableDeltaStreamer for table: {}",
context.getTableName(), e);
- failedTables.add(Helpers.getTableWithDatabase(context));
+ log.error("error while running MultiTableDeltaStreamer for table: {}",
table, e);
+ failedTables.add(table);
} finally {
if (streamer != null) {
- streamer.shutdownGracefully();
+ shutdownQuietly(streamer, table);
}
}
}
+ }
+
+ /**
+ * Syncs all tables concurrently, one thread per table. Used for continuous
mode where each table's sync blocks
+ * indefinitely.
+ *
+ * <p>When {@code --fail-fast-on-continuous} is enabled, the first table
failure fails the whole job. The sibling
+ * streamers are shut down and a {@link HoodieException} is thrown so the
caller can exit with a non-zero status.
+ * Otherwise, every table is synced independently and a single failure does
not affect the others.
+ */
+ private void syncContinuously() {
+ if (tableExecutionContexts.isEmpty()) {
+ return;
+ }
+ // Streamer instances are registered from worker threads, so a thread-safe
list is required.
+ final List<HoodieStreamer> streamerInstances = new
CopyOnWriteArrayList<>();
+ // Set once fail fast trips, so tasks that register their streamer
afterwards stop before starting the sync.
+ final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
+ final ExecutorService executor =
Executors.newFixedThreadPool(tableExecutionContexts.size(),
+ new CustomizedThreadFactory("multi-table-streamer", true));
+ try {
+ final List<CompletableFuture<Void>> tableFutures =
tableExecutionContexts.stream()
+ .map(context -> CompletableFuture.runAsync(
+ () -> runTableSync(context, streamerInstances,
shutdownRequested), executor))
+ .collect(Collectors.toList());
+
+ if (failFastOnContinuousMode) {
+ log.info("Fail fast enabled in continuous mode. The whole job fails on
any single table failure");
+ awaitFailFast(tableFutures);
+ } else {
+ CompletableFuture.allOf(tableFutures.toArray(new
CompletableFuture[0])).join();
+ }
+ } finally {
+ // On an abnormal exit the siblings are still ingesting, since
FutureUtils.allOf only cancels their futures.
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.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
