voonhous commented on code in PR #19391:
URL: https://github.com/apache/hudi/pull/19391#discussion_r3665764905
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java:
##########
@@ -256,6 +265,7 @@ public static final class Dispatch {
private final int total;
private final AtomicInteger settled = new AtomicInteger(0);
private final AtomicBoolean aborted = new AtomicBoolean(false);
+ private final AtomicReference<Throwable> firstFailure = new
AtomicReference<>();
Review Comment:
+1
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java:
##########
@@ -313,4 +313,47 @@ void awaitAllStopsLaterWorkerWhenEarlierFutureIsSlow()
throws Exception {
+ "even while an earlier future on another worker is still
running");
}
}
+
+ /**
+ * Regression for the swallowed-failure race: awaitAll must still report the
error when
+ * its own cancel() sweep has already marked the failing task's future
CANCELLED.
+ *
+ * <p>The failing worker aborts the batch from inside its catch block, which
releases
+ * awaitAll, but its exception only reaches the FutureTask after {@code
call()} returns.
+ * {@code FutureTask.cancel(false)} succeeds on any task still in state NEW
- a task
+ * mid-unwind included - so the cancel wins the state CAS, the later {@code
setException}
+ * becomes a no-op, and {@code get()} reports CancellationException instead
of the error.
+ * awaitAll then counted it as merely cancelled and returned normally,
reporting a failed
+ * partition-DDL batch as a successful sync.
+ *
+ * <p>Deterministic where the three tests above are not: instead of hoping
the awaiting
+ * thread wins, this cancels the future explicitly - exactly what
cancelPending() does -
+ * while the Driver is parked, and only then lets it throw.
+ */
+ @Test
+ void awaitAllReportsFailureWhenFailingFutureIsCancelledMidFlight() throws
Exception {
+ HiveSyncConfig config = configWithEmptyHiveConf();
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ HiveDriverPool.DriverFactory factory = (db) -> {
+ Driver d = mock(Driver.class);
+ doAnswer(inv -> {
+ entered.countDown();
+ release.await(10, TimeUnit.SECONDS);
+ throw new RuntimeException("boom");
+ }).when(d).run(anyString());
+ return d;
+ };
+ try (HiveDriverPool pool = new HiveDriverPool(config, 1, factory)) {
+ HiveDriverPool.Dispatch dispatch =
pool.dispatchAll(Collections.singletonList("FAIL"));
+ assertTrue(entered.await(10, TimeUnit.SECONDS), "Driver must have
started the statement");
+ assertTrue(dispatch.futureAt(0).cancel(false),
+ "Sanity: a running FutureTask is still NEW, so cancel(false) must
succeed");
+ release.countDown();
+
+ HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
+ () -> pool.awaitAll(dispatch));
+ assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ }
Review Comment:
+1
--
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]
