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.
+ *
+ * <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);
}
}
}
+ }
- 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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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<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 {
+ // 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 nothing was rethrown above. Continuous mode is not
meant to end, so returning here with a
+ // failed table would exit 0 and tell an orchestrator that nothing is
wrong while nothing is ingesting.
if (!failedTables.isEmpty()) {
- log.info("Ingestion failed for topics: {}", failedTables);
+ throw new HoodieException("Continuous mode ended with failed tables: " +
failedTables);
+ }
+ }
+
+ /**
+ * Warns when several tables share a SparkContext that schedules FIFO, where
one table's job holds the cluster
+ * until it finishes and the others wait behind it. FAIR has to be set at
submit time to interleave them.
+ */
+ private void warnIfSchedulerIsNotFair() {
+ if (tableExecutionContexts.size() < 2) {
+ return;
+ }
+ String schedulerMode =
jssc.getConf().get(SchedulerConfGenerator.SPARK_SCHEDULER_MODE_KEY, "FIFO");
+ if
(!SchedulerConfGenerator.SPARK_SCHEDULER_FAIR_MODE.equalsIgnoreCase(schedulerMode))
{
+ log.warn("Syncing {} tables concurrently on a SparkContext with {}={}.
One table's job will hold the cluster "
+ + "until it completes while the others queue behind it; set {}=FAIR
at submit time to interleave them.",
+ tableExecutionContexts.size(),
SchedulerConfGenerator.SPARK_SCHEDULER_MODE_KEY, schedulerMode,
+ SchedulerConfGenerator.SPARK_SCHEDULER_MODE_KEY);
+ }
+ }
+
+ /**
+ * Syncs one table on the calling worker thread. Rethrows only under fail
fast; otherwise the failure is recorded
+ * in {@link #failedTables} and the sibling tables carry on.
+ */
+ private void runTableSync(TableExecutionContext context,
List<HoodieStreamer> streamerInstances, AtomicBoolean shutdownRequested) {
Review Comment:
🤖 nit: `streamerInstances` is filled in from inside this method (an output
argument) and `shutdownRequested` is shared state too — have you considered
making both instance fields (like `successTables`/`failedTables`) so
`runTableSync(context)` and `interruptAllIngestion()` don't need them threaded
through as parameters?
<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]
