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.
+ *
+ * <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));
+ boolean terminated = false;
+ try {
+ final List<CompletableFuture<Void>> 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.
+ *
+ * <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));
+ boolean terminated = false;
+ try {
+ final List<CompletableFuture<Void>> 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:
A sibling that fail-fast interrupts also lands in failedTables, because this
catch has no shutdownRequested guard like the successTables side does - in
testFailFastOnContinuousThrowsWhenATableFails both tables end up in the set. Is
that intended, or should the tables we tore down stay out so the one that
actually failed is identifiable?
##########
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.
+ *
+ * <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));
+ boolean terminated = false;
+ try {
+ final List<CompletableFuture<Void>> 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:
This duplicates the summary sync() already logs on the success path, and the
fail-fast throw skips both, so the breakdown is logged twice when it does not
matter and never when it does. Drop it here, or move it ahead of the rethrow in
awaitFailFast - follow-up, not a blocker.
##########
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:
shutdownForcefully skips the ds.close() that shutdownGracefully does, so the
write client, format adapter, embedded timeline server and metrics reporter are
released only if a shutdownGracefully() follows. Is the contract that callers
must always pair the two, or should this close() as well?
--
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]
