voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956902031
##########
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:
**major:** This branch, `forceStopIngestion` (:1826) and
`dsFuture.cancel(true)` (:1827) are reached by no test. The only test entering
`stopLeakedStreamer` is
`dyingStreamerWithTerminationStrategyIsSurfacedNotWaitedOut`, whose mock makes
`shutdownGracefully()` a no-op, so `stop.get()` returns normally and the outer
clause is never taken. `STREAMER_STOP_TIMEOUT_SECS = 60` is hardcoded, so a
test costs 60s.
I added this in `ae6065c` and `9f3cd75`, so this is mine to fix. Could
`stopLeakedStreamer` take the stop bound as a parameter, the same seam
`9f3cd75` added for `pollIntervalMs`, so a test can drive it at ~200ms?
##########
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:
**major:** This javadoc says the test pins the `InterruptedException`
branch, but it does not: turning that catch into a catch-all leaves this test
green, because the `!executor.isShutdown()` guard stops the worker on its own.
So the interrupt branch has no discriminating test.
The claim was true when written. The guard landed later in `6b07137`, from
my own suggestion in r3916671595, and silently made it stale. Could the test
interrupt the poller thread directly while the executor is still live, so it
discriminates the branch again?
##########
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();
Review Comment:
**minor:** Not blocking. On the failure branch this adds nothing:
`dsFuture.cancel(true)` already delivered the single interrupt, and an ingest
task that swallows it keeps the pool thread alive regardless. The PR's own
reasoning at `HoodieDeltaStreamerTestBase.java:797` argues for `shutdownNow()`
in exactly this case, and `waitTillCondition`'s finally uses it.
Could the `!stoppedCleanly` branch use `shutdownNow()`, keeping `shutdown()`
for the clean exit?
##########
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"),
+ "walking the cause chain has to tolerate the null-message
TimeoutException this path attaches, "
+ + "which is what the multi-writer test hits when its ingestion
wait times out");
+ }
+
+ /**
+ * Conditions in the continuous-mode tests catch their own failures and
return false rather than throwing,
+ * so this is the branch a real timeout reports. It has to say how many
evaluations ran, since that is the
+ * only signal separating it from a condition that never completed one.
+ */
+ @Test
+ void timeoutReportsEvaluationsThatReturnedFalse() {
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> waitTillCondition(ignored -> false, RUNNING,
CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS));
+
+ assertTrue(error.getMessage().contains("returned false without throwing"),
Review Comment:
**minor:** Not blocking. The javadoc above says the report "has to say how
many evaluations ran", but this assertion never inspects the count, so a
mutation formatting it as a constant `0` would still pass.
Could we add `assertFalse(error.getMessage().contains("0 evaluations
completed"))`? It cannot add flake: if zero evaluations had completed the
message would take the other branch and the existing assertion would already
fail.
##########
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:
**minor:** Not blocking. This assertion is off-topic for the test's name and
now duplicates `TestJavaTestUtils:35`. Post-fix it can only fail via an NPE
from a reverted `JavaTestUtils`, and it never checks that this path actually
attaches a message-less throwable.
Could we replace it with `assertInstanceOf(TimeoutException.class,
error.getCause())` plus `assertNull(error.getCause().getMessage())`, pinning
the coupling that makes the null check necessary and leaving the walk itself to
`TestJavaTestUtils`?
##########
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"),
+ "walking the cause chain has to tolerate the null-message
TimeoutException this path attaches, "
+ + "which is what the multi-writer test hits when its ingestion
wait times out");
+ }
+
+ /**
+ * Conditions in the continuous-mode tests catch their own failures and
return false rather than throwing,
+ * so this is the branch a real timeout reports. It has to say how many
evaluations ran, since that is the
+ * only signal separating it from a condition that never completed one.
+ */
+ @Test
+ void timeoutReportsEvaluationsThatReturnedFalse() {
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> waitTillCondition(ignored -> false, RUNNING,
CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS));
+
+ assertTrue(error.getMessage().contains("returned false without throwing"),
+ () -> "a condition that kept returning false should be reported as
such, but was: " + error.getMessage());
+ }
+
+ /**
+ * The bound exists so a hung poll cannot run for the life of the JVM. Both
production callers of waitFor
+ * are currently disabled (HUDI-8951), so this is the only thing exercising
it.
+ */
+ @Test
+ void waitForGivesUpAtItsBound() {
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> waitFor(() -> false, 1));
+
+ assertTrue(error.getMessage().contains("did not hold within 1 seconds"),
+ () -> "the bound should name itself in the failure, but was: " +
error.getMessage());
+ }
+
+ /**
+ * When a streamer configured with a post-write termination strategy dies,
the wait returns because the
+ * future is done, and {@code deltaStreamerTestRunner} has to surface that
failure. Without the
+ * {@code dsFuture.isDone()} guard it would instead call {@code
awaitDeltaStreamerShutdown} and report the
+ * misleading "Deltastreamer should have shutdown by now" two minutes later
- here, on a mock with no
+ * ingestion service, it would NPE.
+ */
+ @Test
+ void dyingStreamerWithTerminationStrategyIsSurfacedNotWaitedOut() throws
Exception {
+ HoodieDeltaStreamer ds = Mockito.mock(HoodieDeltaStreamer.class);
+ Mockito.doThrow(new IllegalStateException("source is
unreachable")).when(ds).sync();
+ HoodieDeltaStreamer.Config cfg = new HoodieDeltaStreamer.Config();
+ cfg.postWriteTerminationStrategyClass =
NoNewDataTerminationStrategy.class.getName();
+
+ ExecutionException failure = assertThrows(ExecutionException.class,
+ () -> TestHoodieDeltaStreamer.deltaStreamerTestRunner(ds, cfg, ignored
-> false, "dying_ds_job"));
+
+ assertTrue(JavaTestUtils.checkNestedExceptionContains(failure, "source is
unreachable"),
+ () -> "the streamer's own failure should be surfaced, but was: " +
failure);
+ }
+
+ @Test
+ void satisfiedConditionReturnsNormally() {
+ assertDoesNotThrow(() -> waitTillCondition(
+ ignored -> true, RUNNING, NEVER_REACHED_TIMEOUT_SECS,
FAST_POLL_INTERVAL_MS));
+ }
+
+ /**
+ * When the streamer finishes first the wait returns rather than failing,
and the caller
+ * ({@code deltaStreamerTestRunner}) surfaces the streamer's own outcome.
Pinned so the timeout handling
+ * above does not turn this into a failure.
+ */
+ @Test
+ void finishedStreamerEndsTheWaitWithoutFailing() {
+ Future<?> finished = CompletableFuture.completedFuture(null);
+
+ assertDoesNotThrow(() -> waitTillCondition(
+ ignored -> false, finished, NEVER_REACHED_TIMEOUT_SECS,
FAST_POLL_INTERVAL_MS));
+ }
+
+ /**
+ * Unreachable through the helper, which reads the evaluation counter before
the last error and so can see a
Review Comment:
**nit:** Feel free to ignore. This says "Unreachable through the helper" and
then describes the interleaving by which the helper does reach it (counter read
before the error, worker writing them the other way round). The intended claim
seems to be that no test can drive it deterministically.
Could we say "Not deterministically reproducible through the helper" instead?
##########
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.
Review Comment:
**nit:** Feel free to ignore. `HoodieAsyncService.shutdown(true)` only calls
`executor.shutdownNow()` and does not wait (`HoodieAsyncService.java:106-112`),
and a Spark job is not interrupt-responsive, so forcing the executor down
narrows this window rather than closing it.
Could we soften to "so the ingest round is at least interrupted before the
close"?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +776,131 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+ /**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * <p>On timeout the last error the condition threw is attached to the
failure, so the report names the
+ * assertion that never held rather than only this method.
+ */
static void waitTillCondition(Function<Boolean, Boolean> condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future<Boolean> res = Executors.newSingleThreadExecutor().submit(() -> {
- boolean ret = false;
- while (!ret && !dsFuture.isDone()) {
- try {
- Thread.sleep(2000);
- ret = condition.apply(true);
- log.info("Condition completed successfully");
- } catch (Throwable error) {
- log.debug("Got error waiting for condition", error);
- ret = false;
+ waitTillCondition(condition, dsFuture, timeoutInSecs, POLL_INTERVAL_MS);
+ }
+
+ /** The poll interval is a parameter only so this helper's own tests need
not spend the production cadence. */
+ static void waitTillCondition(Function<Boolean, Boolean> condition, Future
dsFuture, long timeoutInSecs,
+ long pollIntervalMs) throws Exception {
+ AtomicReference<Throwable> lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<Boolean> res = executor.submit(() -> {
+ boolean ret = false;
+ // The executor check matters as well as the interrupt flag: the
interrupt from shutdownNow is
+ // delivered once, and a condition that swallows it would otherwise
leave the flag clear and keep
+ // this thread polling for the lifetime of the JVM.
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted() && !executor.isShutdown()) {
+ try {
+ Thread.sleep(pollIntervalMs);
+ ret = condition.apply(true);
+ completedEvaluations.incrementAndGet();
+ if (ret) {
+ log.info("Condition completed successfully");
+ }
+ } catch (InterruptedException interrupted) {
+ // Thread.sleep clears the interrupt flag when it throws, so
catching this with everything
+ // else would re-enter the loop. Restore the flag and stop; this
is not a condition failure,
+ // so it is deliberately not recorded as one.
+ Thread.currentThread().interrupt();
+ break;
+ } catch (Throwable error) {
+ log.debug("Got error waiting for condition", error);
+ lastError.set(error);
+ completedEvaluations.incrementAndGet();
+ ret = false;
+ }
+ }
+ return ret;
+ });
+ try {
+ Boolean satisfied = res.get(timeoutInSecs, TimeUnit.SECONDS);
+ // Not a failure - the caller surfaces the streamer's own outcome -
but the wait should still say
+ // what it was waiting for instead of looking like success.
+ if (!Boolean.TRUE.equals(satisfied)) {
+ log.warn("Wait ended because the deltastreamer future finished,
not because the condition held. {}",
+ describeProgress(lastError.get(), completedEvaluations.get()));
Review Comment:
**minor:** Not blocking. Arguments evaluate left to right, so this reads
`lastError` first and the counter second, the exact order the comment at :862
says would "deny an error that did happen". The timeout path at :833 hoists the
counter first.
It is harmless here only because `res.get()` returning establishes
happens-before with the worker, which the comment does not say. Could we hoist
both into locals counter-first, or add that one clause to the comment?
##########
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"),
+ "walking the cause chain has to tolerate the null-message
TimeoutException this path attaches, "
+ + "which is what the multi-writer test hits when its ingestion
wait times out");
+ }
+
+ /**
+ * Conditions in the continuous-mode tests catch their own failures and
return false rather than throwing,
+ * so this is the branch a real timeout reports. It has to say how many
evaluations ran, since that is the
+ * only signal separating it from a condition that never completed one.
Review Comment:
**major:** This says the continuous-mode conditions "catch their own
failures and return false rather than throwing, so this is the branch a real
timeout reports". For #16228 the opposite holds: the prep condition at
`TestHoodieDeltaStreamerWithMultiWriter.java:124` only ever throws, since
`assertAtleastNDeltaCommits` uses JUnit `assertTrue`
(`HoodieDeltaStreamerTestBase.java:723,731`). Only `testHoodieIndexer` (:2233)
and :2565 return false.
This also contradicts paragraph 3 of the PR body. Could we reword it to say
#16228 takes the last-error branch, which is the stronger result for this PR?
##########
hudi-common/src/test/java/org/apache/hudi/common/testutils/TestJavaTestUtils.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.common.testutils;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Pins the null-message handling of {@link
JavaTestUtils#checkNestedExceptionContains}: a throwable
+ * with no message must neither NPE the walk nor match an errorMsg of "null".
The helper's callers all
+ * live in other modules, so the test lives beside the helper to keep it
covered where it is defined.
+ */
+public class TestJavaTestUtils {
+
+ @Test
+ public void testNullMessageOnHeadStillMatchesDeeperCause() {
+ // Pre-fix this NPE'd on the head. The explicit (String) null cast
matters: new RuntimeException(cause)
+ // would set the message to the cause's toString and hide the null-message
case entirely.
+ Throwable t = new RuntimeException((String) null, new
IllegalStateException("boom"));
+ assertTrue(JavaTestUtils.checkNestedExceptionContains(t, "boom"));
+ }
+
+ @Test
+ public void testNullMessageMidChainDoesNotStopTheWalk() {
Review Comment:
**nit:** Feel free to ignore. This and
`testNullMessageOnHeadStillMatchesDeeperCause` exercise the identical `message
!= null` check; the walk has no head-vs-mid special case, so no mutation
separates them.
Could we keep one of the two and add a `checkNestedExceptionContains(null,
"x")` case instead? The `throwable != null` loop guard is the one shape of this
four-line helper that nothing covers.
##########
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],
Review Comment:
**nit:** Feel free to ignore. If the suppressed timeout is ever dropped this
throws `ArrayIndexOutOfBoundsException` rather than failing with the message
above.
```suggestion
assertEquals(1, error.getSuppressed().length,
"the timeout should stay attached as a suppressed exception once the
condition's error becomes the cause");
assertInstanceOf(TimeoutException.class, error.getSuppressed()[0],
```
##########
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);
Review Comment:
**nit:** Feel free to ignore. Line 141 writes the same duration as
`TimeUnit.SECONDS.toMillis(60)`.
```suggestion
Thread.sleep(TimeUnit.SECONDS.toMillis(60));
```
--
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]
