hudi-agent commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957259470
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -1744,22 +1749,100 @@ static void
deltaStreamerTestRunner(HoodieDeltaStreamer ds, HoodieDeltaStreamer.
static void deltaStreamerTestRunner(HoodieDeltaStreamer ds,
HoodieDeltaStreamer.Config cfg, Function<Boolean, Boolean> condition, String
jobId) throws Exception {
ExecutorService executor = Executors.newSingleThreadExecutor();
- Future dsFuture = executor.submit(() -> {
+ Future dsFuture = null;
+ boolean stoppedCleanly = false;
+ try {
+ dsFuture = executor.submit(() -> {
+ try {
+ ds.sync();
+ } catch (Exception ex) {
+ log.warn("DS continuous job failed, hence not proceeding with
condition check for {}", jobId);
+ throw new RuntimeException(ex.getMessage(), ex);
+ }
+ });
+ TestHelpers.waitTillCondition(condition, dsFuture, 360);
+ if (cfg != null && !cfg.postWriteTerminationStrategyClass.isEmpty()) {
+ // If the streamer died, waitTillCondition returns as soon as the
future completes. Surface that
+ // failure here rather than letting awaitDeltaStreamerShutdown time
out and report the misleading
+ // "Deltastreamer should have shutdown by now" two minutes later.
+ if (dsFuture.isDone()) {
+ dsFuture.get();
+ }
+ awaitDeltaStreamerShutdown(ds);
+ } else {
+ ds.shutdownGracefully();
+ dsFuture.get();
+ }
+ stoppedCleanly = true;
+ } finally {
+ if (!stoppedCleanly) {
+ try {
+ stopLeakedStreamer(ds, dsFuture);
+ } catch (Throwable cleanupFailure) {
+ // Never let the cleanup replace the failure the caller is already
propagating.
+ log.warn("Failed to stop the streamer after a failure",
cleanupFailure);
+ }
+ }
+ executor.shutdown();
+ }
+ }
+
+ /**
+ * Stops a streamer that a failure left running, without letting the stop
hang the test.
+ * <p>
+ * Surefire runs this module with forkCount=1 and reuseForks=true, so a live
streamer reads on into the
+ * next test, whose setup deletes basePath and whose teardown closes the
data generators underneath it.
+ * The stop has to be bounded: shutdownGracefully awaits the ingest executor
for up to 24 hours, and it
+ * returns immediately without waiting when shutdown was already requested,
so neither the wait nor the
+ * absence of one can be relied on here.
+ */
+ private static void stopLeakedStreamer(HoodieDeltaStreamer ds, Future
dsFuture) {
+ ExecutorService stopper = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> stop = stopper.submit(ds::shutdownGracefully);
try {
- ds.sync();
- } catch (Exception ex) {
- log.warn("DS continuous job failed, hence not proceeding with
condition check for {}", jobId);
- throw new RuntimeException(ex.getMessage(), ex);
+ stop.get(STREAMER_STOP_TIMEOUT_SECS, TimeUnit.SECONDS);
+ } catch (ExecutionException stopThrew) {
+ // The stop itself failing does not excuse leaving the ingest task
running, so fall through to the join
+ // below rather than take the outer clause, which tolerates only the
ingest task's own failure.
+ log.warn("Stopping the streamer threw after a failure", stopThrew);
}
- });
- TestHelpers.waitTillCondition(condition, dsFuture, 360);
- if (cfg != null && !cfg.postWriteTerminationStrategyClass.isEmpty()) {
- awaitDeltaStreamerShutdown(ds);
- } else {
- ds.shutdownGracefully();
- dsFuture.get();
+ if (dsFuture != null) {
+ dsFuture.get(STREAMER_STOP_TIMEOUT_SECS, TimeUnit.SECONDS);
+ }
+ } catch (ExecutionException ingestFailure) {
+ // Expected rather than anomalous: the ingest task failing is usually
why the caller is unwinding at
+ // all, and the caller reports it. Nothing to warn about here.
+ } catch (Exception stopFailure) {
+ // Swallowed on purpose: this runs while another failure is propagating,
and replacing that failure
+ // with this one would hide the diagnostic the caller is about to report.
+ if (stopFailure instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ log.warn("Could not stop the streamer cleanly after a failure,
cancelling the ingest task", stopFailure);
+ // The 60s bound only stops this thread waiting:
HoodieAsyncService.shutdown(false) swallows the interrupt
+ // that stopper.shutdownNow() sends, and
HoodieStreamer.shutdownGracefully runs ds.close() regardless, so
+ // without forcing the executor down the write client can close under a
still-running ingest round.
+ forceStopIngestion(ds);
+ if (dsFuture != null) {
+ dsFuture.cancel(true);
+ }
+ } finally {
+ stopper.shutdownNow();
Review Comment:
🤖 On the timeout branch, `shutdownNow()` here interrupts the stopper out of
`awaitTermination`, and `HoodieAsyncService.shutdown(false)` swallows that
interrupt, so `shutdownGracefully` goes on to run `ds.close()` on the stopper
thread *after* this method has returned and the next test's setup may already
be deleting basePath. Would a short bounded `stopper.awaitTermination(...)`
after the `shutdownNow()` be worth it, so `close()` at least gets a chance to
finish before the runner hands control back?
<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]
