Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957508101
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java:
##
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.utilities.deltastreamer;
+
+import org.apache.hudi.common.testutils.JavaTestUtils;
+import org.apache.hudi.utilities.streamer.NoNewDataTerminationStrategy;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static
org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerTestBase.TestHelpers.describeTimeout;
+import static
org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerTestBase.TestHelpers.waitFor;
+import static
org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers the deltastreamer test helpers every continuous-mode test runs on:
the wait in
+ * {@code HoodieDeltaStreamerTestBase.TestHelpers} and the runner in {@code
TestHoodieDeltaStreamer}.
+ *
+ * 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}
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5585922638 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) * aca668b073e0e8137266f98656f0383d6262bf01 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16905) * d74dd15494635cd7a91df7c3f96922800c3912fb Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16909) * a9c354a5c8d7ff2b24d3daa3ff94b84ff09267e2 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956273259
##
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 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.
+ *
+ * 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:
Fixed in `9f3cd75`. The stop future has its own `catch (ExecutionException
stopThrew)` now, which logs and falls through, so the ingest future is still
joined and cancelled when `shutdownGracefully` throws. The outer clause keeps
tolerating only the ingest task's own failure.
##
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 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
+
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3955745126
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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) {
Review Comment:
Done in `4555b92`: the three branches moved into `describeTimeout(last,
completed, timeoutInSecs)`, logic and strings unchanged. Left it without a
direct unit test, since all three branches are already driven end to end by
`timeoutFailureNamesTheLastConditionFailure`,
`timeoutDistinguishesAConditionThatNeverCompletedAnEvaluation` and
`timeoutReportsEvaluationsThatReturnedFalse`, so a direct test would only
duplicate them.
##
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;
Review Comment:
Done in `4555b92`. Both `WAIT_FOR_TIMEOUT_SECS` and `POLL_INTERVAL_MS` now
sit at the top of `TestHelpers` with the other statics.
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5586005896 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) * aca668b073e0e8137266f98656f0383d6262bf01 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16905) * d74dd15494635cd7a91df7c3f96922800c3912fb Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16909) * a9c354a5c8d7ff2b24d3daa3ff94b84ff09267e2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16912) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous merged PR #19485: URL: https://github.com/apache/hudi/pull/19485 -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957501394
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##
@@ -1744,22 +1749,100 @@ static void
deltaStreamerTestRunner(HoodieDeltaStreamer ds, HoodieDeltaStreamer.
static void deltaStreamerTestRunner(HoodieDeltaStreamer ds,
HoodieDeltaStreamer.Config cfg, Function 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.
+ *
+ * Surefire runs this module with forkCount=1 and reuseForks=true, so a live
streamer reads on into the
+ * next test, whose setup deletes basePath and whose teardown closes the
data generators underneath it.
+ * The stop has to be bounded: shutdownGracefully awaits the ingest executor
for up to 24 hours, and it
+ * returns immediately without waiting when shutdown was already requested,
so neither the wait nor the
+ * absence of one can be relied on here.
+ */
+ private static void stopLeakedStreamer(HoodieDeltaStreamer ds, Future
dsFuture) {
+ExecutorService stopper = Executors.newSingleThreadExecutor();
+try {
+ Future stop = stopper.submit(ds::shutdownGracefully);
try {
-ds.sync();
- } catch (Exception ex) {
-log.warn("DS continuous job failed, hence not proceeding with
condition check for {}", jobId);
-throw new RuntimeException(ex.getMessage(), ex);
+stop.get(STREAMER_STOP_TIMEOUT_SECS, TimeUnit.SECONDS);
+ } catch (ExecutionException stopThrew) {
Review Comment:
Fixed in `d74dd15`. `stopLeakedStreamer` now takes the bound as a parameter,
the same seam `waitTillCondition` has for its poll interval, and two cases
drive it at 1s: `stopThatThrowsStillCancelsTheIngestTask` and
`stopThatHangsIsBoundedAndCancelsTheIngestTask`, both asserting the ingest task
ends up cancelled. Confirmed they discriminate: removing the stop-threw
fall-through fails the first and leaves the second passing.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java:
##
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.utilities.deltastreamer;
+
+import org.apache.hudi.common.testutils.JavaTestUtils;
+import org.apache.hudi.utilities.streamer.NoNewDataTerminationStrategy;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5584778634 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) * aca668b073e0e8137266f98656f0383d6262bf01 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16905) * d74dd15494635cd7a91df7c3f96922800c3912fb UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5587306137 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * d74dd15494635cd7a91df7c3f96922800c3912fb Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16909) * a9c354a5c8d7ff2b24d3daa3ff94b84ff09267e2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16912) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5587772323 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * a9c354a5c8d7ff2b24d3daa3ff94b84ff09267e2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16912) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5584425442 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) * aca668b073e0e8137266f98656f0383d6262bf01 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16905) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5586091147 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * aca668b073e0e8137266f98656f0383d6262bf01 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16905) * d74dd15494635cd7a91df7c3f96922800c3912fb Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16909) * a9c354a5c8d7ff2b24d3daa3ff94b84ff09267e2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16912) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on PR #19485:
URL: https://github.com/apache/hudi/pull/19485#issuecomment-5585869195
One more pass over `d74dd15`, this time on the two commits that had landed
after the last full read. Everything it found is fixed in `093458a` (the review
items) and `a9c354a` (a config mirror on two sibling tests, kept separate so it
can be dropped), and the description is updated to match. Nothing here is a
wrong result from the helpers: the first item was a flake in the new test
class, the rest are coverage and prose. Nine of the eleven commits are now
mine, so this needs an independent committer's approval rather than mine.
The 15 items and what changed
| | where | finding | fix |
|---|---|---|---|
| major | `TestDeltaStreamerTestHelpers.java:224` | `contains("0 evaluations
completed")` also matches 10/20/30 (18 measured locally), so the assertion
flaked on any multiple of ten | parses the count and checks it against the
condition's own tally, allowing one in-flight evaluation |
| major | `:366` | `stopThatHangsIsBounded...` never observed the bound:
widening `stop.get` to 20s left it green, 20s slower | the stop cases run under
a 5s `assertTimeoutPreemptively` ceiling (two bounds plus slack) |
| major | `TestHoodieDeltaStreamer.java:1863`, `:1850` |
`ingestionService.shutdown(true)` and the close-wait timeout were reached by no
test, since the mock's `getIngestionService()` is null | new
`stopThatOutlivesItsBoundIsForceStoppedAndLetRunOn`: stubbed service,
`verify(shutdown(true))`, lower-bound timing on the close-wait |
| minor | `HoodieDeltaStreamerTestBase.java:557,579,723,731` |
`assertRecordCount` and `assertDistanceCount` produced identical text and every
commit helper a bare `Got=N, exp >=M`, so the one-line report could not name
the assertion | every helper's message starts with its own name |
| minor | description | the sample output was the test fixture string; the
code formats `org.opentest4j.AssertionFailedError: assertAtleastNDeltaCommits:
Got=2, exp >=3 ==> expected: but was: ` | description quotes what
the code formats |
| minor | `:1803` + description | "three times" the bound is unreachable: a
stop that times out skips the join and a stop that returns leaves nothing for
the close-wait, so at most two run in sequence | javadoc and description say
twice |
| minor | `:225` | two 60s bounds on top of the 360s wait left 120s of the
`@Timeout(600)` budget; once that blows, JUnit's `SameThreadTimeoutInvocation`
replaces the failure and demotes the diagnostic to a suppressed exception |
bound is 30s |
| minor | `TestDeltaStreamerTestHelpers.java:122`, `:142`, `:179` |
`pollingStopsOnce...` killed no mutant the other tests miss, and the `polls`
equality in both polling tests had a narrow race | the test is dropped; the
equality is dropped, `isAlive()` after `join` stays |
| minor | `:311`, `:330` | the `polling.await` result was discarded, so a
slow first poll would be reported as a broken interrupt path; the elapsed
assertion could not be what fails | latch result asserted first; elapsed check
dropped |
| minor | `TestHoodieDeltaStreamer.java:1832`, `:1850` | a caller arriving
interrupted made `awaitTermination` throw at once and skip the close-wait |
flag cleared for the wait and restored; new
`interruptedCallerStillGetsTheBoundedCloseWait` |
| minor | `HoodieDeltaStreamerTestBase.java:833` | the "future finished"
warn is false on the interrupt exit, which the direct-interrupt test drives |
branches on `dsFuture.isDone()` |
| nit | `TestJavaTestUtils.java:44` | `testNullThrowableReturnsFalse` had no
unique kill | dropped |
| nit | `TestDeltaStreamerTestHelpers.java:339-387` | the two stop tests
were 20-line twins | shared helper |
| minor | `TestHoodieDeltaStreamerWithMultiWriter.java:186`, `:263` |
`WithoutConflicts` and `testLatestCheckpointCarryOver...` wait on the same MOR
prep condition as `ForConflicts` but never got the compaction settings
f5d03a61749f (#9072) added there | `a9c354a`, separate commit |
Gates on JDK 11: checkstyle clean on both modules, `TestJavaTestUtils` 4/4,
`TestDeltaStreamerTestHelpers` 14/14 three times (~14.5s). Each new assertion
was checked against a mutation that flips only it.
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5585153974 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) * aca668b073e0e8137266f98656f0383d6262bf01 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16905) * d74dd15494635cd7a91df7c3f96922800c3912fb Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16909) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957506061
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 condition, Future
dsFuture, long timeoutInSecs,
+ long pollIntervalMs) throws Exception {
+ AtomicReference lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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:
Fixed in `d74dd15`. Both values are hoisted into locals in the counter-first
order, matching the timeout path, with a note that the worker has already
finished here so neither read can be stale.
##
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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957503614
##
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}.
+ *
+ * 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}
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5584293291 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 4555b92e5eada7f19f34644c52b34136957bd8e2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16897) * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) * aca668b073e0e8137266f98656f0383d6262bf01 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957276515
##
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}.
+ *
+ * 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}
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957278552
##
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 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.
+ *
+ * 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:
Done in `aca668b`, softened to "forcing the executor down at least
interrupts the ingest
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957273057
##
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}.
+ *
+ * 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}
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-agent commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3957259470
##
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 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.
+ *
+ * Surefire runs this module with forkCount=1 and reuseForks=true, so a live
streamer reads on into the
+ * next test, whose setup deletes basePath and whose teardown closes the
data generators underneath it.
+ * The stop has to be bounded: shutdownGracefully awaits the ingest executor
for up to 24 hours, and it
+ * returns immediately without waiting when shutdown was already requested,
so neither the wait nor the
+ * absence of one can be relied on here.
+ */
+ private static void stopLeakedStreamer(HoodieDeltaStreamer ds, Future
dsFuture) {
+ExecutorService stopper = Executors.newSingleThreadExecutor();
+try {
+ Future stop = stopper.submit(ds::shutdownGracefully);
try {
-ds.sync();
- } catch (Exception ex) {
-log.warn("DS continuous job failed, hence not proceeding with
condition check for {}", jobId);
-throw new RuntimeException(ex.getMessage(), ex);
+stop.get(STREAMER_STOP_TIMEOUT_SECS, TimeUnit.SECONDS);
+ } catch (ExecutionException stopThrew) {
+// The stop itself failing does not excuse leaving the ingest task
running, so fall through to the join
+// below rather than take the outer clause, which tolerates only the
ingest task's own failure.
+log.warn("Stopping the streamer threw after a failure", stopThrew);
}
-});
-TestHelpers.waitTillCondition(condition, dsFuture, 360);
-if (cfg != null && !cfg.postWriteTerminationStrategyClass.isEmpty()) {
- awaitDeltaStreamerShutdown(ds);
-} else {
- ds.shutdownGracefully();
- dsFuture.get();
+ if (dsFuture != null) {
+dsFuture.get(STREAMER_STOP_TIMEOUT_SECS, TimeUnit.SECONDS);
+ }
+} catch (ExecutionException ingestFailure) {
+ // Expected rather than anomalous: the ingest task failing is usually
why the caller is unwinding at
+ // all, and the caller reports it. Nothing to warn about here.
+} catch (Exception stopFailure) {
+ // Swallowed on purpose: this runs while another failure is propagating,
and replacing that failure
+ // with this one would hide the diagnostic the caller is about to report.
+ if (stopFailure instanceof InterruptedException) {
+Thread.currentThread().interrupt();
+ }
+ log.warn("Could not stop the streamer cleanly after a failure,
cancelling the ingest task", stopFailure);
+ // The 60s bound only stops this thread waiting:
HoodieAsyncService.shutdown(false) swallows the interrupt
+ // that stopper.shutdownNow() sends, and
HoodieStreamer.shutdownGracefully runs ds.close() regardless, so
+ // without forcing the executor down the write client can close under a
still-running ingest round.
+ forceStopIngestion(ds);
+ if (dsFuture != null) {
+dsFuture.cancel(true);
+ }
+
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5584079540 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 4555b92e5eada7f19f34644c52b34136957bd8e2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16897) * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
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 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.
+ *
+ * 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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5581776700 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 8e9e54f793583fc46e0bc96fab454793ca65220b Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16884) * ae6065c97378310c06e2fd4fecee506fb648a757 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16894) * 4555b92e5eada7f19f34644c52b34136957bd8e2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16897) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956285803
##
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}.
+ *
+ * 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:
Fixed in `9f3cd75`, with `assertInstanceOf(TimeoutException.class,
error.getSuppressed()[0], ...)` and the two imports it needs.
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956281146
##
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}.
+ *
+ * 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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956283413
##
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}.
+ *
+ * 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:
Fixed in `9f3cd75`, it asserts the "returned false without throwing" text is
absent instead, which does discriminate.
##
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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
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 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.
+ *
+ * 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 condition, String
jobId) throws Exception {
ExecutorService executor = Executors.newSingleThreadExecutor();
-Future dsFuture = executor.submit(() -> {
-
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5582674237 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * ae6065c97378310c06e2fd4fecee506fb648a757 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16894) * 4555b92e5eada7f19f34644c52b34136957bd8e2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16897) * 9f3cd752beaf80ae095dcb0525051a184286203f Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16899) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956277289
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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:
Fixed in `9f3cd75`. The result of `res.get` is captured, and the
`dsFuture.isDone()` exit logs the same detail at warn through a new
`describeProgress`, which `describeTimeout` also calls, so all three existing
messages are byte-for-byte unchanged.
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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() &&
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5582573187 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * ae6065c97378310c06e2fd4fecee506fb648a757 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16894) * 4555b92e5eada7f19f34644c52b34136957bd8e2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16897) * 9f3cd752beaf80ae095dcb0525051a184286203f UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956279167
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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:
Fixed in `9f3cd75`.
`describeTimeoutReportsAnErrorEvenWithNoCompletedEvaluation` calls it directly
with `(new AssertionError("boom"), 0, ...)` and asserts the error is reported
and the "No evaluation of the condition completed" text is not.
##
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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3956275302
##
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 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.
+ *
+ * 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:
Fixed in `9f3cd75`. The `stopLeakedStreamer` call in the runner's finally is
wrapped in its own try/catch, so a throw out of it cannot replace the caller's
diagnostic and `executor.shutdown()` always runs.
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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() &&
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5582275735 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * ae6065c97378310c06e2fd4fecee506fb648a757 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16894) * 4555b92e5eada7f19f34644c52b34136957bd8e2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16897) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5581704372 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 8e9e54f793583fc46e0bc96fab454793ca65220b Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16884) * ae6065c97378310c06e2fd4fecee506fb648a757 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16894) * 4555b92e5eada7f19f34644c52b34136957bd8e2 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3955536813
##
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 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:
Agreed, and fixed in `ae6065c` (this was raised against `8e9e54f`). The stop
now runs on its own thread with a 60s bound and cancels the ingest task if it
does not return, so the diagnostic `AssertionError` is thrown either way rather
than waiting on the 24h `awaitTermination`.
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3955409519
##
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.
+ *
+ * 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:
Fixed in `ae6065c`. `POLL_INTERVAL_MS` is now a constant on `TestHelpers`
and this sleeps `2 * POLL_INTERVAL_MS`, so the tie to the poll period is
explicit.
##
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3955408360
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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:
Fixed in `ae6065c`. The `TimeoutException` is attached with `addSuppressed`
when it is not the cause, and the deadline check is now `System.nanoTime() -
deadline > 0`.
##
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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5581101234 Pushed fixes for all twelve as `ae6065c`, @rangareddy, rather than send you round-tripping on our own review. Happy to revert any of them; the class rename is the most opinion-dependent, so say the word if you would rather keep the old name. Gates on JDK 11: `TestDeltaStreamerTestHelpers` 8/8 (30.2s), and `TestHoodieDeltaStreamer#testHoodieAsyncClusteringJob*` 5/5 (83s). The second one because the runner restructure sits under every continuous-mode test, so the mock alone was not enough to trust it. Description refreshed for the rename, the test count and the 5s timeout. The two open items from before are still yours: `awaitCondition` in the multi-writer test, and the issue links. -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3955405661
##
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 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:
Fixed in `ae6065c`. The stop now runs on its own thread with a 60s bound and
cancels the ingest task if it does not come back, so a wedged streamer can no
longer outlive the failure it is reporting.
##
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:
Fixed in `ae6065c`.
`timeoutDistinguishesAConditionThatNeverCompletedAnEvaluation` now also asserts
`assertFalse(checkNestedExceptionContains(error, "no such text"))`, which walks
past the null-message `TimeoutException` this path attaches and NPEs without
the `String.valueOf`.
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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();
+
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3955406860
##
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 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:
Fixed in `ae6065c`. `stopLeakedStreamer` joins the ingest future with the
same 60s bound after the stop, so the already-requested no-op case is covered
too.
##
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 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:
Fixed in `ae6065c`. The stop moved into a `finally` covering the whole body,
guarded by a `stoppedCleanly` flag, so a failure in
`awaitDeltaStreamerShutdown` stops the streamer as well.
##
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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5580949770 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 8e9e54f793583fc46e0bc96fab454793ca65220b Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16884) * ae6065c97378310c06e2fd4fecee506fb648a757 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16894) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5580728901 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 8e9e54f793583fc46e0bc96fab454793ca65220b Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16884) * ae6065c97378310c06e2fd4fecee506fb648a757 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
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 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=120` 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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-agent commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3953944591
##
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 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:
🤖 `shutdownGracefully()` goes through `HoodieAsyncService.shutdown(false)`,
which does `executor.awaitTermination(24, HOURS)` — so if the reason the
condition timed out is that the streamer is stuck mid-`ingestOnce()`, this call
blocks and the new diagnostic `AssertionError` is never thrown before CI kills
the job. Would it be worth logging `failure` (at error level) before calling
shutdown, so the report this PR adds survives that case?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
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;
Review Comment:
🤖 nit: would you mind moving `WAIT_FOR_TIMEOUT_SECS` to the top of the
TestHelpers class (with the other statics) rather than sandwiched between two
methods? It's only used by `waitFor` further down, so it's a little surprising
to find it above `waitTillCondition`.
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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);
+
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5575561257 ## CI report: * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 8e9e54f793583fc46e0bc96fab454793ca65220b Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16884) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5574686336 ## CI report: * 6b07137bad325df6396f951ac2a67b5a423f352a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16807) * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5574774546 ## CI report: * 6b07137bad325df6396f951ac2a67b5a423f352a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16807) * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 8e9e54f793583fc46e0bc96fab454793ca65220b Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16884) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5574731654 ## CI report: * 6b07137bad325df6396f951ac2a67b5a423f352a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16807) * a36cc9306a4d417fcf2704d27e4257221557ce63 UNKNOWN * 8e9e54f793583fc46e0bc96fab454793ca65220b UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3952104110
##
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.
+ *
+ * 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 condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ AtomicInteger completedEvaluations = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future 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 {
+// this used to leak the polling thread on every call, and it is
called by every continuous-mode test
+executor.shutdownNow();
Review Comment:
Done in `8e9e54f`, reworded to say what the call does and why it matters:
"stop the polling thread: this method runs once per continuous-mode test, so a
leak accumulates".
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3952103232
##
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.
+ *
+ * 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 the same margin {@link #pollingStopsOnceTheWaitHasGivenUp} already
relies on, and keeps the four
+ * 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, 5));
Review Comment:
Done in `8e9e54f`. Also reworded the constant's javadoc while there: it
cited this test as the source of the margin, which is circular once the test
reads the constant, and it still said four tests when the class now has six.
--
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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-agent commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3952060435
##
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.
+ *
+ * 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 the same margin {@link #pollingStopsOnceTheWaitHasGivenUp} already
relies on, and keeps the four
+ * 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, 5));
Review Comment:
🤖 nit: could this use `CONDITION_TIMEOUT_SECS` instead of the literal `5`?
The constant's javadoc says this test relies on the same margin, so the two are
meant to move together.
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,34 +767,92 @@ static HoodieInstant
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3952029195
##
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;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * 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.
+ */
+/** 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;
Review Comment:
Pushed as `a36cc93`: the constant and its own doc comment now sit above the
block, so the `Polls ...` javadoc pairs with `waitTillCondition` again. Sorry
for going straight to a commit on your branch, @rangareddy, it seemed less
churn than another round trip for a three-line move. Happy to revert if you
would rather own it.
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3952031544
##
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;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * 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.
+ */
Review Comment:
Addressed in `a36cc93`: the constant and its doc comment moved above the
`waitTillCondition` javadoc. Worth noting the effect went past ordering, since
javadoc keeps only the comment immediately before a declaration, the method had
ended up with no doc at all.
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3951967035
##
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;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * 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.
+ */
+/** 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;
Review Comment:
**nit:** Two Javadoc blocks now sit in a row, and javadoc keeps only the one
immediately before a declaration, so the `Polls {@code condition} ...` block is
dropped and `waitTillCondition` ends up with no doc at all. Could the constant
move above that block, or below the method?
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3951967035
##
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;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * 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.
+ */
+/** 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;
Review Comment:
**nit:** same spot as the bot's note, though the effect is a bit more than
ordering. Two Javadoc blocks now sit in a row, and javadoc keeps only the one
immediately before a declaration, so the `Polls {@code condition} ...` block is
dropped and `waitTillCondition` ends up with no doc at all. Could the constant
move above that block, or below the method?
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5574423798 Two nits from my review body did not get a reply. Both are still open, and both are fine to decline: - `TestHoodieDeltaStreamerWithMultiWriter.awaitCondition` (line 486) still only logs on expiry, so the backfill can start early and surface later as an unrelated conflict. Sibling helper, so arguably out of scope here. Could it fail with a message instead, or would you rather keep it separate? - The description cites #16228 but not #15686 (HUDI-5550), which is the same flaky test. Could we link it, and cite `f9a597bd783c` (HUDI-6842, #9671) as the precedent where knowing which assertion failed was the fix? Everything else from that review looks addressed, verified against the code at `6b07137` rather than the replies. -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-agent commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3927532311
##
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;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * 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.
+ */
Review Comment:
🤖 nit: the constant declaration ends up sandwiched between the Javadoc for
`waitTillCondition` and the method itself — a reader scanning top-to-bottom
sees the Javadoc, then a constant, and has to mentally re-attach the doc to the
method below. Could you move `WAIT_FOR_TIMEOUT_SECS` (and its own `/** … */`)
to just *before* the `waitTillCondition` Javadoc so the doc–method pairing is
unbroken?
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
--
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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5527662791 ## CI report: * 6b07137bad325df6396f951ac2a67b5a423f352a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16807) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5526408746 ## CI report: * 1d01b6279e4c2942d08580bc054e51412f89a71a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16232) * 6b07137bad325df6396f951ac2a67b5a423f352a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16807) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5526320148 ## CI report: * 1d01b6279e4c2942d08580bc054e51412f89a71a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16232) * 6b07137bad325df6396f951ac2a67b5a423f352a UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3924855907
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,55 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
+try {
+ Thread.sleep(2000);
+ ret = condition.apply(true);
+ if (ret) {
+log.info("Condition completed successfully");
+ }
+} catch (InterruptedException interrupted) {
+ // shutdownNow below interrupts this thread once the wait has
given up. Thread.sleep clears
+ // the interrupt flag when it throws, so catching this with
everything else would re-enter
+ // the loop and keep polling forever. 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) {
+ lastError.set(error);
Review Comment:
Restored.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,55 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
Review Comment:
Trimmed both, keeping the contract and the interrupt note and leaving the
history to the PR description.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,55 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
Review Comment:
Leaving the current shape, if that is alright. You are right that Awaitility
would fold the executor, last-error and interrupt handling into one call, and I
would pick it for a new helper. Here it would rewrite the method a fifth time
and change the failure type every continuous-mode test reports on, for a diff
that is now small and covered by six cases; the interrupt subtlety it would
retire is pinned by `pollingStopsOnceTheWaitHasGivenUp`. Worth its own change
if you would like it done.
--
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:
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3924853935
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##
@@ -1754,6 +1754,12 @@ static void deltaStreamerTestRunner(HoodieDeltaStreamer
ds, HoodieDeltaStreamer.
});
TestHelpers.waitTillCondition(condition, dsFuture, 360);
Review Comment:
Fixed. The wait is now wrapped so a failure calls `ds.shutdownGracefully()`
(adding any secondary failure as suppressed) before rethrowing, and
`executor.shutdown()` moved into a `finally`. Good catch on the fork reuse
being the actual mechanism.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,55 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
Review Comment:
Applied your suggestion, the guard now checks `executor.isShutdown()` too.
Also fixed `waitFor`: it restores the interrupt and throws rather than
swallowing it, and is bounded at 120s, so the two tests that run it inside
their conditions no longer defeat the stop.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,55 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
+try {
+ Thread.sleep(2000);
+ ret = condition.apply(true);
+ if (ret) {
+log.info("Condition completed successfully");
+ }
+} catch (InterruptedException interrupted) {
+ // shutdownNow below interrupts this thread once the wait has
given up. Thread.sleep clears
+ // the interrupt flag when it throws, so catching this with
everything else would re-enter
+ // the loop and keep polling forever. 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) {
+ lastError.set(error);
+ ret = false;
+}
}
+ return ret;
+
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
voonhous commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3916671585
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##
@@ -1754,6 +1754,12 @@ static void deltaStreamerTestRunner(HoodieDeltaStreamer
ds, HoodieDeltaStreamer.
});
TestHelpers.waitTillCondition(condition, dsFuture, 360);
Review Comment:
**major:** On the timeout path the `AssertionError` leaves this method
before `ds.shutdownGracefully()` or `executor.shutdown()` run, so the
continuous streamer keeps going into the next test in the same fork
(`forkCount=1`, `reuseForks=true`). That test's `UtilitiesTestBase.setup()`
deletes `basePath` and its `teardown()` calls `TestDataSource.resetDataGen()`,
which closes the generators the leaked streamer is still reading. Pre-existing,
but it is the exact path this PR is about. Could we shut the streamer down when
the wait fails (catch, `ds.shutdownGracefully()`, rethrow) and move
`executor.shutdown()` into a `finally`?
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,55 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
Review Comment:
**minor:** The interrupt from `shutdownNow()` is delivered once, so a
condition that swallows it leaves this guard false forever and the poller runs
for the JVM lifetime. `TestHelpers.waitFor` (line 824) does exactly that,
`catch (Throwable)` around `Thread.sleep(5)`, and it runs inside the conditions
at `TestHoodieDeltaStreamer.java:2197` and `:2257`, so for those two tests the
stop that `pollingStopsOnceTheWaitHasGivenUp` pins does not hold. Not blocking.
Could the guard also check the executor, and could `waitFor` restore the flag
(and get a timeout) while here?
```suggestion
while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted() && !executor.isShutdown()) {
```
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,55 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
+try {
+
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy closed pull request #19485: test(utilities): report why the continuous-mode wait timed out URL: https://github.com/apache/hudi/pull/19485 -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5318079257 `Commit Co-author Validation` is red here because of a transient network failure on the runner, not because a commit carries a co-author trailer. The job log shows the `gh api` call itself failed before any commit was inspected: ``` Checking commits of PR #19485 for Claude co-author trailers... Get "https://api.github.com/repos/apache/hudi/pulls/19485/commits?per_page=100": tls: failed to verify certificate: x509: certificate is not valid for any names, but wanted to match api.github.com ##[error]Process completed with exit code 1. ``` The step runs under `bash -e`, so the failed `gh api` aborts it and the check goes red without any commit having been flagged. For what it is worth, the branch has a single commit (`1d01b627`) with no `Co-authored-by` line, the workflow's own regex run against the same endpoint returns nothing, and this workflow passed on an earlier revision of this branch (`ae2e970a`). Could a committer re-run the failed job? Nothing needs changing on the branch. -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5263600244 ## CI report: * 1d01b6279e4c2942d08580bc054e51412f89a71a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16232) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5262753825 ## CI report: * ae2e970ab73ef8505686cdeddce52a606bcd377a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16216) * 1d01b6279e4c2942d08580bc054e51412f89a71a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16232) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5262716717 ## CI report: * ae2e970ab73ef8505686cdeddce52a606bcd377a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16216) * 1d01b6279e4c2942d08580bc054e51412f89a71a UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5262633798 Both review nits addressed in `1d01b62`, and the branch is rebased onto current master so CI gets a fresh run. Replies are on the individual threads. On the one red check — **`test-flink-1 (flink2.1)` is not this PR.** It failed in `ITTestHoodieDataSource.testLookupJoin` with `Runtime Failed to fetch next result`, in `hudi-flink`: ``` [ERROR] Tests run: 264, Failures: 0, Errors: 1, Skipped: 1 -- in org.apache.hudi.table.ITTestHoodieDataSource [ERROR] ITTestHoodieDataSource.testLookupJoin:1263 » Runtime Failed to fetch next result ``` Three things place it outside this change: 1. This PR touches three files, all under `hudi-utilities/src/test/`. There is no path by which a `hudi-utilities` test-only change affects a `hudi-flink` integration test. 2. **The same branch, same code, passed `test-flink-1` five-for-five** in the earlier run (`30811001377`, 2026-08-03). Only the 2026-08-11 run failed, with nothing changed in between — so the failure is not deterministic. 3. `test-flink-1` is green on 11 other recent runs across the repo in the same window, including `master` itself. Azure also reports SUCCESS for the same head. I have deliberately not "fixed" anything for it — the honest read is a flaky Flink lookup-join test, and papering over it here would hide it. Worth its own issue if it recurs; happy to file one. One note for the record, since it cost me a detour and might catch someone else: `mvn compile -pl hudi-utilities` **without** a Spark profile fails on `KafkaAvroSchemaDeserializer.java:71` (`method does not override or implement a method from a supertype`) on plain master too — `confluent.version` 5.5.0 resolves differently, and the `deserialize(String, byte[], Schema)` overload is not on the supertype. With `-Dscala-2.12 -Dspark3.5` it compiles clean. Not a master breakage, just an invocation that needs the profile. -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3763778820
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##
@@ -0,0 +1,111 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+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.
+ *
+ * HUDI-6843 is a flaky timeout in that wait whose only output was
+ * {@code java.util.concurrent.TimeoutException} at this method, with no
indication of which assertion in
+ * the condition never held - the condition's error was logged at debug and
discarded. That is why every
+ * report of the flake looks the same and none of them is actionable.
+ */
+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 several
evaluations. A value close to
+ * one interval would make this test itself flaky on a loaded machine - if
the worker started late and the
+ * first evaluation landed after the timeout, nothing would have been
recorded to report.
+ */
+ private static final int TIMEOUT_WITH_SLACK_SECS = 15;
+
+ @Test
Review Comment:
Renamed to `CONDITION_TIMEOUT_SECS` in `1d01b62`. Agreed the Javadoc already
carries the reasoning, so the name does not need to. All three usages moved
together (the declaration, the call, and the assertion on the message that
reuses it, so the value and the expected text still cannot drift apart).
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,54 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
+try {
+ Thread.sleep(2000);
+ ret = condition.apply(true);
+ if (ret) {
+log.info("Condition completed successfully");
+ }
+} catch (InterruptedException interrupted) {
+ // shutdownNow below interrupts this thread once the wait has
given up. Thread.sleep clears
+ // the interrupt flag when it throws, so catching
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5252172425 ## CI report: * ae2e970ab73ef8505686cdeddce52a606bcd377a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16216) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-agent commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3756716482
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,54 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
+try {
+ Thread.sleep(2000);
+ ret = condition.apply(true);
+ if (ret) {
+log.info("Condition completed successfully");
+ }
+} catch (InterruptedException interrupted) {
+ // shutdownNow below interrupts this thread once the wait has
given up. Thread.sleep clears
+ // the interrupt flag when it throws, so catching this with
everything else would re-enter
+ // the loop and keep polling forever. 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) {
+ lastError.set(error);
+ ret = false;
+}
}
+ return ret;
+});
+try {
+ res.get(timeoutInSecs, TimeUnit.SECONDS);
+} catch (TimeoutException e) {
+ Throwable last = lastError.get();
+ throw new AssertionError(String.format(
Review Comment:
🤖 nit: the nested ternaries inside the `AssertionError` constructor are a
little hard to parse at a glance — might be cleaner to compute `cause` and
`detail` as local variables before the `throw`.
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##
@@ -0,0 +1,111 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+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.
+ *
+ * HUDI-6843 is a flaky timeout in that wait whose only output was
+ * {@code java.util.concurrent.TimeoutException} at this method, with no
indication of which assertion in
+ * the condition never held - the condition's error was logged at debug and
discarded. That is why every
+ * report of the flake looks the
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5251261191 ## CI report: * 541e7bb7a59d68da5a3a244cee683b524f944c7a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15991) * ae2e970ab73ef8505686cdeddce52a606bcd377a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16216) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5251206083 ## CI report: * 541e7bb7a59d68da5a3a244cee683b524f944c7a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15991) * ae2e970ab73ef8505686cdeddce52a606bcd377a UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5172150782 ## CI report: * 541e7bb7a59d68da5a3a244cee683b524f944c7a Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15991) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5171951345 ## CI report: * 3285bc52920c428748544979e0266adb2e805992 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15988) * 541e7bb7a59d68da5a3a244cee683b524f944c7a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15991) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
codecov-commenter commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5167722942 ## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/19485?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report :white_check_mark: All modified and coverable lines are covered by tests. :white_check_mark: Project coverage is 76.97%. Comparing base ([`637996c`](https://app.codecov.io/gh/apache/hudi/commit/637996c5ab2078dfd858a2a425da8f28af1ce789?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`541e7bb`](https://app.codecov.io/gh/apache/hudi/commit/541e7bb7a59d68da5a3a244cee683b524f944c7a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)). :warning: Report is 1 commits behind head on master. Additional details and impacted files ```diff @@Coverage Diff@@ ## master #19485 +/- ## = Coverage 76.97% 76.97% + Complexity3385533854-1 = Files 2575 2575 Lines143378 143378 Branches 1757317573 = Hits 110362 110362 Misses2475524755 Partials 8261 8261 ``` | [Components](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | |---|---|---| | [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.26% <ø> (-0.01%)` | :arrow_down: | | [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.83% <ø> (+<0.01%)` | :arrow_up: | | [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.96% <ø> (ø)` | | | [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.10% <ø> (ø)` | | | [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.65% <ø> (ø)` | | | [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `15.32% <ø> (ø)` | | | [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `63.49% <ø> (ø)` | | | [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.87% <ø> (ø)` | | | [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.60% <ø> (ø)` | | | [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.23% <ø> (ø)` | | | [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `64.00% <ø> (ø)` | | | [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/19485/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ø> (ø)` | | | [Flag](https://app.codecov.io/gh/apache/hudi/pull/19485/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | |---|---|---| |
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5166028118 ## CI report: * 3285bc52920c428748544979e0266adb2e805992 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15988) * 541e7bb7a59d68da5a3a244cee683b524f944c7a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15991) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5165974326 ## CI report: * 3285bc52920c428748544979e0266adb2e805992 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15988) * 541e7bb7a59d68da5a3a244cee683b524f944c7a UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3703601379
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,47 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone()) {
+try {
+ Thread.sleep(2000);
+ ret = condition.apply(true);
+ if (ret) {
+log.info("Condition completed successfully");
+ }
+} catch (Throwable error) {
Review Comment:
Good catch, and you are right — this was a real bug in my own change, not a
nit. Fixed in `541e7bb`.
`shutdownNow()` interrupts the polling thread, but `Thread.sleep` clears the
interrupt flag when it throws, and my `catch (Throwable)` swallowed the
`InterruptedException`, set `ret = false`, and re-entered the loop. `dsFuture`
never completes in the timeout case, so the loop had no other exit: **the
executor shutdown closed nothing in exactly the scenario it was added for.**
Now:
```java
while (!ret && !dsFuture.isDone() &&
!Thread.currentThread().isInterrupted()) {
try {
Thread.sleep(2000);
...
} catch (InterruptedException interrupted) {
Thread.currentThread().interrupt();
break;
} catch (Throwable error) {
lastError.set(error);
ret = false;
}
}
```
Handled before the catch-all, flag restored so the loop condition also sees
it, and deliberately **not** recorded as `lastError` — an interrupt is not a
condition failure, and recording it would have overwritten the assertion error
that the whole change exists to report.
Added `pollingStopsOnceTheWaitHasGivenUp`, which counts evaluations, lets
the wait fail, then waits two more poll intervals and asserts the count is
unchanged. With the interrupt handling reverted it fails with the leak
quantified:
```
AssertionFailedError: the polling thread should have stopped when the wait
gave up,
not carried on in the background ==> expected: <2> but was: <4>
```
Two extra polls after the wait had given up. Green with the fix.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##
@@ -0,0 +1,80 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+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
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
rangareddy commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5165664578 Validation update: I have now run `testUpsertsContinuousModeWithMultipleWritersForConflicts` locally, which I could not do when I opened this. **It passes with these changes** — four runs over both `HoodieTableType` parameters, 8 green executions at ~88s each: ``` run 1: Tests run: 2, Failures: 0, Errors: 0, Skipped: 0 run 2: Tests run: 2, Failures: 0, Errors: 0, Skipped: 0 run 3: Tests run: 2, Failures: 0, Errors: 0, Skipped: 0 run 4: Tests run: 2, Failures: 0, Errors: 0, Skipped: 0 ``` Full `TestHoodieDeltaStreamerWithMultiWriter`: 5 tests, 157.6s, green. `checkstyle:check` and `apache-rat:check` clean. The PR description is updated accordingly. No flake reproduced in those runs, so this change is not masking one. That also means it is not evidence the flake is gone — four local runs on one machine say little about CI, where the original report came from a much busier executor. The claim in this PR is unchanged: it does not fix the flake, it makes the next occurrence name the assertion that was still failing. ### Two environment traps worth recording, since anyone reproducing this will hit them **1. Switching Spark profiles without `clean` leaves bytecode bound to the wrong Spark API.** The test was failing for me with: ``` NoClassDefFoundError: org/apache/spark/sql/internal/SQLConf$LegacyBehaviorPolicy$ at org.apache.spark.sql.avro.AvroDeserializer.(AvroDeserializer.scala:74) at org.apache.spark.sql.avro.HoodieSpark3_5AvroDeserializer. ``` `hudi-spark3-common` vendors `AvroDeserializer`, and as its own comment explains, `LegacyBehaviorPolicy` is nested in `SQLConf` on Spark 3.3/3.4 but a top-level `org.apache.spark.sql.internal.LegacyBehaviorPolicy` on 3.5 — it relies on two wildcard imports so the same source compiles either way. Which one it binds to is therefore fixed at **compile** time. My local `hudi-spark3-common` had been built earlier under a 3.3/3.4 profile. Re-running `mvn install -Dspark3.5` did **not** fix it: the jar was repackaged but the Scala classes were not recompiled, so it still emitted ``` getstatic org/apache/spark/sql/internal/SQLConf$LegacyBehaviorPolicy$.MODULE$ ``` against a Spark 3.5.5 runtime where `spark-catalyst` has no nested `LegacyBehaviorPolicy` at all (it lives in `spark-sql-api` as a top-level class). Only `mvn clean install -Dspark3.5 -Dscala-2.12` rebound it to `internal/LegacyBehaviorPolicy$`. Worth knowing that this silent-until-runtime mismatch is what a stale incremental build looks like here. **2. The test needs Docker, and Ryuk can block it.** `HoodieDeltaStreamerTestBase.initClass` starts a Kafka container via testcontainers, and it aborted in `@BeforeAll` with: ``` IllegalStateException: Could not connect to Ryuk at localhost:50073 at org.testcontainers.utility.RyukResourceReaper.maybeStart at org.apache.hudi.utilities.testutils.KafkaTestUtils.setup(KafkaTestUtils.java:55) ``` No stale Ryuk container existed; the reaper simply failed to bind. `TESTCONTAINERS_RYUK_DISABLED=true` gets past it — Ryuk only reaps leftover containers, so the cost is manual cleanup. For anyone reproducing, the working invocation was: ```bash mvn clean install -DskipTests -Dspark3.5 -Dscala-2.12 -pl hudi-utilities -am TESTCONTAINERS_RYUK_DISABLED=true mvn test -pl hudi-utilities -Dspark3.5 -Dscala-2.12 -Punit-tests \ -Dtest='TestHoodieDeltaStreamerWithMultiWriter#testUpsertsContinuousModeWithMultipleWritersForConflicts' ``` -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-agent commented on code in PR #19485:
URL: https://github.com/apache/hudi/pull/19485#discussion_r3703323676
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java:
##
@@ -763,22 +766,47 @@ static HoodieInstant
assertCommitMetadataForIncrSource(String expected, String t
return lastInstant;
}
+/**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * On timeout the last error the condition threw is attached to the
failure. Without it the only
+ * output is a bare {@link TimeoutException} pointing at this method,
which says nothing about which
+ * assertion never held - the reason HUDI-6843 stayed open: every report
of it looks identical.
+ */
static void waitTillCondition(Function condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future 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 lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+Future res = executor.submit(() -> {
+ boolean ret = false;
+ while (!ret && !dsFuture.isDone()) {
+try {
+ Thread.sleep(2000);
+ ret = condition.apply(true);
+ if (ret) {
+log.info("Condition completed successfully");
+ }
+} catch (Throwable error) {
Review Comment:
🤖 Since this catches `Throwable` (which includes `InterruptedException`) and
just loops again, does the new `executor.shutdownNow()` in the `finally`
actually stop the worker when `dsFuture` never completes — the exact timeout
case this targets? `Thread.sleep` clears the interrupt flag when it throws, so
the loop re-enters `sleep(2000)` and keeps polling. Might be worth breaking out
on interrupt (e.g. `!Thread.currentThread().isInterrupted()` in the loop
condition) so the leak the comment references is really closed.
⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.
##
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestWaitTillCondition.java:
##
@@ -0,0 +1,80 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+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.
+ *
+ * HUDI-6843 is a flaky timeout in that wait whose only output was
+ * {@code java.util.concurrent.TimeoutException} at this method, with no
indication of which assertion in
+ * the condition never held - the condition's error was logged at debug and
discarded. That is why every
+ * report of the flake looks the same and none of them is actionable.
+ */
+class TestWaitTillCondition {
+
+ /** A deltastreamer future that never finishes, as a continuous-mode job
would be. */
+ private static final Future RUNNING = new CompletableFuture<>();
+
+ @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);
Review Comment:
🤖 The poll interval inside
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5165434119 ## CI report: * 3285bc52920c428748544979e0266adb2e805992 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15988) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] test(utilities): report why the continuous-mode wait timed out [hudi]
hudi-bot commented on PR #19485: URL: https://github.com/apache/hudi/pull/19485#issuecomment-5165388486 ## CI report: * 3285bc52920c428748544979e0266adb2e805992 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
