nsivabalan commented on code in PR #19033: URL: https://github.com/apache/hudi/pull/19033#discussion_r3762992921
########## hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/ParallelDispatch.java: ########## @@ -0,0 +1,272 @@ +/* + * 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.hive.util; + +import org.apache.hudi.common.util.VisibleForTesting; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Handle to one fan-out batch of partition work: the submitted futures plus the shared + * abort flag the tasks consult before running. + * + * <p>The abort flag exists because waiting on futures in <i>submission</i> order is not + * enough to stop queued work. If a later task fails quickly while an earlier one is slow, + * the awaiting thread is still parked on the earlier {@code Future.get()}, and the + * executor happily keeps starting every queued task in the meantime. By the time the + * failure is observed, most of the "not-yet-started" work has already run. + * + * <p>Two mechanisms fix that, and both are needed: + * <ul> + * <li>a {@link CountDownLatch} tripped by the <i>first</i> abort, so the awaiting + * thread wakes on failure rather than on its turn in submission order;</li> + * <li>a task-side {@link #aborted()} check on entry, because cancelling from the + * awaiting thread is inherently late — a worker can pull its next task off the + * queue at any moment.</li> + * </ul> + * + * <p>The failing task also records its own throwable as it aborts, so the reported root + * cause is the failure that stopped the batch. Picking the error by scanning futures in + * submission order would instead surface whichever failure happens to sit earliest in the + * list, which is not necessarily the one that aborted the run. + * + * <p>Shared by {@link HiveDriverPool} (Hive {@code Driver} statements) and + * {@link HiveMetaStoreClientPool} (Thrift {@code dropPartition} batches), which fan out over + * different execution models but need identical abort-on-first-error semantics. + */ +public final class ParallelDispatch { + + private final List<Future<?>> futures; + private final int total; + private final AtomicInteger settled = new AtomicInteger(0); + private final AtomicBoolean aborted = new AtomicBoolean(false); + private final AtomicReference<Throwable> abortCause = new AtomicReference<>(); + private final CountDownLatch done = new CountDownLatch(1); + private volatile boolean sealed; + + ParallelDispatch(int total) { + this.total = total; + this.futures = new ArrayList<>(total); + } + + void add(Future<?> future) { + futures.add(future); + } + + // Called once submission finishes. A task that settles before the last submit + // would otherwise see settled < total and never trip the latch, so re-check here. + void sealed() { + sealed = true; + signalIfComplete(); + } + + boolean aborted() { + return aborted.get(); + } + + void abort() { + aborted.set(true); + done.countDown(); + } + + // Records the failure that triggered the abort, first writer wins. Selecting the error + // by walking futures in submission order instead would report whichever failure sits + // earliest in the list, not the one that actually stopped the batch: a fast failure at + // index 1 aborts the run, and a slow index 0 that fails later would take its place. + void abort(Throwable cause) { + abortCause.compareAndSet(null, cause); + abort(); + } + + void taskSettled() { + settled.incrementAndGet(); + signalIfComplete(); + } + + private void signalIfComplete() { + if (sealed && settled.get() >= total) { + done.countDown(); + } + } + + void awaitSettledOrAborted() { + if (total == 0) { + return; + } + try { + done.await(); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + aborted.set(true); + } Review Comment: Confirmed and fixed in 749d6719. You traced the whole chain correctly: `aborted.set(true)` without recording the cause means `cancelPending()` marks the remaining futures CANCELLED, the drain sees only `CancellationException`s, `firstError` stays null, and `awaitAll` reports a clean batch. On the DROP path that lets `HiveSyncTool` advance the last-synced commit marker over work that was cancelled or is still running — silent under-sync, which is the worst shape for this. The fix is to route it through the existing abort-cause channel: ```java } catch (InterruptedException ie) { Thread.currentThread().interrupt(); abort(ie); } ``` `abort(Throwable)` already does `compareAndSet(null, cause)` then `abort()`, so `awaitOutcome()` seeds `firstError` from it and both `awaitAll()` methods throw. Since `HiveDriverPool` and `HiveMetaStoreClientPool` share `ParallelDispatch`, this closes the gap on both. Added `interruptedAwaitIsReportedAsAFailureNotASuccess` along the lines you suggested — blocks a dispatched action, interrupts the thread parked in `awaitAll`, and asserts it does not return normally. Verified against the pre-fix code: ``` awaitAll must not report success when the wait was interrupted; the batch's work was cancelled or is still in flight ==> expected: <false> but was: <true> ``` While I was in here I checked the other interrupt handlers in the dispatch path. The two in `close()` (`HiveDriverPool` and `HiveMetaStoreClientPool` executor shutdown) restore the flag and force `shutdownNow()`, which is right — there is no outcome there to corrupt. `HiveMetaStoreClientPool.run()` declares `throws Exception`, so an interrupt from `available.take()` propagates into `guard` and is recorded by `abort(t)` already. `awaitSettledOrAborted` was the only site with the report-success hazard. -- 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]
