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. + * + * <p>The wait used to fail with a bare {@code TimeoutException} naming only the helper, with the + * condition's own error logged at debug and discarded, so a timeout said nothing about which assertion + * never held (HUDI-6843). + */ +class TestWaitTillCondition { + + /** A deltastreamer future that never finishes, as a continuous-mode job would be. */ + private static final Future<?> RUNNING = new CompletableFuture<>(); + + /** + * The helper polls every 2s, so the timeout has to leave room for at least one evaluation to be recorded. + * 5s is 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. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java: ########## @@ -763,34 +767,92 @@ static HoodieInstant assertCommitMetadataForIncrSource(String expected, String t return lastInstant; } + /** Bound for {@link #waitFor}; generous, since it only exists to stop a hung poll running forever. */ + private static final long WAIT_FOR_TIMEOUT_SECS = 120; + + /** + * Polls {@code condition} until it holds, the deltastreamer future finishes, or the timeout expires. + * + * <p>On timeout the last error the condition threw is attached to the failure, so the report names the + * assertion that never held rather than only this method. + */ static void waitTillCondition(Function<Boolean, Boolean> condition, Future dsFuture, long timeoutInSecs) throws Exception { - Future<Boolean> res = Executors.newSingleThreadExecutor().submit(() -> { - boolean ret = false; - while (!ret && !dsFuture.isDone()) { - try { - Thread.sleep(2000); - ret = condition.apply(true); - log.info("Condition completed successfully"); - } catch (Throwable error) { - log.debug("Got error waiting for condition", error); - ret = false; + AtomicReference<Throwable> lastError = new AtomicReference<>(); + AtomicInteger completedEvaluations = new AtomicInteger(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future<Boolean> res = executor.submit(() -> { + boolean ret = false; + // The executor check matters as well as the interrupt flag: the interrupt from shutdownNow is + // delivered once, and a condition that swallows it would otherwise leave the flag clear and keep + // this thread polling for the lifetime of the JVM. + while (!ret && !dsFuture.isDone() && !Thread.currentThread().isInterrupted() && !executor.isShutdown()) { + try { + Thread.sleep(2000); + ret = condition.apply(true); + completedEvaluations.incrementAndGet(); + if (ret) { + log.info("Condition completed successfully"); + } + } catch (InterruptedException interrupted) { + // Thread.sleep clears the interrupt flag when it throws, so catching this with everything + // else would re-enter the loop. Restore the flag and stop; this is not a condition failure, + // so it is deliberately not recorded as one. + Thread.currentThread().interrupt(); + break; + } catch (Throwable error) { + log.debug("Got error waiting for condition", error); + lastError.set(error); + completedEvaluations.incrementAndGet(); + ret = false; + } } + return ret; + }); + try { + res.get(timeoutInSecs, TimeUnit.SECONDS); + } catch (TimeoutException e) { + Throwable last = lastError.get(); + int completed = completedEvaluations.get(); + String detail; + if (completed == 0) { + // Distinguishes a condition that is stuck part-way through its first evaluation - a hung Spark + // read, say - from one that simply kept returning false. + detail = "No evaluation of the condition completed, so it was still running or never started."; + } else if (last == null) { + detail = String.format("%d evaluations completed and returned false without throwing, " + + "so there is no further detail.", completed); + } else { + detail = String.format("%d evaluations completed; the last failure reported was: %s", completed, last); + } + Throwable cause = last == null ? e : last; + throw new AssertionError( + String.format("Condition was not met within %d seconds. %s", timeoutInSecs, detail), cause); } - return ret; - }); - res.get(timeoutInSecs, TimeUnit.SECONDS); + } finally { + // this used to leak the polling thread on every call, and it is called by every continuous-mode test + executor.shutdownNow(); Review Comment: 🤖 nit: the "this used to leak" comment describes history rather than the current code; it might be worth rephrasing to why it matters now, e.g. "stop the polling thread; this runs once per continuous-mode test". <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
