voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3954798810
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -1744,22 +1744,43 @@ 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(() -> {
+ try {
+ Future 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);
+ }
+ });
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);
+ } catch (Throwable failure) {
+ // Surefire runs this module with forkCount=1 and reuseForks=true, so
a continuous streamer left
+ // running here reads on into the next test, whose setup deletes
basePath and whose teardown closes
+ // the data generators underneath it. Stop it before letting the
failure out.
+ try {
+ ds.shutdownGracefully();
Review Comment:
**major:** This stop is unbounded, so it can swallow the diagnostic this PR
adds. `shutdownGracefully()` reaches `HoodieAsyncService.shutdown(false)`,
which calls `executor.awaitTermination(24, TimeUnit.HOURS)`
(`HoodieAsyncService.java:114`). The target test is the worst case:
`TestHoodieDeltaStreamerWithMultiWriter` sets
`hoodie.write.lock.wait_time_ms=1200000` with 10 retries, so a lock-blocked
round outlasts any CI budget and a clean 360s failure becomes a job timeout.
Could we run the stop on a throwaway thread with a bounded `get`, falling back
to `dsFuture.cancel(true)`?
##########
hudi-common/src/test/java/org/apache/hudi/common/testutils/JavaTestUtils.java:
##########
@@ -28,7 +28,10 @@ public static boolean checkNestedExceptionContains(Throwable
t, String errorMsg)
Throwable throwable = t;
boolean res = false;
while (throwable != null) {
- if (throwable.getMessage().contains(errorMsg)) {
+ // String.valueOf rather than getMessage().contains: a null message
anywhere in the chain would
+ // otherwise NPE here and lose the failure the caller was trying to
assert on. A TimeoutException
+ // raised before its condition ever threw is one such case, and NPEs in
a chain are another.
+ if (String.valueOf(throwable.getMessage()).contains(errorMsg)) {
Review Comment:
**major:** This fix is load-bearing but nothing exercises it. Pre-PR, when
the ingestion wait timed out, `TestHoodieDeltaStreamerWithMultiWriter.java:451`
walked `ExecutionException -> RuntimeException -> TimeoutException`;
`FutureTask.get(timeout)` throws a null-message `TimeoutException`, so the walk
NPE'd and that NPE replaced the timeout report. The one new call
(`TestWaitTillCondition.java:142`) matches on the outermost message and never
reaches here. Could we add `assertFalse(checkNestedExceptionContains(error, "no
such text"))` to
`timeoutDistinguishesAConditionThatNeverCompletedAnEvaluation`, whose cause is
exactly that null-message `TimeoutException`?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +767,92 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+ /** Bound for {@link #waitFor}; generous, since it only exists to stop a
hung poll running forever. */
+ private static final long WAIT_FOR_TIMEOUT_SECS = 120;
+
+ /**
+ * 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;
+ 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(2000);
+ 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 {
+ res.get(timeoutInSecs, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ Throwable last = lastError.get();
+ int completed = completedEvaluations.get();
+ String detail;
+ if (completed == 0) {
+ // Distinguishes a condition that is stuck part-way through its
first evaluation - a hung Spark
+ // read, say - from one that simply kept returning false.
+ detail = "No evaluation of the condition completed, so it was
still running or never started.";
+ } else if (last == null) {
+ detail = String.format("%d evaluations completed and returned
false without throwing, "
+ + "so there is no further detail.", completed);
+ } else {
+ detail = String.format("%d evaluations completed; the last failure
reported was: %s", completed, last);
+ }
+ Throwable cause = last == null ? e : last;
+ throw new AssertionError(
+ String.format("Condition was not met within %d seconds. %s",
timeoutInSecs, detail), cause);
}
- return ret;
- });
- res.get(timeoutInSecs, TimeUnit.SECONDS);
+ } finally {
+ // stop the polling thread: this method runs once per continuous-mode
test, so a leak accumulates
+ executor.shutdownNow();
+ }
}
/**
* Waits for booleanSupplier to return true
* @param booleanSupplier Boolean supplier
*/
static void waitFor(BooleanSupplier booleanSupplier) {
+ // Bounded, and the interrupt is restored rather than swallowed: this
runs inside conditions passed to
Review Comment:
**major:** This change has no live caller and no test, and I asked for it on
a wrong premise. I said the two tests at `TestHoodieDeltaStreamer.java:2212`
and `:2272` run it, but both have been `@Disabled("HUDI-8951")` since
`1bdba3600a7f` and `995508d3ef24` (Jan 2025), including at the head I reviewed.
Sorry for the churn. Could we add a `waitFor(BooleanSupplier, long
timeoutSecs)` overload so a 1s case can pin the bound, or drop the change from
this PR?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -1744,22 +1744,43 @@ 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(() -> {
+ try {
+ Future 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);
+ }
+ });
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);
+ } catch (Throwable failure) {
+ // Surefire runs this module with forkCount=1 and reuseForks=true, so
a continuous streamer left
+ // running here reads on into the next test, whose setup deletes
basePath and whose teardown closes
+ // the data generators underneath it. Stop it before letting the
failure out.
+ try {
+ ds.shutdownGracefully();
+ } catch (Exception shutdownFailure) {
+ failure.addSuppressed(shutdownFailure);
+ }
+ throw failure;
}
- });
- TestHelpers.waitTillCondition(condition, dsFuture, 360);
- if (cfg != null && !cfg.postWriteTerminationStrategyClass.isEmpty()) {
- awaitDeltaStreamerShutdown(ds);
- } else {
- ds.shutdownGracefully();
- dsFuture.get();
+ 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);
Review Comment:
**minor:** The stop wraps only the wait. If this line fails,
`Assertions.fail("Deltastreamer should have shutdown by now")` at :1795 fires
precisely when the streamer has not shut down, and that AssertionError escapes
through `finally { executor.shutdown(); }`, which does not stop it. That is the
same leak the comment at :1759 describes. Could the stop move into a finally
covering the whole body, skipped when already shut down? Not blocking.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -1744,22 +1744,43 @@ 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(() -> {
+ try {
+ Future 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);
+ }
+ });
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);
+ } catch (Throwable failure) {
+ // Surefire runs this module with forkCount=1 and reuseForks=true, so
a continuous streamer left
+ // running here reads on into the next test, whose setup deletes
basePath and whose teardown closes
+ // the data generators underneath it. Stop it before letting the
failure out.
+ try {
+ ds.shutdownGracefully();
+ } catch (Exception shutdownFailure) {
+ failure.addSuppressed(shutdownFailure);
+ }
+ throw failure;
Review Comment:
**minor:** The catch never joins `dsFuture`, and `shutdownGracefully()` is a
silent no-op when shutdown was already requested: `HoodieAsyncService.java:107`
guards the whole body, `awaitTermination` included, with `if
(!shutdownRequested || force)`, and a post-write termination strategy sets that
flag from inside the loop. So the streamer can still be mid-round when this
rethrows. Could we add a bounded `dsFuture.get(N, SECONDS)` with the result
ignored? Not blocking; it also bounds the point above.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +767,92 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+ /** Bound for {@link #waitFor}; generous, since it only exists to stop a
hung poll running forever. */
+ private static final long WAIT_FOR_TIMEOUT_SECS = 120;
+
+ /**
+ * 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;
+ 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(2000);
+ 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 {
+ res.get(timeoutInSecs, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ Throwable last = lastError.get();
+ int completed = completedEvaluations.get();
+ String detail;
+ if (completed == 0) {
+ // Distinguishes a condition that is stuck part-way through its
first evaluation - a hung Spark
+ // read, say - from one that simply kept returning false.
+ detail = "No evaluation of the condition completed, so it was
still running or never started.";
+ } else if (last == null) {
+ detail = String.format("%d evaluations completed and returned
false without throwing, "
+ + "so there is no further detail.", completed);
+ } else {
+ detail = String.format("%d evaluations completed; the last failure
reported was: %s", completed, last);
+ }
+ Throwable cause = last == null ? e : last;
Review Comment:
**nit:** Two small idiom slips. When `last != null` the `TimeoutException`
is dropped entirely, so the fact that this was a timeout survives only in the
message text. And at :847, `System.nanoTime() > deadline` is the form the
`nanoTime` javadoc warns against, where `t - deadline > 0` is documented as
overflow-safe. Could we suppress the `TimeoutException` onto the AssertionError
and switch that comparison? Feel free to ignore.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +767,92 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+ /** Bound for {@link #waitFor}; generous, since it only exists to stop a
hung poll running forever. */
+ private static final long WAIT_FOR_TIMEOUT_SECS = 120;
+
+ /**
+ * 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;
+ 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(2000);
+ 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 {
+ res.get(timeoutInSecs, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ Throwable last = lastError.get();
+ int completed = completedEvaluations.get();
+ String detail;
+ if (completed == 0) {
+ // Distinguishes a condition that is stuck part-way through its
first evaluation - a hung Spark
+ // read, say - from one that simply kept returning false.
+ detail = "No evaluation of the condition completed, so it was
still running or never started.";
+ } else if (last == null) {
+ detail = String.format("%d evaluations completed and returned
false without throwing, "
Review Comment:
**minor:** This is the branch real callers hit on a timeout, and no test
reaches it. Conditions at `TestHoodieDeltaStreamer.java:2171`, `:2503` and
`:2510` catch, log and `return false`, while the three timing-out cases in
`TestWaitTillCondition` either throw on every poll (`:58`, `:83`) or never
complete one (`:108`). Could we add a case with `ignored -> false` against
`RUNNING` asserting this wording? Not blocking.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.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.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers {@code HoodieDeltaStreamerTestBase.TestHelpers#waitTillCondition},
the helper every
+ * continuous-mode deltastreamer test waits on.
+ *
+ * <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 TestWaitTillCondition {
Review Comment:
**nit:** The class is named for one unit but covers two:
`dyingStreamerWithTerminationStrategyIsSurfacedNotWaitedOut` (:133) exercises
`TestHoodieDeltaStreamer.deltaStreamerTestRunner`, not `waitTillCondition`.
Could we rename to something like `TestDeltaStreamerTestHelpers` so that case
has a home in the name? Feel free to ignore.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +767,92 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+ /** Bound for {@link #waitFor}; generous, since it only exists to stop a
hung poll running forever. */
+ private static final long WAIT_FOR_TIMEOUT_SECS = 120;
+
+ /**
+ * 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;
+ 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(2000);
+ 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 {
+ res.get(timeoutInSecs, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ Throwable last = lastError.get();
Review Comment:
**nit:** The worker sets `lastError` (:805) then increments the counter
(:806), and this reads them in the same order while testing `completed == 0`
first. A timeout landing between the two prints "No evaluation of the condition
completed" when an error was in fact recorded. The cause is still attached, so
it is the message only. Could we test `last != null` first in the chain? Feel
free to ignore.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.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.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers {@code HoodieDeltaStreamerTestBase.TestHelpers#waitTillCondition},
the helper every
+ * continuous-mode deltastreamer test waits on.
+ *
+ * <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 TestWaitTillCondition {
+
+ /** A deltastreamer future that never finishes, as a continuous-mode job
would be. */
+ private static final Future<?> RUNNING = new CompletableFuture<>();
+
+ /**
+ * The helper polls every 2s, so the timeout has to leave room for at least
one evaluation to be recorded.
+ * 5s is enough for that, and keeps the six tests in this class from
spending half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
+
+ @Test
+ void timeoutFailureNamesTheLastConditionFailure() {
+ String assertionText = "assertAtleastNDeltaCommits: expected at least 3
delta commits but got 2";
+
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ throw new AssertionError(assertionText);
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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());
+ assertTrue(error.getMessage().contains("evaluations completed"),
+ () -> "The failure should say how many evaluations completed, which
separates a condition that "
+ + "kept failing from one that never finished an evaluation, but
was: " + error.getMessage());
+ }
+
+ /**
+ * {@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();
+
+ assertThrows(AssertionError.class,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ polls.incrementAndGet();
+ throw new AssertionError("never true");
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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");
+ Thread.sleep(5000);
+ 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,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ try {
+ Thread.sleep(60_000);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return true;
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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());
+ }
+
+ /**
+ * 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 =
"org.apache.hudi.utilities.streamer.NoNewDataTerminationStrategy";
+
+ 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(() ->
HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> true, RUNNING, 30));
Review Comment:
**nit:** Two bare `30` literals remain, here and at :162, after the earlier
round named the `5`. Could we name this one too, say
`NEVER_REACHED_TIMEOUT_SECS = 30`, so both never-timing-out cases read the same
way? Feel free to ignore.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.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.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers {@code HoodieDeltaStreamerTestBase.TestHelpers#waitTillCondition},
the helper every
+ * continuous-mode deltastreamer test waits on.
+ *
+ * <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 TestWaitTillCondition {
+
+ /** A deltastreamer future that never finishes, as a continuous-mode job
would be. */
+ private static final Future<?> RUNNING = new CompletableFuture<>();
+
+ /**
+ * The helper polls every 2s, so the timeout has to leave room for at least
one evaluation to be recorded.
+ * 5s is enough for that, and keeps the six tests in this class from
spending half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
+
+ @Test
+ void timeoutFailureNamesTheLastConditionFailure() {
+ String assertionText = "assertAtleastNDeltaCommits: expected at least 3
delta commits but got 2";
+
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ throw new AssertionError(assertionText);
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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());
+ assertTrue(error.getMessage().contains("evaluations completed"),
+ () -> "The failure should say how many evaluations completed, which
separates a condition that "
+ + "kept failing from one that never finished an evaluation, but
was: " + error.getMessage());
+ }
+
+ /**
+ * {@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();
+
+ assertThrows(AssertionError.class,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ polls.incrementAndGet();
+ throw new AssertionError("never true");
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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");
+ Thread.sleep(5000);
+ 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,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ try {
+ Thread.sleep(60_000);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return true;
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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());
+ }
+
+ /**
+ * 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 =
"org.apache.hudi.utilities.streamer.NoNewDataTerminationStrategy";
Review Comment:
**nit:** This FQN is never resolved: `deltaStreamerTestRunner` only checks
`!cfg.postWriteTerminationStrategyClass.isEmpty()`
(`TestHoodieDeltaStreamer.java:1769`), so any non-empty string passes and a
rename would not break the test. Two classes carry this simple name and the one
in `deltastreamer` is `@Deprecated`. Could we use
`NoNewDataTerminationStrategy.class.getName()`, matching
`TestHoodieDeltaStreamer.java:1706`? Feel free to ignore.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.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.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers {@code HoodieDeltaStreamerTestBase.TestHelpers#waitTillCondition},
the helper every
+ * continuous-mode deltastreamer test waits on.
+ *
+ * <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 TestWaitTillCondition {
+
+ /** A deltastreamer future that never finishes, as a continuous-mode job
would be. */
+ private static final Future<?> RUNNING = new CompletableFuture<>();
+
+ /**
+ * The helper polls every 2s, so the timeout has to leave room for at least
one evaluation to be recorded.
+ * 5s is enough for that, and keeps the six tests in this class from
spending half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
+
+ @Test
+ void timeoutFailureNamesTheLastConditionFailure() {
+ String assertionText = "assertAtleastNDeltaCommits: expected at least 3
delta commits but got 2";
+
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ throw new AssertionError(assertionText);
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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());
+ assertTrue(error.getMessage().contains("evaluations completed"),
+ () -> "The failure should say how many evaluations completed, which
separates a condition that "
+ + "kept failing from one that never finished an evaluation, but
was: " + error.getMessage());
+ }
+
+ /**
+ * {@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();
+
+ assertThrows(AssertionError.class,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
+ ignored -> {
+ polls.incrementAndGet();
+ throw new AssertionError("never true");
+ }, RUNNING, CONDITION_TIMEOUT_SECS));
+
+ 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");
+ Thread.sleep(5000);
Review Comment:
**nit:** This fixed sleep is the slowest thing in the class, 10.0s of its
25.3s in the surefire report. The value is really "two poll periods", and the
period is a bare `Thread.sleep(2000)` at
`HoodieDeltaStreamerTestBase.java:791`. Could we extract that as a
package-private `POLL_INTERVAL_MS` and sleep `2 * POLL_INTERVAL_MS` here, so it
drops to about 4s and the tie is explicit? Feel free to ignore.
--
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]
