voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956042053
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -1744,22 +1748,71 @@ 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 {
- 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);
+ 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();
}
- });
- TestHelpers.waitTillCondition(condition, dsFuture, 360);
- if (cfg != null && !cfg.postWriteTerminationStrategyClass.isEmpty()) {
- awaitDeltaStreamerShutdown(ds);
- } else {
- ds.shutdownGracefully();
- dsFuture.get();
+ stoppedCleanly = true;
+ } finally {
+ if (!stoppedCleanly) {
+ stopLeakedStreamer(ds, dsFuture);
+ }
+ 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 {
+ stopper.submit(ds::shutdownGracefully).get(STREAMER_STOP_TIMEOUT_SECS,
TimeUnit.SECONDS);
+ 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);
+ if (dsFuture != null) {
+ dsFuture.cancel(true);
Review Comment:
**minor:** Not blocking. The 60s bound stops the caller waiting, not the
stopper thread. `stopper.shutdownNow()` interrupts `awaitTermination`, but
`HoodieAsyncService.shutdown` swallows that (`HoodieAsyncService.java:118-120`)
and `HoodieStreamer.java:214-220` runs `ds.close()` regardless, so the write
client can close under a still-running ingest round. Would a guarded
`ds.getIngestionService().shutdown(true)` before this cancel be worth it? It is
`Option.get()`, so it needs the guard.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -1744,22 +1748,71 @@ 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 {
- 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);
+ 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();
}
- });
- TestHelpers.waitTillCondition(condition, dsFuture, 360);
- if (cfg != null && !cfg.postWriteTerminationStrategyClass.isEmpty()) {
- awaitDeltaStreamerShutdown(ds);
- } else {
- ds.shutdownGracefully();
- dsFuture.get();
+ stoppedCleanly = true;
+ } finally {
+ if (!stoppedCleanly) {
+ stopLeakedStreamer(ds, dsFuture);
+ }
+ 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();
Review Comment:
**nit:** Feel free to ignore. This sits outside the try, so if the executor
cannot be created the throw escapes `stopLeakedStreamer` and replaces the
caller's diagnostic, which the comment at :1804 says must not happen, and it
also skips `executor.shutdown()` at :1780. Could the `stopLeakedStreamer(...)`
call at :1778 be wrapped in its own try/finally?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -1744,22 +1748,71 @@ 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 {
- 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);
+ 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();
}
- });
- TestHelpers.waitTillCondition(condition, dsFuture, 360);
- if (cfg != null && !cfg.postWriteTerminationStrategyClass.isEmpty()) {
- awaitDeltaStreamerShutdown(ds);
- } else {
- ds.shutdownGracefully();
- dsFuture.get();
+ stoppedCleanly = true;
+ } finally {
+ if (!stoppedCleanly) {
+ stopLeakedStreamer(ds, dsFuture);
+ }
+ 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 {
+ stopper.submit(ds::shutdownGracefully).get(STREAMER_STOP_TIMEOUT_SECS,
TimeUnit.SECONDS);
+ if (dsFuture != null) {
+ dsFuture.get(STREAMER_STOP_TIMEOUT_SECS, TimeUnit.SECONDS);
+ }
Review Comment:
**major:** `catch (ExecutionException)` at :1800 catches from both futures.
If `shutdownGracefully` itself throws, `dsFuture` is neither joined (:1798) nor
cancelled (:1811) and nothing is logged, which is the leak this method exists
to close. The likely trigger is :1772 throwing, which is why `stoppedCleanly`
stayed false, and this then retries the same call. Could we scope the tolerance
to the ingest future?
```suggestion
Future<?> stop = stopper.submit(ds::shutdownGracefully);
try {
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.
log.warn("Stopping the streamer threw after a failure", stopThrew);
}
if (dsFuture != null) {
dsFuture.get(STREAMER_STOP_TIMEOUT_SECS, TimeUnit.SECONDS);
}
```
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +776,110 @@ 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;
+ 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(POLL_INTERVAL_MS);
+ 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();
+ Throwable cause = last == null ? e : last;
+ AssertionError failure = new AssertionError(
+ describeTimeout(last, completedEvaluations.get(),
timeoutInSecs), cause);
Review Comment:
**nit:** The worker writes `lastError` (:811) then the counter (:812), and
this reads them in the opposite order. An evaluation throwing between the two
reads yields `(null, 1)`, so the report claims "1 evaluations completed and
returned false without throwing" for a condition that did throw. Reading the
counter first puts any skew in the branch :845 already justifies.
```suggestion
int completed = completedEvaluations.get();
Throwable last = lastError.get();
Throwable cause = last == null ? e : last;
AssertionError failure = new AssertionError(
describeTimeout(last, completed, timeoutInSecs), cause);
```
##########
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:
**minor:** Not blocking. This helper still has no test in its own module
(`find . -name 'TestJavaTestUtils*'` returns nothing); its only coverage is an
`assertFalse` over in hudi-utilities that can only fail by NPE-ing. The
null-message *head* case is live at `TestHoodieDeltaStreamer.java:2316` and
uncovered. Could we add a small `TestJavaTestUtils` in hudi-common for null
head, null link mid-chain, and `errorMsg = "null"` not matching?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.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.assertFalse;
+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 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 tests in this class from spending
half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
+
+ /** 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,
+ () -> 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(2 * HoodieDeltaStreamerTestBase.TestHelpers.POLL_INTERVAL_MS);
Review Comment:
**minor:** This fixed sleep makes the test the most expensive in the class,
9.0s of its 30.2s, to prove something a `join` proves in about 0ms and more
strongly: `executor.shutdownNow()` runs in the finally before the
`AssertionError` propagates, so the worker is already stopping when this
returns. Could the condition capture `Thread.currentThread()` into an
`AtomicReference`, so this becomes `poller.join(5000);
assertFalse(poller.isAlive(), ...)` with the `assertEquals` kept?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.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.assertFalse;
+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 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 tests in this class from spending
half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
+
+ /** 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,
+ () -> HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition(
Review Comment:
**nit:** `HoodieDeltaStreamerTestBase.TestHelpers.` is spelled out at seven
call sites in this class. Same-package static import is already the idiom here:
`TestHoodieDeltaStreamerWithMultiWriter.java:69` static-imports
`deltaStreamerTestRunner`. Could we static-import `waitTillCondition` and
`waitFor` and drop the qualifier?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.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.assertFalse;
+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 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 tests in this class from spending
half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
+
+ /** 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,
+ () -> 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());
+ }
Review Comment:
**nit:** The `addSuppressed(e)` at `HoodieDeltaStreamerTestBase.java:828`
keeps the fact that this was a timeout once the condition's own error becomes
the cause, and nothing asserts it. Would
`assertInstanceOf(TimeoutException.class, error.getSuppressed()[0])` be worth
adding here? It needs two new imports, so it is not a one-click suggestion.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +776,110 @@ 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;
+ 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()) {
Review Comment:
**major:** The `!executor.isShutdown()` guard is reached by none of the
eight new tests, so the hazard it was added for is unpinned.
`pollingStopsOnceTheWaitHasGivenUp` exits through the `InterruptedException`
break at :808, and
`timeoutDistinguishesAConditionThatNeverCompletedAnEvaluation` restores the
flag and returns true, exiting on `!ret`. Could we add a case whose condition
sleeps past the timeout, swallows the interrupt without restoring the flag and
returns false, asserting the poll count then freezes?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +776,110 @@ 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;
+ 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(POLL_INTERVAL_MS);
+ 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();
+ Throwable cause = last == null ? e : last;
+ AssertionError failure = new AssertionError(
+ describeTimeout(last, completedEvaluations.get(),
timeoutInSecs), cause);
+ if (cause != e) {
+ // The condition's own error is the more useful cause, but the
fact that this was a timeout is
+ // still part of the diagnosis, so it is carried along rather than
dropped.
+ failure.addSuppressed(e);
}
+ throw failure;
}
- 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();
+ }
+ }
+
+ /**
+ * Builds the timeout report. Which of the three shapes it takes is the
whole diagnostic: an error the
+ * condition threw, a condition that never completed an evaluation, or one
that kept returning false.
+ */
+ static String describeTimeout(Throwable last, int completed, long
timeoutInSecs) {
Review Comment:
**nit:** This was extracted to be directly testable, and the one branch
combination `waitTillCondition` cannot produce is the one its own comment
justifies: `last != null && completed == 0`. Would a one-line
`assertTrue(describeTimeout(new RuntimeException("boom"), 0,
5).contains("boom"))` be worth adding, to pin the ordering :845 argues for?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##########
@@ -763,34 +776,110 @@ 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;
+ 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(POLL_INTERVAL_MS);
+ 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);
Review Comment:
**minor:** The loop has two exits and only one of them reports. When it ends
because `dsFuture.isDone()`, the `Boolean` from `res.get` is discarded and
`lastError` is dropped, so "the streamer finished before the condition ever
held" reads exactly like success. Could we log `describeTimeout(...)` at warn
when `res.get` returns false, so both exits carry the same diagnostic?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.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.assertFalse;
+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 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 tests in this class from spending
half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
Review Comment:
**minor:** Not blocking. Four of the eight tests pay a 5s timeout only
because `POLL_INTERVAL_MS` is fixed at 2000, so roughly 20s of the class's
30.2s is spent waiting out the helper's poll cadence rather than pinning
anything. Would a package-private `waitTillCondition` overload taking a poll
interval be worth it, so these four run at 50ms polls and the class drops to
about a second?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.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.assertFalse;
+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 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 tests in this class from spending
half a minute asleep in the
+ * shared utilities job.
+ */
+ private static final int CONDITION_TIMEOUT_SECS = 5;
+
+ /** 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,
+ () -> 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());
Review Comment:
**nit:** Feel free to ignore. This does not discriminate: the "returned
false" branch message also contains "evaluations completed"
(`HoodieDeltaStreamerTestBase.java:846`), and the branch is already pinned
exactly by the `contains(assertionText)` check above it. Could it assert the
other branch is absent instead?
```suggestion
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());
```
--
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]
