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 db8d02cbfac1 fix(hive-sync): stop HiveDriverPool from swallowing a
failed SQL batch (#19391)
db8d02cbfac1 is described below
commit db8d02cbfac1eb45ae2cfecd9bad049083ac6d48
Author: Vova Kolmakov <[email protected]>
AuthorDate: Wed Jul 29 09:08:56 2026 +0700
fix(hive-sync): stop HiveDriverPool from swallowing a failed SQL batch
(#19391)
* fix(hive-sync): stop HiveDriverPool from swallowing a failed SQL batch
* fix(hive-sync): address review nits on HiveDriverPool
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../org/apache/hudi/hive/util/HiveDriverPool.java | 34 ++++++++++++--
.../apache/hudi/hive/util/TestHiveDriverPool.java | 54 ++++++++++++++++++++--
2 files changed, 80 insertions(+), 8 deletions(-)
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
index 5950bf442313..d9ebea7e10ef 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveDriverPool.java
@@ -42,6 +42,7 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
@@ -135,7 +136,7 @@ public class HiveDriverPool implements AutoCloseable {
worker.driver.run(sql);
}
} catch (Throwable t) {
- dispatch.abort();
+ dispatch.recordFailure(t);
throw t;
} finally {
dispatch.taskSettled();
@@ -179,7 +180,7 @@ public class HiveDriverPool implements AutoCloseable {
try {
worker.driver.run(sql);
} catch (Throwable t) {
- dispatch.abort();
+ dispatch.recordFailure(t);
throw t;
} finally {
dispatch.taskSettled();
@@ -208,7 +209,11 @@ public class HiveDriverPool implements AutoCloseable {
dispatch.awaitSettledOrAborted();
int cancelled = dispatch.cancelPending();
- Exception firstError = null;
+ // Seeded from the batch's own record rather than discovered by walking
the futures:
+ // cancelPending() above may have erased the failing task's exception. See
+ // Dispatch#recordFailure. The walk below still runs, to count outcomes
and to catch
+ // a failure that somehow never made it into the record.
+ Throwable firstError = dispatch.firstFailure();
int completed = 0;
for (Future<?> f : dispatch.futures()) {
try {
@@ -229,7 +234,11 @@ public class HiveDriverPool implements AutoCloseable {
cancelled++;
} else if (firstError == null) {
firstError = cause;
- } else {
+ } else if (ee.getCause() != firstError) {
+ // Identity check against the raw cause, not the unwrapped one: when
the failing
+ // task wins the race against cancelPending(), its future reports
the very
+ // Throwable already held in firstError, and re-logging it here
would duplicate
+ // the exception this method is about to throw.
LOG.warn("Additional SQL batch failed (suppressed in favor of first
error)", cause);
}
}
@@ -256,6 +265,7 @@ public class HiveDriverPool implements AutoCloseable {
private final int total;
private final AtomicInteger settled = new AtomicInteger(0);
private final AtomicBoolean aborted = new AtomicBoolean(false);
+ private final AtomicReference<Throwable> firstFailureRef = new
AtomicReference<>();
private final CountDownLatch done = new CountDownLatch(1);
private volatile boolean sealed;
@@ -279,11 +289,25 @@ public class HiveDriverPool implements AutoCloseable {
return aborted.get();
}
- private void abort() {
+ /**
+ * Records a task's failure and aborts the batch. The Throwable is kept
here rather
+ * than being left for {@link Future#get()} to report, because the failing
task is
+ * racing the awaiting thread: this call releases {@link
#awaitSettledOrAborted()},
+ * but the task's exception only reaches its {@code FutureTask} after
{@code call()}
+ * returns. {@link #cancelPending()} in between wins the {@code
FutureTask} state CAS
+ * (cancel succeeds on any task still NEW, which includes one mid-unwind),
turning the
+ * later {@code setException} into a no-op and the error into a
CancellationException.
+ */
+ private void recordFailure(Throwable t) {
+ firstFailureRef.compareAndSet(null, t);
aborted.set(true);
done.countDown();
}
+ private Throwable firstFailure() {
+ return firstFailureRef.get();
+ }
+
private void taskSettled() {
settled.incrementAndGet();
signalIfComplete();
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
index fd58b7023aa1..5b946447ba4d 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveDriverPool.java
@@ -40,6 +40,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
@@ -138,7 +139,8 @@ class TestHiveDriverPool {
HiveDriverPool.Dispatch futures = pool.dispatchAll(Arrays.asList("OK",
"FAIL", "OK"));
HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
() -> pool.awaitAll(futures));
- assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ assertNotNull(ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("boom"));
}
}
@@ -254,7 +256,8 @@ class TestHiveDriverPool {
HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
() -> pool.awaitAll(dispatch));
- assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ assertNotNull(ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("boom"));
assertEquals(Collections.singletonList("FAIL"), executed,
"Statements queued behind the failure must never reach the Driver");
}
@@ -307,10 +310,55 @@ class TestHiveDriverPool {
HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
() -> pool.awaitAll(dispatch));
- assertTrue(ex.getCause() != null &&
ex.getCause().getMessage().contains("boom"));
+ assertNotNull(ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("boom"));
assertFalse(executed.contains("AFTER_FAIL"),
"Statement queued behind a failure on the same worker must not be
applied, "
+ "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));
+ assertNotNull(ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("boom"));
+ }
+ }
}
