This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 56eae7b48089 test(utilities): report why the continuous-mode wait
timed out (#19485)
56eae7b48089 is described below
commit 56eae7b48089478616e5109a15efa11aae23e30e
Author: Ranga Reddy <[email protected]>
AuthorDate: Tue Sep 8 22:36:20 2026 +0530
test(utilities): report why the continuous-mode wait timed out (#19485)
test(utilities): diagnose wait timeouts (#19485)
HoodieDeltaStreamerTestBase.TestHelpers.waitTillCondition polled its
condition every two seconds and swallowed whatever it threw, so when a
continuous-mode test timed out the failure named only the helper. Every
report of the flaky multi-writer test in #16228 (HUDI-6843) is that
bare TimeoutException, and none says which of the four assertions in
its condition was still failing.
The wait now keeps the last Throwable the condition threw and attaches
it as the cause, counts completed evaluations, and reports one of three
shapes on timeout: the condition's own error, no evaluation completed,
or evaluations that returned false. The condition helpers name
themselves in their messages. The "Condition completed successfully"
log on a false result is gone, and the exit where the streamer future
finishes first is logged at warn with the same detail.
A failure no longer leaves the streamer running into the next test
(surefire runs this module with forkCount=1, reuseForks=true). The
runner stops it on any failure, each wait bounded at 30s with at most
two in sequence, force-stops the ingestion service if the stop outlives
its bound, and surfaces a streamer that died instead of waiting two
minutes for "Deltastreamer should have shutdown by now". The polling
executor is shut down, InterruptedException is handled ahead of the
catch-all so that shutdown takes effect, and waitFor is bounded at 120s
and restores the interrupt.
JavaTestUtils.checkNestedExceptionContains tolerates a null message in
the chain. TestHoodieDeltaStreamerWithMultiWriter records whether its
backfill prerequisite held, so an expected-conflict run whose writers
never overlapped says so instead of blaming conflict handling, and its
two sibling prep jobs get the MOR compaction settings HUDI-6445 gave
the ForConflicts one.
TestDeltaStreamerTestHelpers (14 tests, no Spark) and TestJavaTestUtils
cover the helpers directly. This does not fix the flake in #16228; it
makes the next occurrence diagnosable.
---------
Co-authored-by: voon <[email protected]>
---
.../hudi/common/testutils/JavaTestUtils.java | 6 +-
.../hudi/common/testutils/TestJavaTestUtils.java | 58 +++
.../deltastreamer/HoodieDeltaStreamerTestBase.java | 178 ++++++--
.../TestDeltaStreamerTestHelpers.java | 455 +++++++++++++++++++++
.../deltastreamer/TestHoodieDeltaStreamer.java | 145 ++++++-
.../TestHoodieDeltaStreamerWithMultiWriter.java | 33 +-
6 files changed, 828 insertions(+), 47 deletions(-)
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/testutils/JavaTestUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/testutils/JavaTestUtils.java
index e91e2559339a..e023fed82a65 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/testutils/JavaTestUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/testutils/JavaTestUtils.java
@@ -28,7 +28,11 @@ public class JavaTestUtils {
Throwable throwable = t;
boolean res = false;
while (throwable != null) {
- if (throwable.getMessage().contains(errorMsg)) {
+ // A null message is treated as "not a match" rather than an NPE: a
TimeoutException raised before its
+ // condition ever threw carries no message, and neither do many wrapped
NPEs. String.valueOf would also
+ // avoid the NPE, but would make an errorMsg of "null" match a
message-less throwable: a trap, not a match.
+ String message = throwable.getMessage();
+ if (message != null && message.contains(errorMsg)) {
res = true;
break;
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/testutils/TestJavaTestUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/testutils/TestJavaTestUtils.java
new file mode 100644
index 000000000000..bcde3e263b17
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/testutils/TestJavaTestUtils.java
@@ -0,0 +1,58 @@
+/*
+ * 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.common.testutils;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Pins the null-message handling of {@link
JavaTestUtils#checkNestedExceptionContains}: a throwable
+ * with no message must neither NPE the walk nor match an errorMsg of "null".
The helper's callers all
+ * live in other modules, so the test lives beside the helper to keep it
covered where it is defined.
+ */
+public class TestJavaTestUtils {
+
+ @Test
+ public void testNullMessageMidChainDoesNotStopTheWalk() {
+ // Pre-fix this NPE'd on the message-less link. The explicit (String) null
cast matters:
+ // new RuntimeException(cause) would set the message to the cause's
toString and hide the case entirely.
+ Throwable deepest = new IllegalArgumentException("boom");
+ Throwable t = new RuntimeException("head", new RuntimeException((String)
null, deepest));
+ assertTrue(JavaTestUtils.checkNestedExceptionContains(t, "boom"));
+ }
+
+ @Test
+ public void testErrorMsgNullDoesNotMatchMessagelessThrowable() {
+ assertFalse(JavaTestUtils.checkNestedExceptionContains(new
RuntimeException((String) null), "null"));
+ }
+
+ @Test
+ public void testMessageContainingTextMatchesOnHead() {
+ assertTrue(JavaTestUtils.checkNestedExceptionContains(new
RuntimeException("a boom happened"), "boom"));
+ }
+
+ @Test
+ public void testNoMatchAnywhereInChainReturnsFalse() {
+ Throwable t = new RuntimeException("head", new
IllegalStateException("cause"));
+ assertFalse(JavaTestUtils.checkNestedExceptionContains(t, "boom"));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
index 8bece4c3df2c..3207e980c272 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
@@ -76,9 +76,13 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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 java.util.function.BooleanSupplier;
import java.util.function.Function;
@@ -550,7 +554,8 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
void assertRecordCount(long expected, String tablePath, SQLContext
sqlContext) {
sqlContext.clearCache();
long recordCount =
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tablePath).count();
- assertEquals(expected, recordCount);
+ // Named, so a one-line failure report says which of the near-identical
count helpers it came from.
+ assertEquals(expected, recordCount, () -> "assertRecordCount(" + tablePath
+ ")");
}
void assertDistinctRecordCount(long expected, String tablePath, SQLContext
sqlContext) {
@@ -572,7 +577,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
sqlContext.read().options(hudiOpts).format("org.apache.hudi").load(tablePath).registerTempTable("tmp_trips");
long recordCount =
sqlContext.sql("select * from tmp_trips where haversine_distance is
not NULL").count();
- assertEquals(expected, recordCount);
+ assertEquals(expected, recordCount, () -> "assertDistanceCount(" +
tablePath + ")");
}
void assertDistanceCountWithExactValue(long expected, String tablePath,
SQLContext sqlContext) {
@@ -604,6 +609,15 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
public static class TestHelpers {
+ /**
+ * Default bound for {@link #waitFor(BooleanSupplier)}; generous, since it
only exists to stop a hung poll
+ * running forever.
+ */
+ private static final long WAIT_FOR_TIMEOUT_SECS = 120;
+
+ /** The cadence production callers poll at; the helper's own tests pass a
faster one of their own. */
+ private static final long POLL_INTERVAL_MS = 2000;
+
static HoodieDeltaStreamer.Config makeDropAllConfig(String basePath,
WriteOperationType op) {
return makeConfig(basePath, op,
Collections.singletonList(TestHoodieDeltaStreamer.DropAllTransformer.class.getName()));
}
@@ -707,7 +721,8 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().getCommitAndReplaceTimeline().filterCompletedInstants();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numCompactionCommits = timeline.countInstants();
- assertTrue(minExpected <= numCompactionCommits, "Got=" +
numCompactionCommits + ", exp >=" + minExpected);
+ assertTrue(minExpected <= numCompactionCommits,
+ "assertAtleastNCompactionCommits: Got=" + numCompactionCommits + ",
exp >=" + minExpected);
}
static void assertAtleastNDeltaCommits(int minExpected, String tablePath) {
@@ -715,7 +730,8 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numDeltaCommits = timeline.countInstants();
- assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ",
exp >=" + minExpected);
+ assertTrue(minExpected <= numDeltaCommits,
+ "assertAtleastNDeltaCommits: Got=" + numDeltaCommits + ", exp >=" +
minExpected);
}
static void assertAtleastNCompactionCommitsAfterCommit(int minExpected,
String lastSuccessfulCommit, String tablePath) {
@@ -723,7 +739,8 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().getCommitAndReplaceTimeline().findInstantsAfter(lastSuccessfulCommit).filterCompletedInstants();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numCompactionCommits = timeline.countInstants();
- assertTrue(minExpected <= numCompactionCommits, "Got=" +
numCompactionCommits + ", exp >=" + minExpected);
+ assertTrue(minExpected <= numCompactionCommits,
+ "assertAtleastNCompactionCommitsAfterCommit: Got=" +
numCompactionCommits + ", exp >=" + minExpected);
}
static void assertAtleastNDeltaCommitsAfterCommit(int minExpected, String
lastSuccessfulCommit, String tablePath) {
@@ -731,7 +748,8 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.reloadActiveTimeline().getDeltaCommitTimeline().findInstantsAfter(lastSuccessfulCommit).filterCompletedInstants();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numDeltaCommits = timeline.countInstants();
- assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ",
exp >=" + minExpected);
+ assertTrue(minExpected <= numDeltaCommits,
+ "assertAtleastNDeltaCommitsAfterCommit: Got=" + numDeltaCommits + ",
exp >=" + minExpected);
}
static HoodieInstant assertCommitMetadata(String expected, String
tablePath, int totalCommits)
@@ -763,22 +781,111 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
return lastInstant;
}
+ /**
+ * Polls {@code condition} until it holds, the deltastreamer future
finishes, or the timeout expires.
+ *
+ * <p>On timeout the last error the condition threw is attached to the
failure, so the report names the
+ * assertion that never held rather than only this method.
+ */
static void waitTillCondition(Function<Boolean, Boolean> condition, Future
dsFuture, long timeoutInSecs) throws Exception {
- Future<Boolean> res = Executors.newSingleThreadExecutor().submit(() -> {
- boolean ret = false;
- while (!ret && !dsFuture.isDone()) {
- try {
- Thread.sleep(2000);
- ret = condition.apply(true);
- log.info("Condition completed successfully");
- } catch (Throwable error) {
- log.debug("Got error waiting for condition", error);
- ret = false;
+ 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<Boolean, Boolean> condition, Future
dsFuture, long timeoutInSecs,
+ long pollIntervalMs) throws Exception {
+ 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(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)) {
+ // Read in the same order as the timeout path below, so
describeProgress's note on the read
+ // order holds for both callers. The worker has finished here, so
neither can be stale.
+ int completed = completedEvaluations.get();
+ Throwable last = lastError.get();
+ log.warn("Wait ended without the condition holding: {}. {}",
+ dsFuture.isDone() ? "the deltastreamer future finished" : "the
polling thread was stopped",
+ describeProgress(last, completed));
+ }
+ } catch (TimeoutException e) {
+ int completed = completedEvaluations.get();
+ Throwable last = lastError.get();
+ Throwable cause = last == null ? e : last;
+ AssertionError failure = new AssertionError(describeTimeout(last,
completed, 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) {
+ return String.format("Condition was not met within %d seconds. %s",
timeoutInSecs,
+ describeProgress(last, completed));
+ }
+
+ /** The progress half of the report on its own, so an exit that is not a
timeout can say the same thing. */
+ static String describeProgress(Throwable last, int completed) {
+ String detail;
+ if (last != null) {
+ // Tested first because the worker records the error before it
increments the counter: a timeout
+ // landing between the two would otherwise report that no evaluation
completed. lastError is only
+ // ever set, never cleared, so its presence is decisive in every
interleaving. The reader takes the
+ // counter first, the opposite order, so any skew between the two
reads lands in this branch - a real
+ // error reported with a possibly stale count - rather than in the
"returned false without throwing"
+ // branch, which would deny an error that did happen.
+ detail = String.format("%d evaluations completed; the last failure
reported was: %s", completed, last);
+ } else 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 {
+ detail = String.format("%d evaluations completed and returned false
without throwing, "
+ + "so there is no further detail.", completed);
+ }
+ return detail;
}
/**
@@ -786,11 +893,24 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
* @param booleanSupplier Boolean supplier
*/
static void waitFor(BooleanSupplier booleanSupplier) {
+ waitFor(booleanSupplier, WAIT_FOR_TIMEOUT_SECS);
+ }
+
+ static void waitFor(BooleanSupplier booleanSupplier, long timeoutSecs) {
+ // Bounded, and the interrupt is restored rather than swallowed: this
runs inside conditions passed to
+ // waitTillCondition, so swallowing it would defeat the stop that
shutdownNow signals.
+ long deadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(timeoutSecs);
while (!booleanSupplier.getAsBoolean()) {
+ // Subtraction rather than a bare comparison: nanoTime is only
meaningful as a difference, so this is
+ // the form its javadoc documents as overflow-safe.
+ if (System.nanoTime() - deadline > 0) {
+ throw new AssertionError(String.format("Condition did not hold
within %d seconds", timeoutSecs));
+ }
try {
Thread.sleep(5);
- } catch (Throwable error) {
- log.debug("Got error waiting for condition", error);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError("Interrupted while waiting for condition",
interrupted);
}
}
}
@@ -800,7 +920,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().filterCompletedInstants();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numDeltaCommits = timeline.countInstants();
- assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ",
exp >=" + minExpected);
+ assertTrue(minExpected <= numDeltaCommits, "assertAtLeastNCommits: Got="
+ numDeltaCommits + ", exp >=" + minExpected);
}
static void assertAtLeastNReplaceCommits(int minExpected, String
tablePath) {
@@ -808,7 +928,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().getCompletedReplaceTimeline();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numDeltaCommits = timeline.countInstants();
- assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ",
exp >=" + minExpected);
+ assertTrue(minExpected <= numDeltaCommits,
"assertAtLeastNReplaceCommits: Got=" + numDeltaCommits + ", exp >=" +
minExpected);
}
static void assertPendingIndexCommit(String tablePath) {
@@ -816,7 +936,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.reloadActiveTimeline().getAllCommitsTimeline().filterPendingIndexTimeline();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numIndexCommits = timeline.countInstants();
- assertEquals(1, numIndexCommits, "Got=" + numIndexCommits + ", exp=1");
+ assertEquals(1, numIndexCommits, "assertPendingIndexCommit: Got=" +
numIndexCommits + ", exp=1");
}
static void assertCompletedIndexCommit(String tablePath) {
@@ -824,7 +944,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.reloadActiveTimeline().getAllCommitsTimeline().filterCompletedIndexTimeline();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numIndexCommits = timeline.countInstants();
- assertEquals(1, numIndexCommits, "Got=" + numIndexCommits + ", exp=1");
+ assertEquals(1, numIndexCommits, "assertCompletedIndexCommit: Got=" +
numIndexCommits + ", exp=1");
}
static void assertNoReplaceCommits(String tablePath) {
@@ -832,7 +952,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().getCompletedReplaceTimeline();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numDeltaCommits = timeline.countInstants();
- assertEquals(0, numDeltaCommits, "Got=" + numDeltaCommits + ", exp =" +
0);
+ assertEquals(0, numDeltaCommits, "assertNoReplaceCommits: Got=" +
numDeltaCommits + ", exp =" + 0);
}
static void assertAtLeastNClusterRequests(int minExpected, String
tablePath) {
@@ -840,7 +960,7 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().filterPendingClusteringTimeline();
log.info("Timeline Instants={}", meta.getActiveTimeline().getInstants());
int numDeltaCommits = timeline.countInstants();
- assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ",
exp >=" + minExpected);
+ assertTrue(minExpected <= numDeltaCommits,
"assertAtLeastNClusterRequests: Got=" + numDeltaCommits + ", exp >=" +
minExpected);
}
static void assertAtLeastNCommitsAfterRollback(int minExpectedRollback,
int minExpectedCommits, String tablePath) {
@@ -848,13 +968,13 @@ public class HoodieDeltaStreamerTestBase extends
UtilitiesTestBase {
HoodieTimeline timeline =
meta.getActiveTimeline().getRollbackTimeline().filterCompletedInstants();
log.info("Rollback Timeline Instants={}",
meta.getActiveTimeline().getInstants());
int numRollbackCommits = timeline.countInstants();
- assertTrue(minExpectedRollback <= numRollbackCommits, "Got=" +
numRollbackCommits + ", exp >=" + minExpectedRollback);
+ assertTrue(minExpectedRollback <= numRollbackCommits,
"assertAtLeastNCommitsAfterRollback: Got=" + numRollbackCommits + ", exp >=" +
minExpectedRollback);
HoodieInstant firstRollback = timeline.getInstants().get(0);
//
HoodieTimeline commitsTimeline =
meta.getActiveTimeline().filterCompletedInstants()
.filter(instant -> compareTimestamps(instant.requestedTime(),
GREATER_THAN, firstRollback.requestedTime()));
int numCommits = commitsTimeline.countInstants();
- assertTrue(minExpectedCommits <= numCommits, "Got=" + numCommits + ",
exp >=" + minExpectedCommits);
+ assertTrue(minExpectedCommits <= numCommits,
"assertAtLeastNCommitsAfterRollback: Got=" + numCommits + ", exp >=" +
minExpectedCommits);
}
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java
new file mode 100644
index 000000000000..2ac43cb9cb0d
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestDeltaStreamerTestHelpers.java
@@ -0,0 +1,455 @@
+/*
+ * 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.ingestion.HoodieIngestionService;
+import org.apache.hudi.utilities.streamer.NoNewDataTerminationStrategy;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers the deltastreamer test helpers every continuous-mode test runs on:
the wait in
+ * {@code HoodieDeltaStreamerTestBase.TestHelpers} and the runner in {@code
TestHoodieDeltaStreamer}.
+ *
+ * <p>The wait used to fail with a bare {@code TimeoutException} naming only
the helper, with the
+ * condition's own error logged at debug and discarded, so a timeout said
nothing about which assertion
+ * never held (HUDI-6843).
+ */
+class TestDeltaStreamerTestHelpers {
+
+ /** A deltastreamer future that never finishes, as a continuous-mode job
would be. */
+ private static final Future<?> RUNNING = new CompletableFuture<>();
+
+ /**
+ * The 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;
+
+ /**
+ * Slow enough that the poll spends almost all of its time asleep, so a
directly delivered interrupt lands in
+ * the sleep rather than in the condition.
+ */
+ private static final long SLOW_POLL_INTERVAL_MS = 500;
+
+ /**
+ * Generous, so a catch-all that polled through the interrupt fails by
throwing at the timeout rather than
+ * by a near miss.
+ */
+ private static final int INTERRUPT_TIMEOUT_SECS = 10;
+
+ /** The stop bound the stop-path cases drive, instead of the production one.
*/
+ private static final long FAST_STOP_TIMEOUT_SECS = 1;
+
+ /**
+ * At most two of the stop's bounds can run in sequence (a stop that times
out skips the join, and a stop that
+ * returns leaves nothing for the close-wait), so two seconds of this is
bound and the rest is slack. A stop that
+ * lost its bound would take the ten minutes the mock sleeps.
+ */
+ private static final long STOP_PATH_CEILING_SECS = 5;
+
+ @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());
+ assertEquals(1, error.getSuppressed().length,
+ () -> "the timeout should stay attached as a suppressed exception once
the condition's error becomes "
+ + "the cause, but the suppressed list was: " +
Arrays.toString(error.getSuppressed()));
+ assertInstanceOf(TimeoutException.class, error.getSuppressed()[0],
+ "the suppressed exception should be the timeout the wait gave up on");
+ }
+
+ /**
+ * The interrupt from {@code shutdownNow} is delivered once, and {@code
Thread.sleep} clears the flag when it
+ * throws, so a condition that swallows it without restoring it leaves the
loop with no interrupt to see. The
+ * {@code executor.isShutdown()} guard is what stops the worker in that case.
+ */
+ @Test
+ void pollingStopsEvenWhenTheConditionSwallowsTheInterrupt() throws Exception
{
+ AtomicInteger polls = new AtomicInteger();
+ AtomicReference<Thread> poller = new AtomicReference<>();
+
+ assertThrows(AssertionError.class,
+ () -> waitTillCondition(
+ ignored -> {
+ poller.set(Thread.currentThread());
+ polls.incrementAndGet();
+ try {
+ Thread.sleep(TimeUnit.SECONDS.toMillis(60));
+ } catch (InterruptedException interrupted) {
+ // The missing Thread.currentThread().interrupt() is the point
of the test: a condition that
+ // swallows the interrupt is exactly what the isShutdown()
guard exists for, so do not "fix"
+ // this catch.
+ }
+ return false;
+ }, RUNNING, CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS));
+
+ assertTrue(polls.get() > 0,
+ "the condition should have been entered at least once, otherwise there
is no polling thread to join");
+ poller.get().join(TimeUnit.SECONDS.toMillis(5));
+ assertFalse(poller.get().isAlive(),
+ "the isShutdown() guard should have stopped the polling thread even
though the condition swallowed "
+ + "the interrupt without restoring the flag");
+ }
+
+ /**
+ * A condition that hangs part-way through its first evaluation is a
different failure from one that keeps
+ * returning false, and the report has to say which: with no completed
evaluation there is no last error,
+ * and claiming the condition "returned false without throwing" would assert
the wrong thing.
+ */
+ @Test
+ void timeoutDistinguishesAConditionThatNeverCompletedAnEvaluation() {
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> waitTillCondition(
+ ignored -> {
+ try {
+ Thread.sleep(TimeUnit.SECONDS.toMillis(60));
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return true;
+ }, RUNNING, CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS));
+
+ assertTrue(error.getMessage().contains("No evaluation of the condition
completed"),
+ () -> "a condition still running its first evaluation should be
reported as such, but was: "
+ + error.getMessage());
+ assertInstanceOf(TimeoutException.class, error.getCause(),
+ "with no error recorded the timeout itself should be the cause");
+ assertNull(error.getCause().getMessage(),
+ "the cause is a message-less TimeoutException, which is the shape that
made the null check in "
+ + "JavaTestUtils.checkNestedExceptionContains necessary");
+ }
+
+ /**
+ * The branch for a condition that swallows its own failure and returns
false, as {@code testHoodieIndexer}
+ * does. The HUDI-6843 condition is not one of those: it only ever throws,
so a real timeout there takes the
+ * last-error branch instead. This one has to say how many evaluations ran,
since that is the only signal
+ * separating it from a condition that never completed one.
+ */
+ @Test
+ void timeoutReportsEvaluationsThatReturnedFalse() {
+ AtomicInteger polls = new AtomicInteger();
+
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> waitTillCondition(
+ ignored -> {
+ polls.incrementAndGet();
+ return false;
+ }, RUNNING, CONDITION_TIMEOUT_SECS, FAST_POLL_INTERVAL_MS));
+
+ assertTrue(error.getMessage().contains("returned false without throwing"),
+ () -> "a condition that kept returning false should be reported as
such, but was: " + error.getMessage());
+ Matcher count = Pattern.compile("(\\d+) evaluations
completed").matcher(error.getMessage());
+ assertTrue(count.find(),
+ () -> "the report should carry an evaluation count, but was: " +
error.getMessage());
+ // The worker records an evaluation only after the condition returns, so
at most one can be in flight when
+ // the timeout reads the counter, and none can follow it because the loop
then sees the shutdown. The tally
+ // the condition kept is therefore the reported count or exactly one more.
+ int reported = Integer.parseInt(count.group(1));
+ int observed = polls.get();
+ assertTrue(reported >= 1 && reported <= observed && observed - reported <=
1,
+ () -> "the count should be the real number of completed evaluations,
but the report said " + reported
+ + " while the condition ran " + observed + " times: " +
error.getMessage());
+ }
+
+ /**
+ * The bound exists so a hung poll cannot run for the life of the JVM. Both
production callers of waitFor
+ * are currently disabled (HUDI-8951), so this is the only thing exercising
it.
+ */
+ @Test
+ void waitForGivesUpAtItsBound() {
+ AssertionError error = assertThrows(AssertionError.class,
+ () -> waitFor(() -> false, 1));
+
+ assertTrue(error.getMessage().contains("did not hold within 1 seconds"),
+ () -> "the bound should name itself in the failure, but was: " +
error.getMessage());
+ }
+
+ /**
+ * When a streamer configured with a post-write termination strategy dies,
the wait returns because the
+ * future is done, and {@code deltaStreamerTestRunner} has to surface that
failure. Without the
+ * {@code dsFuture.isDone()} guard it would instead call {@code
awaitDeltaStreamerShutdown} and report the
+ * misleading "Deltastreamer should have shutdown by now" two minutes later
- here, on a mock with no
+ * ingestion service, it would NPE.
+ */
+ @Test
+ void dyingStreamerWithTerminationStrategyIsSurfacedNotWaitedOut() throws
Exception {
+ HoodieDeltaStreamer ds = Mockito.mock(HoodieDeltaStreamer.class);
+ Mockito.doThrow(new IllegalStateException("source is
unreachable")).when(ds).sync();
+ HoodieDeltaStreamer.Config cfg = new HoodieDeltaStreamer.Config();
+ cfg.postWriteTerminationStrategyClass =
NoNewDataTerminationStrategy.class.getName();
+
+ ExecutionException failure = assertThrows(ExecutionException.class,
+ () -> TestHoodieDeltaStreamer.deltaStreamerTestRunner(ds, cfg, ignored
-> false, "dying_ds_job"));
+
+ assertTrue(JavaTestUtils.checkNestedExceptionContains(failure, "source is
unreachable"),
+ () -> "the streamer's own failure should be surfaced, but was: " +
failure);
+ }
+
+ @Test
+ void satisfiedConditionReturnsNormally() {
+ assertDoesNotThrow(() -> waitTillCondition(
+ ignored -> true, RUNNING, NEVER_REACHED_TIMEOUT_SECS,
FAST_POLL_INTERVAL_MS));
+ }
+
+ /**
+ * When the streamer finishes first the wait returns rather than failing,
and the caller
+ * ({@code deltaStreamerTestRunner}) surfaces the streamer's own outcome.
Pinned so the timeout handling
+ * above does not turn this into a failure.
+ */
+ @Test
+ void finishedStreamerEndsTheWaitWithoutFailing() {
+ Future<?> finished = CompletableFuture.completedFuture(null);
+
+ assertDoesNotThrow(() -> waitTillCondition(
+ ignored -> false, finished, NEVER_REACHED_TIMEOUT_SECS,
FAST_POLL_INTERVAL_MS));
+ }
+
+ /**
+ * Not deterministically reproducible through the helper: it reads the
evaluation counter before the last
+ * error, so it can see a recorded error alongside a count of zero, but only
on an interleaving a test cannot
+ * force. Pinned here by calling the report builder directly.
+ */
+ @Test
+ void describeTimeoutReportsAnErrorEvenWithNoCompletedEvaluation() {
+ String message = describeTimeout(new AssertionError("boom"), 0,
CONDITION_TIMEOUT_SECS);
+
+ assertTrue(message.contains("boom"),
+ () -> "an error recorded before the counter caught up should still be
reported, but was: " + message);
+ assertFalse(message.contains("No evaluation of the condition completed"),
+ () -> "a recorded error should not be reported as no evaluation having
completed, but was: " + message);
+ }
+
+ /**
+ * The {@code InterruptedException} branch on its own: an interrupt
delivered while the poll is sleeping has
+ * to end the wait, rather than be recorded as a condition failure and
polled through. A catch-all around the
+ * sleep would clear the flag and keep polling until the timeout, which is
what this discriminates.
+ *
+ * <p>The slow poll makes the sleep the overwhelmingly likely place for the
interrupt to land. Landing outside
+ * it is also a pass, since the loop guard then ends the wait, so this
cannot flake either way.
+ */
+ @Test
+ void directInterruptEndsTheWaitWithoutRunningToTheTimeout() throws Exception
{
+ AtomicReference<Thread> poller = new AtomicReference<>();
+ CountDownLatch polling = new CountDownLatch(1);
+ AtomicBoolean interruptSent = new AtomicBoolean();
+ Thread interrupter = new Thread(() -> {
+ try {
+ if (polling.await(5, TimeUnit.SECONDS)) {
+ poller.get().interrupt();
+ interruptSent.set(true);
+ }
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ interrupter.start();
+
+ assertDoesNotThrow(() -> waitTillCondition(
+ ignored -> {
+ poller.set(Thread.currentThread());
+ polling.countDown();
+ return false;
+ }, RUNNING, INTERRUPT_TIMEOUT_SECS, SLOW_POLL_INTERVAL_MS));
+ interrupter.join(TimeUnit.SECONDS.toMillis(5));
+
+ assertTrue(interruptSent.get(),
+ "the interrupt was never sent because the first poll did not complete
within 5s, so this run says "
+ + "nothing about the interrupt path");
+ }
+
+ /**
+ * A stop that throws does not excuse leaving the ingest task running: the
wait falls through to the join,
+ * which times out, and the task is force-stopped and cancelled rather than
left reading into the next test.
+ */
+ @Test
+ void stopThatThrowsStillCancelsTheIngestTask() {
+ HoodieDeltaStreamer ds = Mockito.mock(HoodieDeltaStreamer.class);
+ Mockito.doThrow(new IllegalStateException("stop blew
up")).when(ds).shutdownGracefully();
+
+ assertStopEndsWithTheIngestTaskCancelled(ds);
+ }
+
+ /**
+ * The same outcome when the stop hangs instead of throwing, which is the
case the bound exists for:
+ * {@code shutdownGracefully} can await the ingest executor for up to 24
hours. The preemptive ceiling in the
+ * helper is what pins the bound here.
+ */
+ @Test
+ void stopThatHangsIsBoundedAndCancelsTheIngestTask() {
+ HoodieDeltaStreamer ds = Mockito.mock(HoodieDeltaStreamer.class);
+ Mockito.doAnswer(invocation -> {
+ Thread.sleep(TimeUnit.MINUTES.toMillis(10));
+ return null;
+ }).when(ds).shutdownGracefully();
+
+ assertStopEndsWithTheIngestTaskCancelled(ds);
+ }
+
+ /**
+ * The branch the bound exists for. When the stop is still running after its
bound, the ingestion service is
+ * force-stopped and the ingest task cancelled, and the close-wait gives up
after its own bound and lets the
+ * stop run on rather than block the next test.
+ */
+ @Test
+ void stopThatOutlivesItsBoundIsForceStoppedAndLetRunOn() {
+ HoodieDeltaStreamer ds = Mockito.mock(HoodieDeltaStreamer.class);
+ HoodieIngestionService service =
Mockito.mock(HoodieIngestionService.class);
+ Mockito.when(ds.getIngestionService()).thenReturn(service);
+ CountDownLatch release = new CountDownLatch(1);
+ stubStopThatOutlivesTheInterrupt(ds, release);
+
+ try {
+ long startedAt = System.nanoTime();
+ assertStopEndsWithTheIngestTaskCancelled(ds);
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startedAt);
+
+ Mockito.verify(service).shutdown(true);
+ // The stop's own bound expires first, then the close-wait runs for its
bound before giving up, so the
+ // branch that lets the stop run on costs at least two bounds. A lower
bound cannot flake under load.
+ assertTrue(elapsedMs >= TimeUnit.SECONDS.toMillis(2 *
FAST_STOP_TIMEOUT_SECS),
+ () -> "the close-wait should have run for its bound after the stop's
bound expired, which is the branch "
+ + "that lets the stop run on, but the helper returned after " +
elapsedMs + "ms");
+ } finally {
+ release.countDown();
+ }
+ }
+
+ /**
+ * A caller that arrives already interrupted must not skip the close-wait:
the flag would make
+ * {@code awaitTermination} throw at once. The wait runs for its bound and
the interrupt is restored
+ * afterwards.
+ */
+ @Test
+ void interruptedCallerStillGetsTheBoundedCloseWait() {
+ HoodieDeltaStreamer ds = Mockito.mock(HoodieDeltaStreamer.class);
+ CountDownLatch release = new CountDownLatch(1);
+ stubStopThatOutlivesTheInterrupt(ds, release);
+
+ try {
+ assertTimeoutPreemptively(Duration.ofSeconds(STOP_PATH_CEILING_SECS), ()
-> {
+ Thread.currentThread().interrupt();
+ long startedAt = System.nanoTime();
+
+ TestHoodieDeltaStreamer.stopLeakedStreamer(ds, null,
FAST_STOP_TIMEOUT_SECS);
+
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startedAt);
+ assertTrue(Thread.interrupted(),
+ "the caller's interrupt should be restored after the wait, not
lost");
+ assertTrue(elapsedMs >=
TimeUnit.SECONDS.toMillis(FAST_STOP_TIMEOUT_SECS),
+ () -> "the close-wait should have run for its bound instead of
being skipped because the caller "
+ + "was interrupted, but returned after " + elapsedMs + "ms");
+ });
+ } finally {
+ release.countDown();
+ }
+ }
+
+ /** Runs the stop against a ten-minute ingest task and asserts it ends,
within the ceiling, with the task cancelled. */
+ private static void
assertStopEndsWithTheIngestTaskCancelled(HoodieDeltaStreamer ds) {
+ ExecutorService ingest = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> dsFuture = ingest.submit(() -> {
+ Thread.sleep(TimeUnit.MINUTES.toMillis(10));
+ return null;
+ });
+
+ assertTimeoutPreemptively(Duration.ofSeconds(STOP_PATH_CEILING_SECS),
+ () -> TestHoodieDeltaStreamer.stopLeakedStreamer(ds, dsFuture,
FAST_STOP_TIMEOUT_SECS),
+ "the stop should be bounded, not wait out the ten minutes the mock
sleeps");
+
+ assertTrue(dsFuture.isCancelled(),
+ "the ingest task should have been cancelled, since that leak is what
the helper exists to close");
+ } finally {
+ ingest.shutdownNow();
+ }
+ }
+
+ /**
+ * A stop that sleeps through the interrupt, as {@code
HoodieAsyncService.shutdown(false)} does, so the stopper
+ * thread outlives the close-wait until the test releases it.
+ */
+ private static void stubStopThatOutlivesTheInterrupt(HoodieDeltaStreamer ds,
CountDownLatch release) {
+ Mockito.doAnswer(invocation -> {
+ boolean released = false;
+ while (!released) {
+ try {
+ released = release.await(10, TimeUnit.MINUTES);
+ } catch (InterruptedException swallowed) {
+ // deliberately swallowed without restoring the flag: that is the
shape being modelled
+ }
+ }
+ return null;
+ }).when(ds).shutdownGracefully();
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
index 81c8560e61d9..acebde9283a7 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
@@ -110,6 +110,7 @@ import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.config.HoodieStreamerConfig;
import org.apache.hudi.utilities.config.SourceTestConfig;
import org.apache.hudi.utilities.ingestion.HoodieIngestionException;
+import org.apache.hudi.utilities.ingestion.HoodieIngestionService;
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
import org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor;
import org.apache.hudi.utilities.schema.SchemaProvider;
@@ -187,6 +188,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -219,6 +221,11 @@ import static
org.junit.jupiter.params.provider.Arguments.arguments;
@Slf4j
public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
+ // Bounds the stop a failure triggers, so a wedged streamer cannot hang the
test it already failed. Kept
+ // well inside what the @Timeout(600) continuous-mode tests have left after
the 360s they already spend in
+ // the wait: once that budget blows, JUnit replaces the test's own failure
with its timeout.
+ private static final long STREAMER_STOP_TIMEOUT_SECS = 30;
+
// Per-field verdict for the corrupt logical-repair fixtures: relabel
ts_millis to millis and
// attach the local-timestamp logical types that 0.x dropped. ts_micros is
already micros.
private static final String LOGICAL_REPAIR_TS_OVERRIDES =
@@ -1744,22 +1751,133 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
static void deltaStreamerTestRunner(HoodieDeltaStreamer ds,
HoodieDeltaStreamer.Config cfg, Function<Boolean, Boolean> 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);
+ }
+ // The ingest task has already had its one interrupt from
cancel(true). If it swallowed that,
+ // an orderly shutdown() would never reach it and the pool thread
would outlive the fork.
+ executor.shutdownNow();
+ } else {
+ executor.shutdown();
+ }
+ }
+ }
+
+ /**
+ * Stops a streamer that a failure left running, without letting the stop
hang the test.
+ * <p>
+ * Surefire runs this module with forkCount=1 and reuseForks=true, so a live
streamer reads on into the
+ * next test, whose setup deletes basePath and whose teardown closes the
data generators underneath it.
+ * The stop has to be bounded: shutdownGracefully awaits the ingest executor
for up to 24 hours, and it
+ * returns immediately without waiting when shutdown was already requested,
so neither the wait nor the
+ * absence of one can be relied on here.
+ * <p>
+ * Each of the three waits - the stop itself, the join of the ingest task,
and the close that runs on the
+ * stopper thread after the interrupt is swallowed - is bounded by {@code
stopTimeoutSecs}, and at most two
+ * of them run in sequence on any one path (a stop that times out skips the
join; a stop that returns leaves
+ * nothing for the close-wait), so a wedged streamer holds this for at most
twice that.
+ */
+ private static void stopLeakedStreamer(HoodieDeltaStreamer ds, Future
dsFuture) {
+ stopLeakedStreamer(ds, dsFuture, STREAMER_STOP_TIMEOUT_SECS);
+ }
+
+ /** The bound is a parameter only so this helper's own tests need not spend
the production one. */
+ static void stopLeakedStreamer(HoodieDeltaStreamer ds, Future dsFuture, long
stopTimeoutSecs) {
+ 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(stopTimeoutSecs, 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(stopTimeoutSecs, 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 bound only stops this thread waiting:
HoodieAsyncService.shutdown(false) swallows the interrupt
+ // that stopper.shutdownNow() sends, and
HoodieStreamer.shutdownGracefully runs ds.close() regardless, so
+ // forcing the executor down at least interrupts the ingest round before
the close.
+ forceStopIngestion(ds);
+ if (dsFuture != null) {
+ dsFuture.cancel(true);
+ }
+ } finally {
+ stopper.shutdownNow();
+ // shutdownNow only interrupts the stopper out of awaitTermination.
HoodieAsyncService.shutdown(false)
+ // swallows that interrupt without restoring the flag, so
shutdownGracefully carries on into ds.close()
+ // on that thread. Give the close a bounded chance to finish here,
rather than let it run on into the
+ // next test's setup, which deletes basePath underneath it.
+ // An interrupted caller would make awaitTermination throw at once and
skip the wait, so the flag is
+ // cleared for the wait and restored afterwards.
+ boolean callerInterrupted = Thread.interrupted();
+ try {
+ if (!stopper.awaitTermination(stopTimeoutSecs, TimeUnit.SECONDS)) {
+ log.warn("The streamer stop did not finish closing within {}s,
letting it run on", stopTimeoutSecs);
+ }
+ } catch (InterruptedException interrupted) {
+ callerInterrupted = true;
+ } finally {
+ if (callerInterrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ }
+
+ private static void forceStopIngestion(HoodieDeltaStreamer ds) {
+ try {
+ HoodieIngestionService ingestionService = ds.getIngestionService();
+ if (ingestionService != null) {
+ ingestionService.shutdown(true);
+ }
+ } catch (Exception noService) {
+ // Nothing to force down: a streamer that never started an ingestion
service. On a real streamer
+ // getIngestionService is an Option.get(), so absence arrives as an
exception; a mock returns null
+ // instead, which the guard above covers.
+ log.debug("No ingestion service to force-stop", noService);
}
- executor.shutdown();
}
static void awaitDeltaStreamerShutdown(HoodieDeltaStreamer ds) throws
InterruptedException {
@@ -2166,6 +2284,7 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
}
@Disabled("HUDI-8951")
+ @Test
public void testHoodieIndexerExecutionAfterCommit() throws Exception {
String tableBasePath = basePath + "/asyncindexer_commit";
Set<String> customConfigs = new HashSet<>();
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java
index 15ad0e9df2b5..c4a29b884601 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerWithMultiWriter.java
@@ -184,6 +184,11 @@ public class TestHoodieDeltaStreamerWithMultiWriter
extends HoodieDeltaStreamerT
prepJobConfig.configs.add(
String.format("%s=%d", SourceTestConfig.MAX_UNIQUE_RECORDS_PROP.key(),
totalRecords));
prepJobConfig.configs.add(String.format("%s=false",
HoodieCleanConfig.AUTO_CLEAN.key()));
+ // if we don't disable small file handling, log files may never get
created and hence for MOR, compaction may not kick in.
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ prepJobConfig.configs.add(String.format("%s=3",
HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key()));
+ prepJobConfig.configs.add(String.format("%s=0",
HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key()));
+ }
HoodieDeltaStreamer prepJob = new HoodieDeltaStreamer(prepJobConfig, jsc);
// Prepare base dataset with some commits
@@ -256,6 +261,11 @@ public class TestHoodieDeltaStreamerWithMultiWriter
extends HoodieDeltaStreamerT
prepJobConfig.configs.add(
String.format("%s=%d", SourceTestConfig.MAX_UNIQUE_RECORDS_PROP.key(),
totalRecords));
prepJobConfig.configs.add(String.format("%s=false",
HoodieCleanConfig.AUTO_CLEAN.key()));
+ // if we don't disable small file handling, log files may never get
created and hence for MOR, compaction may not kick in.
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ prepJobConfig.configs.add(String.format("%s=3",
HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key()));
+ prepJobConfig.configs.add(String.format("%s=0",
HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key()));
+ }
HoodieDeltaStreamer prepJob = new HoodieDeltaStreamer(prepJobConfig, jsc);
// Prepare base dataset with some commits
@@ -406,6 +416,7 @@ public class TestHoodieDeltaStreamerWithMultiWriter extends
HoodieDeltaStreamerT
AtomicBoolean continuousFailed = new AtomicBoolean(false);
AtomicBoolean backfillFailed = new AtomicBoolean(false);
+ AtomicBoolean prerequisiteHeld = new AtomicBoolean(false);
try {
Future regularIngestionJobFuture = service.submit(() -> {
try {
@@ -420,7 +431,7 @@ public class TestHoodieDeltaStreamerWithMultiWriter extends
HoodieDeltaStreamerT
try {
// trigger backfill at least after 1 requested entry is added to
timeline from continuous job. If not, there is a chance that backfill will
complete even before
// continuous job starts.
- awaitCondition(new GetCommitsAfterInstant(tableBasePath,
lastSuccessfulCommit));
+ prerequisiteHeld.set(awaitCondition(new
GetCommitsAfterInstant(tableBasePath, lastSuccessfulCommit)));
backfillJob.sync();
} catch (Throwable ex) {
log.error("Backfilling job failed {}", ex.getMessage());
@@ -431,6 +442,10 @@ public class TestHoodieDeltaStreamerWithMultiWriter
extends HoodieDeltaStreamerT
backfillJobFuture.get();
regularIngestionJobFuture.get();
if (expectConflict) {
+ Assertions.assertTrue(prerequisiteHeld.get(),
+ "The backfill job started before the ingestion job committed
anything after " + lastSuccessfulCommit
+ + ", so the two jobs never overlapped and no conflict could be
raised. This is a test-side "
+ + "prerequisite that did not hold, not a conflict-handling
failure.");
Assertions.fail("Failed to handle concurrent writes");
}
} catch (Exception e) {
@@ -483,18 +498,28 @@ public class TestHoodieDeltaStreamerWithMultiWriter
extends HoodieDeltaStreamerT
}
}
- private static void awaitCondition(GetCommitsAfterInstant callback) throws
InterruptedException {
+ /**
+ * Waits for the continuous ingestion job to place a commit after the
instant the callback was built with, the
+ * prerequisite for the backfill job to overlap with it. The return value
lets the caller tell a prerequisite
+ * that never held from a genuine conflict-handling failure.
+ *
+ * @return true if the commit landed within the budget, false if the budget
expired.
+ */
+ private static boolean awaitCondition(GetCommitsAfterInstant callback)
throws InterruptedException {
long startTime = System.currentTimeMillis();
long soFar = 0;
while (soFar <= 5000) {
if (callback.getCommitsAfterInstant() > 0) {
- break;
+ log.warn("Awaiting completed in {}", System.currentTimeMillis() -
startTime);
+ return true;
} else {
Thread.sleep(500);
soFar += 500;
}
}
- log.warn("Awaiting completed in {}", System.currentTimeMillis() -
startTime);
+ log.error("The continuous job placed no commit after {} within {} ms, so
the backfill job is about to start "
+ + "unsynchronized with it", callback.lastSuccessfulCommit,
System.currentTimeMillis() - startTime);
+ return false;
}
}
