voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957501394
##########
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) {
Review Comment:
Fixed in `d74dd15`. `stopLeakedStreamer` now takes the bound as a parameter,
the same seam `waitTillCondition` has for its poll interval, and two cases
drive it at 1s: `stopThatThrowsStillCancelsTheIngestTask` and
`stopThatHangsIsBoundedAndCancelsTheIngestTask`, both asserting the ingest task
ends up cancelled. Confirmed they discriminate: removing the stop-threw
fall-through fails the first and leaves the second passing.
##########
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
Review Comment:
Fixed in `d74dd15`. New
`directInterruptEndsTheWaitWithoutRunningToTheTimeout` interrupts the poller
while the executor is still live, so the branch is the only thing that can end
the wait. Confirmed: folding that catch into the catch-all fails it after the
full 10s while `pollingStopsOnceTheWaitHasGivenUp` still passes, which is
exactly the gap you flagged. Its javadoc no longer claims otherwise.
--
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]
