voonhous commented on code in PR #19485: URL: https://github.com/apache/hudi/pull/19485#discussion_r3957508101
########## hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java: ########## @@ -0,0 +1,268 @@ +/* + * 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.deltastreamer; + +import org.apache.hudi.common.testutils.JavaTestUtils; +import org.apache.hudi.utilities.streamer.NoNewDataTerminationStrategy; + +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerTestBase.TestHelpers.describeTimeout; +import static org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerTestBase.TestHelpers.waitFor; +import static org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Covers the deltastreamer test helpers every continuous-mode test runs on: the wait in + * {@code HoodieDeltaStreamerTestBase.TestHelpers} and the runner in {@code TestHoodieDeltaStreamer}. + * + * <p>The wait used to fail with a bare {@code TimeoutException} naming only the helper, with the + * condition's own error logged at debug and discarded, so a timeout said nothing about which assertion + * never held (HUDI-6843). + */ +class TestDeltaStreamerTestHelpers { + + /** A deltastreamer future that never finishes, as a continuous-mode job would be. */ + private static final Future<?> RUNNING = new CompletableFuture<>(); + + /** + * The poll interval these tests drive the helper at, so the class does not spend the production 2s cadence + * asleep. + */ + private static final long FAST_POLL_INTERVAL_MS = 50; + + /** + * With the fast poll above, one second still leaves room for many evaluations to be recorded, which is what + * the timeout report needs. + */ + private static final int CONDITION_TIMEOUT_SECS = 1; + + /** For the cases that are not meant to time out: they finish long before this, so it is never reached. */ + private static final int NEVER_REACHED_TIMEOUT_SECS = 30; + + @Test + void timeoutFailureNamesTheLastConditionFailure() { + String assertionText = "assertAtleastNDeltaCommits: expected at least 3 delta commits but got 2"; + + AssertionError error = assertThrows(AssertionError.class, + () -> waitTillCondition( + ignored -> { + throw new AssertionError(assertionText); + }, RUNNING, CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS)); + + assertTrue(error.getMessage().contains("was not met within " + CONDITION_TIMEOUT_SECS + " seconds"), + () -> "The failure should say the condition timed out, but was: " + error.getMessage()); + assertTrue(error.getMessage().contains(assertionText), + () -> "The failure should carry the condition's own error, which is the only clue to why the " + + "wait timed out, but was: " + error.getMessage()); + assertFalse(error.getMessage().contains("returned false without throwing"), + () -> "The failure should carry the condition's error, not the 'kept returning false' branch, " + + "but was: " + error.getMessage()); + assertInstanceOf(TimeoutException.class, error.getSuppressed()[0], + "the timeout should stay attached as a suppressed exception once the condition's error becomes the cause"); + } + + /** + * {@code shutdownNow} interrupts the polling thread, but {@code Thread.sleep} clears the interrupt flag + * when it throws, so a catch-all around the sleep would swallow it and keep polling for the life of the + * JVM. This pins that the worker actually stops. + */ + @Test + void pollingStopsOnceTheWaitHasGivenUp() throws Exception { + AtomicInteger polls = new AtomicInteger(); + AtomicReference<Thread> poller = new AtomicReference<>(); + + assertThrows(AssertionError.class, + () -> waitTillCondition( + ignored -> { + poller.set(Thread.currentThread()); + polls.incrementAndGet(); + throw new AssertionError("never true"); + }, RUNNING, CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS)); + + int pollsWhenItGaveUp = polls.get(); + assertTrue(pollsWhenItGaveUp > 0, + "the condition should have been evaluated at least once before the wait gave up, otherwise the " + + "comparison below passes trivially"); + poller.get().join(TimeUnit.SECONDS.toMillis(5)); + assertFalse(poller.get().isAlive(), + "the polling thread should have exited once the wait gave up, not still be running after the join"); + assertEquals(pollsWhenItGaveUp, polls.get(), + "the polling thread should have stopped when the wait gave up, not carried on in the background"); + } + + /** + * The interrupt from {@code shutdownNow} is delivered once, and {@code Thread.sleep} clears the flag when it + * throws, so a condition that swallows it without restoring it leaves the loop with no interrupt to see. The + * {@code executor.isShutdown()} guard is what stops the worker in that case. + */ + @Test + void pollingStopsEvenWhenTheConditionSwallowsTheInterrupt() throws Exception { + AtomicInteger polls = new AtomicInteger(); + AtomicReference<Thread> poller = new AtomicReference<>(); + + assertThrows(AssertionError.class, + () -> waitTillCondition( + ignored -> { + poller.set(Thread.currentThread()); + polls.incrementAndGet(); + try { + Thread.sleep(TimeUnit.SECONDS.toMillis(60)); + } catch (InterruptedException interrupted) { + // The missing Thread.currentThread().interrupt() is the point of the test: a condition that + // swallows the interrupt is exactly what the isShutdown() guard exists for, so do not "fix" + // this catch. + } + return false; + }, RUNNING, CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS)); + + int pollsWhenItGaveUp = polls.get(); + assertTrue(pollsWhenItGaveUp > 0, + "the condition should have been evaluated at least once before the wait gave up, otherwise the " + + "comparison below passes trivially"); + poller.get().join(TimeUnit.SECONDS.toMillis(5)); + assertFalse(poller.get().isAlive(), + "the isShutdown() guard should have stopped the polling thread even though the condition swallowed " + + "the interrupt without restoring the flag"); + assertEquals(pollsWhenItGaveUp, polls.get(), + "the polling thread should have stopped when the wait gave up, not carried on in the background"); + } + + /** + * A condition that hangs part-way through its first evaluation is a different failure from one that keeps + * returning false, and the report has to say which: with no completed evaluation there is no last error, + * and claiming the condition "returned false without throwing" would assert the wrong thing. + */ + @Test + void timeoutDistinguishesAConditionThatNeverCompletedAnEvaluation() { + AssertionError error = assertThrows(AssertionError.class, + () -> waitTillCondition( + ignored -> { + try { + Thread.sleep(60_000); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } + return true; + }, RUNNING, CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS)); + + assertTrue(error.getMessage().contains("No evaluation of the condition completed"), + () -> "a condition still running its first evaluation should be reported as such, but was: " + + error.getMessage()); + assertFalse(JavaTestUtils.checkNestedExceptionContains(error, "no such text"), Review Comment: Fixed in `d74dd15`, replaced with `assertInstanceOf(TimeoutException.class, error.getCause())` plus `assertNull(error.getCause().getMessage())`, so it pins the message-less throwable this path attaches and leaves the walk itself to `TestJavaTestUtils`. ########## 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: Good catch, taken in `d74dd15`. Confirmed the mechanism: `HoodieAsyncService.shutdown(false)` catches the interrupt from `shutdownNow()` without restoring the flag, and `HoodieStreamer.shutdownGracefully` (:214-221) then runs `ds.close()` regardless. The finally now awaits the stopper for the same bound and warns if the close has not finished, so it no longer runs on into the next test's setup. -- 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]
