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.
+ *
+ * <p>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<Boolean, Boolean> condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future<Boolean> res = Executors.newSingleThreadExecutor().submit(() -> {
- boolean ret = false;
- while (!ret && !dsFuture.isDone()) {
- try {
- Thread.sleep(2000);
- ret = condition.apply(true);
- log.info("Condition completed successfully");
- } catch (Throwable error) {
- log.debug("Got error waiting for condition", error);
- ret = false;
+ AtomicReference<Throwable> lastError = new AtomicReference<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<Boolean> 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.
+ *
+ * <p>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.
+ *
+ * <p>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<Boolean, Boolean> 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:
[email protected]
