This is an automated email from the ASF dual-hosted git repository.
nsivabalan 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 b2174b6f9ea3 feat(hive-sync): parallelize DROP partitions in HiveQL
sync mode (#19033)
b2174b6f9ea3 is described below
commit b2174b6f9ea30235cabf39d8269f179e9f670a87
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Wed Aug 12 07:52:59 2026 -0700
feat(hive-sync): parallelize DROP partitions in HiveQL sync mode (#19033)
HiveQL sync drops partitions one at a time on the session IMetaStoreClient,
so a sync that drops a non-trivial subset of a large table becomes a
multi-minute serial Thrift loop.
#18984 added parallel ADD/UPDATE/TOUCH via HiveDriverPool. DROP cannot
reuse
that pool: ADD/UPDATE/TOUCH go through the Hive Driver, which is
thread-bound
because SessionState is a ThreadLocal, while DROP goes through
IMetaStoreClient.dropPartition -- a plain Thrift socket with no thread
affinity. This adds the Thrift-side equivalent so the existing opt-in
batching flag covers all four partition operations.
Hive has no batch-drop primitive matching dropPartition's semantics, so
each
worker still iterates its chunk one partition at a time. Batching defines
the
unit of parallel work, not a reduction in call count; the win is N
independent
sockets running concurrently instead of one serial loop.
Changes:
* HiveQueryDDLExecutor.dropPartitionsToTable splits the partition list
into
batches of hoodie.datasource.hive_sync.batch_num and either runs them
sequentially on the session client (default) or fans them across the
pool.
* HiveMetaStoreClientPool (new): bounded ArrayBlockingQueue of
RetryingMetaStoreClient instances plus a fixed-size ExecutorService.
Pool
size equals thread count, so in-flight Thrift calls can never exceed
available clients and borrowing cannot deadlock.
* ParallelDispatch (new): fan-out coordination extracted from
HiveDriverPool.Dispatch so both pools share one abort-on-first-error
implementation. Waiting on futures in submission order is not enough on
its
own -- a fast failure on a later batch goes unobserved while the
awaiting
thread is parked on an earlier slow one, and the executor keeps starting
queued work meanwhile.
* Partition values are resolved on the calling thread before fan-out.
PartitionValueExtractor is user-pluggable with no thread-safety
contract, so
invoking it from several workers could yield a garbled clause and drop
the
wrong partition. This also halves the extractor calls, since
partitionExists
and the drop clause previously extracted separately.
Opt-in only. With hoodie.datasource.hive_sync.batching.enabled=false (the
default) the path is byte-identical to before. use_spark_catalog=true
falls
back to sequential, since that client is built reflectively and is not
compatible with the RetryingMetaStoreClient pool. No new config keys, no
storage format changes, no public API changes.
---
.../org/apache/hudi/hive/HiveSyncConfigHolder.java | 18 +-
.../org/apache/hudi/hive/HoodieHiveSyncClient.java | 119 +++++--
.../apache/hudi/hive/ddl/HiveQueryDDLExecutor.java | 94 ++++-
.../org/apache/hudi/hive/util/HiveDriverPool.java | 221 ++----------
.../hudi/hive/util/HiveMetaStoreClientPool.java | 258 ++++++++++++++
.../apache/hudi/hive/util/HivePartitionUtil.java | 22 +-
.../apache/hudi/hive/util/ParallelDispatch.java | 277 +++++++++++++++
.../org/apache/hudi/hive/TestHiveSyncTool.java | 40 +++
.../hudi/hive/TestHoodieHiveSyncClientClose.java | 18 +-
.../util/TestDropPartitionExtractorThreading.java | 196 +++++++++++
.../apache/hudi/hive/util/TestHiveDriverPool.java | 14 +-
.../hive/util/TestHiveMetaStoreClientPool.java | 385 +++++++++++++++++++++
12 files changed, 1407 insertions(+), 255 deletions(-)
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
index 14afa81ea30f..84e2f7bae0e8 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
@@ -132,18 +132,24 @@ public class HiveSyncConfigHolder {
+ "Hive Driver workers, with ADD and TOUCH additionally split into
batches of "
+ "`hoodie.datasource.hive_sync.batch_num` partitions per statement
(ADD was already batched "
+ "before this flag existed; only its dispatch becomes parallel
here). SET_LOCATION remains one "
- + "statement per partition, as Hive SQL has no multi-partition form.
DROP remains serial. "
- + "Table-level statements (create/alter table, last commit time,
writer version) continue to run "
- + "on the single session Driver. Default off; the default HiveQL
path is unchanged unless "
+ + "statement per partition, as Hive SQL has no multi-partition form.
DROP is also parallelized, "
+ + "but over a pool of metastore (Thrift) clients rather than Hive
Driver workers, since it is "
+ + "issued as dropPartition calls rather than SQL; drops are split
into batches of "
+ + "`hoodie.datasource.hive_sync.batch_num` partitions and fanned
across those clients. DROP falls "
+ + "back to sequential execution on the single session client when "
+ + "`hoodie.datasource.hive_sync.use_spark_catalog` is true, as the
Spark catalog client cannot be "
+ + "pooled. Table-level statements (create/alter table, last commit
time, writer version) continue "
+ + "to run on the single session Driver. Default off; the default
HiveQL path is unchanged unless "
+ "explicitly opted in.");
public static final ConfigProperty<Integer> HIVE_SYNC_BATCHING_THREADS =
ConfigProperty
.key("hoodie.datasource.hive_sync.batching.threads")
.defaultValue(4)
.markAdvanced()
.sinceVersion("1.3.0")
- .withDocumentation("Pool size (number of Hive Driver workers) and
worker-thread count for parallel "
- + "HiveQL partition dispatch when
`hoodie.datasource.hive_sync.batching.enabled` is true. "
- + "Ignored otherwise.");
+ .withDocumentation("Number of worker threads used for parallel HiveQL
partition dispatch when "
+ + "`hoodie.datasource.hive_sync.batching.enabled` is true. The same
value sizes both pools: the "
+ + "Hive Driver workers used for ADD/TOUCH/SET_LOCATION and the
metastore (Thrift) clients used "
+ + "for DROP. Ignored otherwise.");
public static final ConfigProperty<String> HIVE_SYNC_MODE = ConfigProperty
.key("hoodie.datasource.hive_sync.mode")
.noDefaultValue()
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
index accc60cb9066..f46bbffce638 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
@@ -38,6 +38,7 @@ import org.apache.hudi.hive.ddl.HiveSyncMode;
import org.apache.hudi.hive.ddl.JDBCBasedMetadataOperator;
import org.apache.hudi.hive.ddl.JDBCExecutor;
import org.apache.hudi.hive.util.HiveDriverPool;
+import org.apache.hudi.hive.util.HiveMetaStoreClientPool;
import org.apache.hudi.hive.util.IMetaStoreClientUtil;
import org.apache.hudi.hive.util.PartitionFilterGenerator;
import org.apache.hudi.sync.common.HoodieSyncClient;
@@ -89,6 +90,11 @@ public class HoodieHiveSyncClient extends HoodieSyncClient {
private final Map<String, Table> initialTableByName = new HashMap<>();
DDLExecutor ddlExecutor;
private IMetaStoreClient client;
+ // Present only when HIVE_SYNC_BATCHING_ENABLED and sync mode is HIVEQL.
Owned by
+ // this class; closed in close() before Hive.closeCurrent().
HiveQueryDDLExecutor
+ // uses it only for DROP (Hive Thrift, not Hive Driver) — see
HiveMetaStoreClientPool
+ // javadoc.
+ private Option<HiveMetaStoreClientPool> partitionClientPool = Option.empty();
// Present only when HIVE_SYNC_BATCHING_ENABLED and sync mode is HIVEQL
(explicit
// or legacy default). Owned by HiveQueryDDLExecutor; this field is kept for
// reference only — close() is delegated through ddlExecutor.close().
@@ -131,8 +137,7 @@ public class HoodieHiveSyncClient extends HoodieSyncClient {
ddlExecutor = new HMSDDLExecutor(config, this.client);
break;
case HIVEQL:
- this.partitionDriverPool = maybeBuildHiveDriverPool(config);
- ddlExecutor = new HiveQueryDDLExecutor(config, this.client,
this.partitionDriverPool);
+ ddlExecutor = buildHiveQueryDDLExecutor(config);
break;
case JDBC:
JDBCExecutor jdbcExecutor = new JDBCExecutor(config);
@@ -150,23 +155,23 @@ public class HoodieHiveSyncClient extends
HoodieSyncClient {
jdbcMetadataOperator = new JDBCBasedMetadataOperator(
jdbcExecutor.getConnection(), databaseName);
} else {
- this.partitionDriverPool = maybeBuildHiveDriverPool(config);
- ddlExecutor = new HiveQueryDDLExecutor(config, this.client,
this.partitionDriverPool);
+ ddlExecutor = buildHiveQueryDDLExecutor(config);
}
}
} catch (Exception e) {
- // The pool owns live daemon threads and Hive Drivers, and is built
before the
- // executor that would otherwise own its lifecycle. Any throw between
those two
- // points would leak it -- notably QueryBasedDDLExecutor's
super(config), which
- // runs the PartitionValueExtractor reflection before
HiveQueryDDLExecutor's own
- // try block is even entered. Closing here covers every such window; the
pool's
- // close() is idempotent, so overlapping with the executor's cleanup is
harmless.
- closePartitionDriverPoolQuietly();
+ // The pools own live daemon threads, Hive Drivers, and Thrift sockets,
and are
+ // built before the executor that would otherwise own their lifecycle.
Any throw
+ // between those two points would leak them -- notably
QueryBasedDDLExecutor's
+ // super(config), which runs the PartitionValueExtractor reflection
before
+ // HiveQueryDDLExecutor's own try block is even entered. Closing here
covers every
+ // such window; both close() methods are idempotent, so overlapping with
the
+ // executor's cleanup (or buildHiveQueryDDLExecutor's rollback) is
harmless.
+ closePartitionPoolsQuietly();
throw new HoodieHiveSyncException("Failed to create
HiveMetaStoreClient", e);
}
}
- private void closePartitionDriverPoolQuietly() {
+ private void closePartitionPoolsQuietly() {
partitionDriverPool.ifPresent(pool -> {
try {
pool.close();
@@ -174,6 +179,15 @@ public class HoodieHiveSyncClient extends HoodieSyncClient
{
log.warn("Error closing HiveDriverPool during failed sync client
construction", e);
}
});
+ partitionDriverPool = Option.empty();
+ partitionClientPool.ifPresent(pool -> {
+ try {
+ pool.close();
+ } catch (Exception e) {
+ log.warn("Error closing IMetaStoreClient pool during failed sync
client construction", e);
+ }
+ });
+ partitionClientPool = Option.empty();
}
/**
@@ -227,6 +241,22 @@ public class HoodieHiveSyncClient extends HoodieSyncClient
{
}
}
+ private Option<HiveMetaStoreClientPool>
maybeBuildPartitionClientPool(HiveSyncConfig config) {
+ if (!config.getBooleanOrDefault(HIVE_SYNC_BATCHING_ENABLED)) {
+ return Option.empty();
+ }
+ if (config.getBooleanOrDefault(HIVE_SYNC_USE_SPARK_CATALOG)) {
+ // The Spark catalog client is constructed via reflection against a
Spark-side
+ // class and isn't compatible with the direct RetryingMetaStoreClient
pool path.
+ // Fall back to single-client sequential behavior rather than failing
the sync.
+ log.warn("{}=true is not supported with {}=true; falling back to
sequential partition drop.",
+ HIVE_SYNC_BATCHING_ENABLED.key(), HIVE_SYNC_USE_SPARK_CATALOG.key());
+ return Option.empty();
+ }
+ int size = config.getIntOrDefault(HIVE_SYNC_BATCHING_THREADS);
+ return Option.of(new HiveMetaStoreClientPool(config, size));
+ }
+
private Option<HiveDriverPool> maybeBuildHiveDriverPool(HiveSyncConfig
config) {
if (!config.getBooleanOrDefault(HIVE_SYNC_BATCHING_ENABLED)) {
return Option.empty();
@@ -235,6 +265,25 @@ public class HoodieHiveSyncClient extends HoodieSyncClient
{
return Option.of(new HiveDriverPool(config, size));
}
+ /**
+ * Builds the (optional) partition-phase pools and the {@link
HiveQueryDDLExecutor}
+ * that uses them, rolling back whichever pool(s) already got built if a
later step
+ * in this sequence throws. Without this, a failure in {@code
maybeBuildPartitionClientPool}
+ * (after {@code partitionDriverPool} was already built) or in the
executor's own
+ * constructor would leak the already-built pool's worker threads and
Thrift/Driver
+ * connections, since this constructor's outer catch just rethrows.
+ */
+ private HiveQueryDDLExecutor buildHiveQueryDDLExecutor(HiveSyncConfig
config) {
+ try {
+ this.partitionDriverPool = maybeBuildHiveDriverPool(config);
+ this.partitionClientPool = maybeBuildPartitionClientPool(config);
+ return new HiveQueryDDLExecutor(config, this.client,
this.partitionDriverPool, this.partitionClientPool);
+ } catch (Exception e) {
+ closePartitionPoolsQuietly();
+ throw e;
+ }
+ }
+
private Table getInitialTable(String table) {
return initialTableByName.computeIfAbsent(table, t -> {
try {
@@ -630,22 +679,38 @@ public class HoodieHiveSyncClient extends
HoodieSyncClient {
@Override
public void close() {
try {
- ddlExecutor.close();
- if (client != null) {
- // Close the proxied IMetaStoreClient directly before
Hive.closeCurrent().
- // When RetryingMetaStoreClient rebuilds the underlying client on a
transient
- // TException, the fresh MSC is reachable only through this proxy,
while the
- // thread-local Hive singleton still references the older instance. So
- // Hive.closeCurrent() alone closes the stale MSC and orphans the
retry-created
- // one, leaking a connection per sync cycle. client.close() releases
the live
- // MSC by identity; Hive.closeCurrent() remains a fallback for the
singleton path.
- try {
- client.close();
- } catch (Exception e) {
- log.warn("Failed to close IMetaStoreClient directly;
Hive.closeCurrent() will run anyway", e);
+ try {
+ ddlExecutor.close();
+ } finally {
+ // Close the partition client pool before Hive.closeCurrent() so the
+ // RetryingMetaStoreClient instances held by the pool release their
Thrift
+ // sockets without racing the ThreadLocal Hive cleanup. Runs even if
+ // ddlExecutor.close() above threw, so the pool's Thrift sockets and
+ // worker threads aren't leaked.
+ if (partitionClientPool.isPresent()) {
+ try {
+ partitionClientPool.get().close();
+ } catch (Exception e) {
+ log.warn("Error closing IMetaStoreClient pool", e);
+ }
+ partitionClientPool = Option.empty();
+ }
+ if (client != null) {
+ // Close the proxied IMetaStoreClient directly before
Hive.closeCurrent().
+ // When RetryingMetaStoreClient rebuilds the underlying client on a
transient
+ // TException, the fresh MSC is reachable only through this proxy,
while the
+ // thread-local Hive singleton still references the older instance.
So
+ // Hive.closeCurrent() alone closes the stale MSC and orphans the
retry-created
+ // one, leaking a connection per sync cycle. client.close() releases
the live
+ // MSC by identity; Hive.closeCurrent() remains a fallback for the
singleton path.
+ try {
+ client.close();
+ } catch (Exception e) {
+ log.warn("Failed to close IMetaStoreClient directly;
Hive.closeCurrent() will run anyway", e);
+ }
+ Hive.closeCurrent();
+ client = null;
}
- Hive.closeCurrent();
- client = null;
}
} catch (Exception e) {
log.error("Could not close connection ", e);
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
index c853313182ed..5447e1fd922f 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java
@@ -18,11 +18,13 @@
package org.apache.hudi.hive.ddl;
+import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.hive.HiveSyncConfig;
import org.apache.hudi.hive.HoodieHiveSyncException;
import org.apache.hudi.hive.util.HiveDriverPool;
+import org.apache.hudi.hive.util.HiveMetaStoreClientPool;
import org.apache.hudi.hive.util.HivePartitionUtil;
import lombok.extern.slf4j.Slf4j;
@@ -59,16 +61,22 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
// (createTable, schema evolution, single-statement runSQL callers) always
uses the
// session `hiveDriver` above. See HiveDriverPool javadoc.
private final Option<HiveDriverPool> driverPool;
+ // When present, dropPartitionsToTable fans batches across this Thrift
client pool.
+ // Owned by HoodieHiveSyncClient; close() is delegated through there. See
+ // HiveMetaStoreClientPool javadoc for the usage contract (partition-row ops
only).
+ private final Option<HiveMetaStoreClientPool> metaStoreClientPool;
public HiveQueryDDLExecutor(HiveSyncConfig config, IMetaStoreClient
metaStoreClient) {
- this(config, metaStoreClient, Option.empty());
+ this(config, metaStoreClient, Option.empty(), Option.empty());
}
public HiveQueryDDLExecutor(HiveSyncConfig config, IMetaStoreClient
metaStoreClient,
- Option<HiveDriverPool> driverPool) {
+ Option<HiveDriverPool> driverPool,
+ Option<HiveMetaStoreClientPool>
metaStoreClientPool) {
super(config);
this.metaStoreClient = metaStoreClient;
this.driverPool = driverPool;
+ this.metaStoreClientPool = metaStoreClientPool;
try {
this.sessionState = new SessionState(config.getHiveConf(),
UserGroupInformation.getCurrentUser().getShortUserName());
@@ -209,21 +217,87 @@ public class HiveQueryDDLExecutor extends
QueryBasedDDLExecutor {
log.info("Drop partitions {} on {}", partitionsToDrop.size(), tableName);
try {
- for (String dropPartition : partitionsToDrop) {
- if (HivePartitionUtil.partitionExists(metaStoreClient, tableName,
dropPartition, partitionValueExtractor,
- config)) {
- String partitionClause =
- HivePartitionUtil.getPartitionClauseForDrop(dropPartition,
partitionValueExtractor, config);
- metaStoreClient.dropPartition(databaseName, tableName,
partitionClause, false);
- }
- log.info("Drop partition {} on {}", dropPartition, tableName);
+ // Resolved here, on the calling thread, rather than inside the workers:
this is the
+ // only sync path that would otherwise call a user-supplied
PartitionValueExtractor
+ // from several threads at once. Extractors are pluggable and not
required to be
+ // thread-safe, and a garbled clause would drop the wrong partition. It
also halves
+ // the extractor calls, since partitionExists and the drop clause share
the values.
+ List<PartitionToDrop> resolved = new
ArrayList<>(partitionsToDrop.size());
+ for (String partition : partitionsToDrop) {
+ List<String> values =
partitionValueExtractor.extractPartitionValuesInPath(partition);
+ resolved.add(new PartitionToDrop(partition, values,
+ HivePartitionUtil.getPartitionClauseForDrop(values, config)));
}
+
+ int batchSyncPartitionNum =
config.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM);
+ List<List<PartitionToDrop>> batches = CollectionUtils.batches(resolved,
batchSyncPartitionNum);
+ runDropBatches(tableName, batches);
} catch (Exception e) {
log.error("{} drop partition failed", tableId(databaseName, tableName),
e);
throw new HoodieHiveSyncException(tableId(databaseName, tableName) + "
drop partition failed", e);
}
}
+ /**
+ * Drops partitions one batch at a time. When {@link #metaStoreClientPool}
is present,
+ * batches fan out across the pool's worker threads (each borrowing an
independent
+ * IMetaStoreClient); otherwise batches are dispatched sequentially against
the
+ * session client. Hive has no batch-drop primitive that matches
dropPartition's
+ * semantics, so each worker still iterates its chunk one partition at a
time — the
+ * win is fanning chunks across independent Thrift clients.
+ *
+ * <p>First-error semantics come from {@link ParallelDispatch}, shared with
+ * {@code HiveDriverPool}: the first failure is rethrown, batches that have
not started
+ * are stopped via the task-side abort flag, and later failures are logged
at WARN.
+ */
+ private void runDropBatches(String tableName, List<List<PartitionToDrop>>
batches) throws Exception {
+ if (!metaStoreClientPool.isPresent()) {
+ for (List<PartitionToDrop> batch : batches) {
+ applyDropBatch(metaStoreClient, tableName, batch);
+ }
+ return;
+ }
+ HiveMetaStoreClientPool pool = metaStoreClientPool.get();
+ pool.awaitAll(
+ pool.dispatchAll(batches, (client, batch) -> applyDropBatch(client,
tableName, batch)),
+ "drop partition");
+ }
+
+ private void applyDropBatch(IMetaStoreClient client, String tableName,
List<PartitionToDrop> batch) throws Exception {
+ int dropped = 0;
+ for (PartitionToDrop dropPartition : batch) {
+ if (HivePartitionUtil.partitionExists(client, tableName,
dropPartition.path,
+ dropPartition.values, config)) {
+ client.dropPartition(databaseName, tableName, dropPartition.clause,
false);
+ dropped++;
+ }
+ // Per-partition detail stays at debug: a batch can hold thousands of
partitions
+ // and N workers log concurrently, so INFO carries the per-batch summary
instead.
+ log.debug("Dropped partition {} on {}", dropPartition.path, tableName);
+ }
+ log.info("Dropped {} of {} partitions in batch on {}", dropped,
batch.size(), tableName);
+ }
+
+ /**
+ * A partition to drop with its extractor-derived values already resolved,
so worker
+ * threads never touch the shared {@link PartitionValueExtractor}.
Immutable: the
+ * {@code values} list is copied and wrapped unmodifiable at construction.
+ */
+ private static final class PartitionToDrop {
+ private final String path;
+ private final List<String> values;
+ private final String clause;
+
+ private PartitionToDrop(String path, List<String> values, String clause) {
+ this.path = path;
+ // Copied, not just wrapped: PartitionValueExtractor may hand back a
buffer it reuses
+ // across calls, and unmodifiableList would leave every entry aliasing
the last
+ // extraction. partitionExists would then check the wrong partition and
skip drops.
+ this.values = Collections.unmodifiableList(new ArrayList<>(values));
+ this.clause = clause;
+ }
+ }
+
@Override
public void close() {
// Close the pool first so the worker threads stop dispatching against
their
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 d9ebea7e10ef..b9f4db106ac1 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
@@ -18,7 +18,6 @@
package org.apache.hudi.hive.util;
-import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.hive.HiveSyncConfig;
import org.apache.hudi.hive.HoodieHiveSyncException;
@@ -32,17 +31,11 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.CancellationException;
-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.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;
@@ -125,24 +118,14 @@ public class HiveDriverPool implements AutoCloseable {
if (setupSqls.isEmpty()) {
return;
}
- Dispatch dispatch = new Dispatch(workers.size());
+ ParallelDispatch dispatch = new ParallelDispatch(workers.size());
for (Worker worker : workers) {
- dispatch.add(worker.executor.submit(() -> {
- if (dispatch.aborted()) {
- throw new CancellationException("Skipped after an earlier setup
statement failed");
- }
- try {
- for (String sql : setupSqls) {
- worker.driver.run(sql);
- }
- } catch (Throwable t) {
- dispatch.recordFailure(t);
- throw t;
- } finally {
- dispatch.taskSettled();
+ dispatch.add(worker.executor.submit(dispatch.guard(() -> {
+ for (String sql : setupSqls) {
+ worker.driver.run(sql);
}
return null;
- }));
+ }, "Skipped after an earlier setup statement failed")));
}
dispatch.sealed();
awaitAll(dispatch);
@@ -151,42 +134,28 @@ public class HiveDriverPool implements AutoCloseable {
/**
* Dispatches each SQL string to a worker (round-robin) and returns a handle
to the
* in-flight batch — this method does not block. The caller is responsible
for
- * awaiting completion via {@link #awaitAll(Dispatch)} and collecting
errors. SQL text
+ * awaiting completion via {@link #awaitAll(ParallelDispatch)} and
collecting errors. SQL text
* is intentionally not logged per-statement here: batched TOUCH/ADD
statements can
* be many kilobytes, and N parallel workers would multiply the log volume.
See
- * {@link #awaitAll(Dispatch)} for the per-call summary log.
+ * {@link #awaitAll(ParallelDispatch)} for the per-call summary log.
*
* <p>Statements are spread round-robin across workers, so worker <i>w</i>
owns
* indices {@code w, w + N, w + 2N, ...}. Each worker drains its own queue
* independently, which is why abort has to be observed by the tasks
themselves
- * rather than by the awaiting thread — see {@link Dispatch}.
+ * rather than by the awaiting thread — see {@link ParallelDispatch}.
*/
- public Dispatch dispatchAll(List<String> sqls) {
+ public ParallelDispatch dispatchAll(List<String> sqls) {
if (closed) {
throw new IllegalStateException("Cannot dispatch to a closed
HiveDriverPool");
}
- Dispatch dispatch = new Dispatch(sqls.size());
+ ParallelDispatch dispatch = new ParallelDispatch(sqls.size());
for (int i = 0; i < sqls.size(); i++) {
String sql = sqls.get(i);
Worker worker = workers.get(i % workers.size());
- dispatch.add(worker.executor.submit(() -> {
- // Abort check inside the task: a worker can dequeue this statement
while the
- // awaiting thread is still parked on some other worker's slower
statement, so
- // Future.cancel() alone cannot stop it in time. Checking here means no
- // statement starts after a sibling has already failed.
- if (dispatch.aborted()) {
- throw new CancellationException("Skipped after an earlier statement
failed");
- }
- try {
- worker.driver.run(sql);
- } catch (Throwable t) {
- dispatch.recordFailure(t);
- throw t;
- } finally {
- dispatch.taskSettled();
- }
+ dispatch.add(worker.executor.submit(dispatch.guard(() -> {
+ worker.driver.run(sql);
return null;
- }));
+ }, "Skipped after an earlier statement failed")));
}
dispatch.sealed();
return dispatch;
@@ -201,167 +170,17 @@ public class HiveDriverPool implements AutoCloseable {
* Callers do not need per-statement results (Hive's Driver.run side-effects
the
* metastore), so this method is void.
*/
- public void awaitAll(Dispatch dispatch) {
+ public void awaitAll(ParallelDispatch dispatch) {
long start = System.currentTimeMillis();
- // Block until either every task settled or one of them aborted the batch.
Only
- // then walk the futures — by that point no un-started task can still
begin, so
- // the walk order no longer affects how much extra DDL gets applied.
- dispatch.awaitSettledOrAborted();
- int cancelled = dispatch.cancelPending();
+ ParallelDispatch.Outcome outcome = dispatch.awaitOutcome();
+ outcome.suppressed().forEach(e ->
+ LOG.warn("Additional SQL batch failed (suppressed in favor of first
error)", e));
- // 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 {
- f.get();
- completed++;
- } catch (CancellationException ce) {
- // Either we cancelled it before it started, or the task itself
observed the
- // abort flag and bailed. Not a new failure; just note it for the
summary.
- cancelled++;
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- if (firstError == null) {
- firstError = ie;
- }
- } catch (ExecutionException ee) {
- Exception cause = unwrap(ee);
- if (cause instanceof CancellationException) {
- cancelled++;
- } else if (firstError == null) {
- firstError = cause;
- } 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);
- }
- }
- }
- if (firstError != null) {
- throw new HoodieHiveSyncException("Failed in executing SQL", firstError);
+ if (outcome.failed()) {
+ throw new HoodieHiveSyncException("Failed in executing SQL",
outcome.firstError());
}
LOG.info("Completed {} SQL statements ({} cancelled) in {} ms across {}
workers",
- completed, cancelled, System.currentTimeMillis() - start, size);
- }
-
- /**
- * Handle to one {@link #dispatchAll(List)} batch: the submitted futures
plus the
- * shared abort flag the tasks consult before running.
- *
- * <p>The flag exists because the futures belong to N independent
single-thread
- * executors. Cancelling from the awaiting thread is inherently late — a
worker can
- * pull its next statement off its own queue at any moment — so each task
also
- * re-checks {@link #aborted()} on entry. That is what actually bounds how
much extra
- * partition DDL a failed sync can apply.
- */
- public static final class Dispatch {
- 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> firstFailureRef = new
AtomicReference<>();
- private final CountDownLatch done = new CountDownLatch(1);
- private volatile boolean sealed;
-
- private Dispatch(int total) {
- this.total = total;
- this.futures = new ArrayList<>(total);
- }
-
- private 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.
- private void sealed() {
- sealed = true;
- signalIfComplete();
- }
-
- private boolean aborted() {
- return aborted.get();
- }
-
- /**
- * 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();
- }
-
- private void signalIfComplete() {
- if (sealed && settled.get() >= total) {
- done.countDown();
- }
- }
-
- private void awaitSettledOrAborted() {
- if (total == 0) {
- return;
- }
- try {
- done.await();
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- aborted.set(true);
- }
- }
-
- // mayInterruptIfRunning=false: the worker thread is bound to a Hive
Driver whose
- // state we don't want to corrupt mid-statement. Cancel only tasks that
haven't
- // started; in-flight statements run to completion.
- private int cancelPending() {
- int cancelled = 0;
- for (Future<?> f : futures) {
- if (f.cancel(false)) {
- cancelled++;
- }
- }
- return cancelled;
- }
-
- private List<Future<?>> futures() {
- return futures;
- }
-
- @VisibleForTesting
- public int size() {
- return futures.size();
- }
-
- @VisibleForTesting
- public Future<?> futureAt(int index) {
- return futures.get(index);
- }
- }
-
- private static Exception unwrap(ExecutionException ee) {
- Throwable cause = ee.getCause();
- return (cause instanceof Exception) ? (Exception) cause : ee;
+ outcome.completed(), outcome.cancelled(), System.currentTimeMillis() -
start, size);
}
public int size() {
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveMetaStoreClientPool.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveMetaStoreClientPool.java
new file mode 100644
index 000000000000..7b2e04b348a0
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveMetaStoreClientPool.java
@@ -0,0 +1,258 @@
+/*
+ * 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.exception.HoodieException;
+import org.apache.hudi.hive.HiveSyncConfig;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.RetryingMetaStoreClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Pool of {@link IMetaStoreClient} instances for parallel partition sync.
+ *
+ * <p>Each pooled client wraps an independent Thrift connection to the Hive
Metastore.
+ * Callers borrow a client via {@link #run(ClientAction)}, which blocks until
a client
+ * is available, executes the action, and returns the client to the pool.
Batches are
+ * fanned out via {@link #dispatchAll(List, ClientConsumer)}, which submits to
an internal
+ * worker pool sized to match the clients, so in-flight Thrift calls can never
exceed the
+ * number of available clients.
+ *
+ * <p><b>Usage contract:</b> pool clients must be used <i>only</i> for
partition-row
+ * operations — {@code add_partitions}, {@code alter_partitions}, {@code
dropPartition},
+ * {@code getPartition}. Table-row operations ({@code createTable}, {@code
alter_table},
+ * {@code getTable} used as the read half of a read-modify-write of table
parameters)
+ * must continue to go through the session client held by
+ * {@code HoodieHiveSyncClient.client} on the sync driver thread. Mixing the
two would
+ * risk lost updates on table parameters such as the last-commit-time-synced
marker.
+ *
+ * <p>The pool is gated behind {@code
hoodie.datasource.hive_sync.batching.enabled} and
+ * is constructed for sync mode HIVEQL, where it backs the DROP path only.
DROP goes
+ * through {@code IMetaStoreClient.dropPartition} (Thrift), whereas
ADD/UPDATE/TOUCH go
+ * through the thread-bound Hive {@code Driver} and use {@code HiveDriverPool}
instead.
+ */
+public class HiveMetaStoreClientPool implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(HiveMetaStoreClientPool.class);
+
+ private final ArrayBlockingQueue<IMetaStoreClient> available;
+ private final List<IMetaStoreClient> all;
+ private final ExecutorService executor;
+ private final int size;
+ private volatile boolean closed;
+
+ public HiveMetaStoreClientPool(HiveSyncConfig config, int size) {
+ this(buildClients(config, size), size);
+ }
+
+ // Package-private for tests: accepts a pre-built list of clients so we can
+ // exercise borrow/return/close semantics without a live metastore.
+ HiveMetaStoreClientPool(List<IMetaStoreClient> clients, int size) {
+ if (size < 1) {
+ throw new IllegalArgumentException("Pool size must be >= 1, got " +
size);
+ }
+ if (clients.size() != size) {
+ throw new IllegalArgumentException("Expected " + size + " clients, got "
+ clients.size());
+ }
+ this.size = size;
+ this.available = new ArrayBlockingQueue<>(size);
+ this.all = new ArrayList<>(clients);
+ this.available.addAll(clients);
+ this.executor = Executors.newFixedThreadPool(size, new
PoolThreadFactory());
+ LOG.info("Initialized IMetaStoreClient pool with {} clients", size);
+ }
+
+ private static List<IMetaStoreClient> buildClients(HiveSyncConfig config,
int size) {
+ // Duplicated with the constructor deliberately: this runs first (the
public
+ // constructor evaluates buildClients before delegating), so it both fails
before any
+ // Thrift connection is opened and keeps the message meaningful for a
negative size,
+ // which would otherwise surface as ArrayList's "Illegal Capacity".
+ if (size < 1) {
+ throw new IllegalArgumentException("Pool size must be >= 1, got " +
size);
+ }
+ HiveConf hiveConf = config.getHiveConf();
+ List<IMetaStoreClient> clients = new ArrayList<>(size);
+ try {
+ for (int i = 0; i < size; i++) {
+ clients.add(newClient(hiveConf));
+ }
+ return clients;
+ } catch (Exception e) {
+ // Construction failed mid-way; close any clients we already built before
+ // surfacing the error so we don't leak Thrift sockets.
+ for (IMetaStoreClient c : clients) {
+ try {
+ c.close();
+ } catch (Exception ignore) {
+ // intentional: best-effort cleanup during failure
+ }
+ }
+ throw new HoodieException("Failed to construct IMetaStoreClient pool of
size " + size, e);
+ }
+ }
+
+ private static IMetaStoreClient newClient(HiveConf hiveConf) {
+ try {
+ // RetryingMetaStoreClient.getProxy returns an independent
IMetaStoreClient
+ // (one Thrift socket per call), bypassing the Hive thread-local cache
that
+ // Hive.get(conf) would use. This is what gives us N truly independent
clients.
+ return RetryingMetaStoreClient.getProxy(hiveConf, true);
+ } catch (Exception e) {
+ throw new HoodieException("Failed to create IMetaStoreClient for pool",
e);
+ }
+ }
+
+ /**
+ * Borrows a client, runs the action, and returns the client to the pool.
+ * Blocks if all clients are in use until one becomes available.
+ */
+ public <T> T run(ClientAction<T> action) throws Exception {
+ if (closed) {
+ throw new IllegalStateException("Cannot borrow from a closed
IMetaStoreClient pool");
+ }
+ IMetaStoreClient client = available.take();
+ try {
+ return action.apply(client);
+ } finally {
+ // Always return the client to the pool, even on failure. Thrift clients
+ // recover transparently from transactional errors at the HMS layer;
+ // RetryingMetaStoreClient handles transient socket failures internally.
+ if (!closed) {
+ available.offer(client);
+ }
+ }
+ }
+
+ /**
+ * Submits one task per item, each borrowing a pooled client for the
duration of its
+ * call, and returns a handle to the in-flight batch — this method does not
block. The
+ * caller awaits completion via {@link #awaitAll(ParallelDispatch, String)}.
+ *
+ * <p>Tasks observe a shared abort flag, so a failure on any item stops
items that have
+ * not started yet even while a slower sibling is still mid-call. See
+ * {@link ParallelDispatch} for why waiting on futures alone does not
achieve that.
+ */
+ public <T> ParallelDispatch dispatchAll(List<T> items, ClientConsumer<T>
action) {
+ if (closed) {
+ throw new IllegalStateException("Cannot dispatch to a closed
IMetaStoreClient pool");
+ }
+ ParallelDispatch dispatch = new ParallelDispatch(items.size());
+ for (T item : items) {
+ dispatch.add(executor.submit(dispatch.guard(() -> {
+ run(client -> {
+ action.accept(client, item);
+ return null;
+ });
+ return null;
+ }, "Skipped after an earlier batch failed")));
+ }
+ dispatch.sealed();
+ return dispatch;
+ }
+
+ /**
+ * Awaits a batch from {@link #dispatchAll(List, ClientConsumer)}, cancels
whatever had
+ * not started, and rethrows the first real failure. Later failures are
logged at WARN.
+ */
+ public void awaitAll(ParallelDispatch dispatch, String description) throws
Exception {
+ long start = System.currentTimeMillis();
+ ParallelDispatch.Outcome outcome = dispatch.awaitOutcome();
+ outcome.suppressed().forEach(e ->
+ LOG.warn("Additional {} batch failed (suppressed in favor of first
error)", description, e));
+
+ if (outcome.failed()) {
+ LOG.error("{} dispatch aborted after first failure ({} batches
cancelled)",
+ description, outcome.cancelled());
+ throw outcome.firstError();
+ }
+ LOG.info("Completed {} {} batches ({} cancelled) in {} ms across {}
clients",
+ outcome.completed(), description, outcome.cancelled(),
+ System.currentTimeMillis() - start, size);
+ }
+
+ public int size() {
+ return size;
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException ie) {
+ executor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ closeQuietly();
+ }
+
+ private void closeQuietly() {
+ for (IMetaStoreClient client : all) {
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing pooled IMetaStoreClient", e);
+ }
+ }
+ available.clear();
+ all.clear();
+ }
+
+ @FunctionalInterface
+ public interface ClientAction<T> {
+ T apply(IMetaStoreClient client) throws Exception;
+ }
+
+ /** Work applied to one fanned-out item using a borrowed client. */
+ @FunctionalInterface
+ public interface ClientConsumer<T> {
+ void accept(IMetaStoreClient client, T item) throws Exception;
+ }
+
+ private static final class PoolThreadFactory implements ThreadFactory {
+ private static final AtomicInteger POOL_ID = new AtomicInteger(0);
+ private final AtomicInteger threadId = new AtomicInteger(0);
+ private final String namePrefix = "hudi-hive-sync-pool-" +
POOL_ID.incrementAndGet() + "-";
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r, namePrefix + threadId.incrementAndGet());
+ t.setDaemon(true);
+ return t;
+ }
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HivePartitionUtil.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HivePartitionUtil.java
index 3e75582266df..4bba1cf2365e 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HivePartitionUtil.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HivePartitionUtil.java
@@ -44,7 +44,16 @@ public class HivePartitionUtil {
* Build String, example as year=2021/month=06/day=25
*/
public static String getPartitionClauseForDrop(String partition,
PartitionValueExtractor partitionValueExtractor, HiveSyncConfig config) {
- List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
+ return
getPartitionClauseForDrop(partitionValueExtractor.extractPartitionValuesInPath(partition),
config);
+ }
+
+ /**
+ * Variant taking values already extracted by the caller, for paths that
must not invoke
+ * a {@link PartitionValueExtractor} themselves — see
+ * {@code HiveQueryDDLExecutor#dropPartitionsToTable}, which extracts on the
calling
+ * thread so a user-supplied extractor is never shared across pool workers.
+ */
+ public static String getPartitionClauseForDrop(List<String> partitionValues,
HiveSyncConfig config) {
ValidationUtils.checkArgument(config.getSplitStrings(META_SYNC_PARTITION_FIELDS).size()
== partitionValues.size(),
"Partition key parts " +
config.getSplitStrings(META_SYNC_PARTITION_FIELDS) + " does not match with
partition values " + partitionValues
+ ". Check partition strategy. ");
@@ -64,9 +73,18 @@ public class HivePartitionUtil {
public static Boolean partitionExists(IMetaStoreClient client, String
tableName, String partitionPath,
PartitionValueExtractor
partitionValueExtractor, HiveSyncConfig config) {
+ return partitionExists(client, tableName, partitionPath,
+ partitionValueExtractor.extractPartitionValuesInPath(partitionPath),
config);
+ }
+
+ /**
+ * Variant taking values already extracted by the caller. {@code
partitionPath} is
+ * retained only for error reporting.
+ */
+ public static Boolean partitionExists(IMetaStoreClient client, String
tableName, String partitionPath,
+ List<String> partitionValues,
HiveSyncConfig config) {
Partition newPartition;
try {
- List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partitionPath);
newPartition =
client.getPartition(config.getStringOrDefault(META_SYNC_DATABASE_NAME),
tableName, partitionValues);
} catch (NoSuchObjectException ignored) {
newPartition = null;
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/ParallelDispatch.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/ParallelDispatch.java
new file mode 100644
index 000000000000..1d101ae62f14
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/ParallelDispatch.java
@@ -0,0 +1,277 @@
+/*
+ * 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();
+ // Recorded as the abort cause, not just flagged: cancelPending() below
can mark
+ // every remaining future CANCELLED, so the drain in awaitOutcome()
would see only
+ // CancellationExceptions, leave firstError null, and report the batch
as a success.
+ // A caller that then advances the last-synced commit marker would be
recording work
+ // that was cancelled or is still in flight.
+ abort(ie);
+ }
+ }
+
+ // mayInterruptIfRunning=false: a worker may be mid-statement against a Hive
Driver or
+ // a Thrift client, and we don't want to tear that down partway. Cancel only
tasks that
+ // haven't started; in-flight work runs to completion.
+ int cancelPending() {
+ int cancelled = 0;
+ for (Future<?> f : futures) {
+ if (f.cancel(false)) {
+ cancelled++;
+ }
+ }
+ return cancelled;
+ }
+
+ List<Future<?>> futures() {
+ return futures;
+ }
+
+ /**
+ * Wraps {@code body} so it observes this batch's abort flag: it skips
itself if a
+ * sibling has already failed, trips the flag if it fails, and always
records that it
+ * settled so the awaiting thread can be released.
+ */
+ Callable<Void> guard(Callable<Void> body, String skipMessage) {
+ return () -> {
+ if (aborted()) {
+ throw new CancellationException(skipMessage);
+ }
+ try {
+ return body.call();
+ } catch (Throwable t) {
+ abort(t);
+ throw t;
+ } finally {
+ taskSettled();
+ }
+ };
+ }
+
+ /**
+ * Waits for the batch to settle (or abort), cancels whatever had not
started, and
+ * returns the outcome. Errors are observed in <i>completion</i> order, not
submission
+ * order, so a failure on a fast worker stops the other queues even while a
slow worker
+ * is still mid-statement.
+ */
+ Outcome awaitOutcome() {
+ awaitSettledOrAborted();
+ int cancelled = cancelPending();
+
+ // Seeded from the task that tripped the abort, so the reported root cause
is the
+ // failure that actually stopped the batch rather than the lowest-indexed
one.
+ Throwable abortedBy = abortCause.get();
+ Exception firstError = asException(abortedBy);
+ int completed = 0;
+ List<Exception> suppressed = new ArrayList<>();
+ for (Future<?> f : futures) {
+ try {
+ f.get();
+ completed++;
+ } catch (CancellationException ce) {
+ // Either we cancelled it before it started, or the task itself
observed the
+ // abort flag and bailed. Not a new failure; just note it for the
summary.
+ cancelled++;
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ if (firstError == null) {
+ firstError = ie;
+ }
+ } catch (ExecutionException ee) {
+ Exception cause = unwrap(ee);
+ if (cause instanceof CancellationException) {
+ cancelled++;
+ } else if (firstError == null) {
+ firstError = cause;
+ } else if (ee.getCause() != abortedBy) {
+ // Identity against the *raw* cause, not the unwrapped one: the
aborting task's
+ // own failure comes back through here too and must not land in its
own
+ // suppressed list. Comparing to abortedBy rather than firstError
matters when
+ // the task threw an Error — asException() wraps that in a new
RuntimeException,
+ // so comparing against firstError would never match and would
report the one
+ // real failure twice.
+ suppressed.add(cause);
+ }
+ }
+ }
+ return new Outcome(firstError, completed, cancelled, suppressed);
+ }
+
+ private static Exception asException(Throwable t) {
+ if (t == null) {
+ return null;
+ }
+ return (t instanceof Exception) ? (Exception) t : new RuntimeException(t);
+ }
+
+ private static Exception unwrap(ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ return (cause instanceof Exception) ? (Exception) cause : ee;
+ }
+
+ @VisibleForTesting
+ public int size() {
+ return futures.size();
+ }
+
+ @VisibleForTesting
+ public Future<?> futureAt(int index) {
+ return futures.get(index);
+ }
+
+ /** Result of awaiting a batch: the first real failure, if any, plus counts
for logging. */
+ static final class Outcome {
+ private final Exception firstError;
+ private final int completed;
+ private final int cancelled;
+ private final List<Exception> suppressed;
+
+ private Outcome(Exception firstError, int completed, int cancelled,
List<Exception> suppressed) {
+ this.firstError = firstError;
+ this.completed = completed;
+ this.cancelled = cancelled;
+ this.suppressed = suppressed;
+ }
+
+ Exception firstError() {
+ return firstError;
+ }
+
+ boolean failed() {
+ return firstError != null;
+ }
+
+ int completed() {
+ return completed;
+ }
+
+ int cancelled() {
+ return cancelled;
+ }
+
+ List<Exception> suppressed() {
+ return suppressed;
+ }
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index baf90079ada7..15029094ef75 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -403,6 +403,46 @@ public class TestHiveSyncTool {
"Incremental add via parallel HiveQL batching should sync the new
partitions");
}
+ /**
+ * Exercises the DROP path in HiveQL mode with batching on. DROP routes
through
+ * IMetaStoreClient.dropPartition (Thrift, not Hive Driver), so when
batching is
+ * enabled it fans out across HiveMetaStoreClientPool. Verifies the
partition set
+ * shrinks as expected when batches drop in parallel.
+ */
+ @Test
+ public void testHiveQLDropPartitionsWithBatching() throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(),
HiveSyncMode.HIVEQL.name());
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_ENABLED.key(), "true");
+ hiveSyncProps.setProperty(HIVE_SYNC_BATCHING_THREADS.key(), "3");
+ // Small batch_num so we get multiple drop batches dispatched in parallel.
+ hiveSyncProps.setProperty(HIVE_BATCH_SYNC_PARTITION_NUM.key(), "2");
+
+ int partitionCount = 8;
+ HiveTestUtil.createCOWTable("100", partitionCount, true);
+ reInitHiveSyncClient();
+ reSyncHiveTable();
+ assertEquals(partitionCount,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size(),
+ "All partitions should be added before drop test");
+
+ // Drop half the partitions through the parallel pool path.
+ List<String> existing =
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).stream()
+ .map(p -> getRelativePartitionPath(new Path(basePath), new
Path(p.getStorageLocation())))
+ .collect(Collectors.toList());
+ List<String> toDrop = existing.subList(0, partitionCount / 2);
+ hiveClient.dropPartitions(HiveTestUtil.TABLE_NAME, toDrop);
+
+ List<Partition> remaining =
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
+ assertEquals(partitionCount - toDrop.size(), remaining.size(),
+ "Parallel DROP should remove exactly the requested partitions");
+ Set<String> remainingPaths = remaining.stream()
+ .map(p -> getRelativePartitionPath(new Path(basePath), new
Path(p.getStorageLocation())))
+ .collect(Collectors.toSet());
+ for (String dropped : toDrop) {
+ assertFalse(remainingPaths.contains(dropped),
+ "Dropped partition " + dropped + " must not appear in remaining
set");
+ }
+ }
+
/**
* Exercises the SET_LOCATION path in HiveQL mode with batching on.
SET_LOCATION
* emits one ALTER PARTITION ... SET LOCATION statement per partition (Hive
SQL
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncClientClose.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncClientClose.java
index 53c725dc6a3b..b080ff82cfc1 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncClientClose.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncClientClose.java
@@ -18,6 +18,7 @@
package org.apache.hudi.hive;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.hive.ddl.DDLExecutor;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
@@ -44,7 +45,7 @@ class TestHoodieHiveSyncClientClose {
@Test
void closeReleasesProxiedMetastoreClientDirectly() throws Exception {
- HoodieHiveSyncClient syncClient = mock(HoodieHiveSyncClient.class,
CALLS_REAL_METHODS);
+ HoodieHiveSyncClient syncClient = newSyncClientUnderTest();
IMetaStoreClient metaStoreClient = mock(IMetaStoreClient.class);
DDLExecutor ddlExecutor = mock(DDLExecutor.class);
setField(syncClient, "client", metaStoreClient);
@@ -58,7 +59,7 @@ class TestHoodieHiveSyncClientClose {
@Test
void closeSwallowsProxyCloseFailure() throws Exception {
- HoodieHiveSyncClient syncClient = mock(HoodieHiveSyncClient.class,
CALLS_REAL_METHODS);
+ HoodieHiveSyncClient syncClient = newSyncClientUnderTest();
IMetaStoreClient metaStoreClient = mock(IMetaStoreClient.class);
DDLExecutor ddlExecutor = mock(DDLExecutor.class);
doThrow(new RuntimeException("transient close
failure")).when(metaStoreClient).close();
@@ -70,6 +71,19 @@ class TestHoodieHiveSyncClientClose {
verify(metaStoreClient).close();
}
+ /**
+ * CALLS_REAL_METHODS skips the constructor, so fields that close()
dereferences are left
+ * null rather than carrying their declared initializers. Seed those here:
close() reads
+ * partitionClientPool before it reaches the client, so leaving it null
fails these tests
+ * with an NPE that looks like "zero interactions" with the client mock.
+ */
+ private static HoodieHiveSyncClient newSyncClientUnderTest() throws
Exception {
+ HoodieHiveSyncClient syncClient = mock(HoodieHiveSyncClient.class,
CALLS_REAL_METHODS);
+ setField(syncClient, "partitionClientPool", Option.empty());
+ setField(syncClient, "partitionDriverPool", Option.empty());
+ return syncClient;
+ }
+
private static void setField(Object target, String name, Object value)
throws Exception {
Field field = HoodieHiveSyncClient.class.getDeclaredField(name);
field.setAccessible(true);
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestDropPartitionExtractorThreading.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestDropPartitionExtractorThreading.java
new file mode 100644
index 000000000000..5d6c3df04b2e
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestDropPartitionExtractorThreading.java
@@ -0,0 +1,196 @@
+/*
+ * 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.Option;
+import org.apache.hudi.hive.HiveSyncConfig;
+import org.apache.hudi.hive.ddl.HiveQueryDDLExecutor;
+import org.apache.hudi.sync.common.model.PartitionValueExtractor;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_BATCH_SYNC_PARTITION_NUM;
+import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
+import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_PARTITION_FIELDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The DROP path is the only sync path that fans work across pool workers
while a
+ * user-supplied {@link PartitionValueExtractor} is in play.
ADD/TOUCH/SET_LOCATION build
+ * their clauses on the calling thread before dispatch, so they never reach a
shared
+ * extractor from more than one thread.
+ *
+ * <p>Extractors are pluggable and carry no thread-safety contract. One
holding mutable
+ * state (a {@code SimpleDateFormat}, say) could return a garbled clause under
concurrency
+ * and drop the wrong partition, so the values are resolved before any fan-out.
+ *
+ * <p>Lives in {@code hive.util} rather than next to the executor because it
needs the
+ * package-private {@link HiveMetaStoreClientPool} constructor that takes
pre-built clients.
+ */
+class TestDropPartitionExtractorThreading {
+
+ @Test
+ void extractorIsNeverInvokedFromAPoolWorker() throws Exception {
+ // Given a pool wide enough to fan out, and more partitions than a single
batch holds.
+ List<IMetaStoreClient> poolClients = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ poolClients.add(mock(IMetaStoreClient.class));
+ }
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(poolClients, 4);
+
+ Set<String> extractorThreads = ConcurrentHashMap.newKeySet();
+ PartitionValueExtractor recordingExtractor =
mock(PartitionValueExtractor.class);
+
when(recordingExtractor.extractPartitionValuesInPath(anyString())).thenAnswer(inv
-> {
+ extractorThreads.add(Thread.currentThread().getName());
+ return Collections.singletonList("2026-08-06");
+ });
+
+ HiveSyncConfig config = mock(HiveSyncConfig.class);
+
when(config.getStringOrDefault(META_SYNC_DATABASE_NAME)).thenReturn("test_db");
+ when(config.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM)).thenReturn(2);
+ when(config.getSplitStrings(META_SYNC_PARTITION_FIELDS))
+ .thenReturn(Collections.singletonList("datestr"));
+
+ HiveQueryDDLExecutor executor = mock(HiveQueryDDLExecutor.class,
CALLS_REAL_METHODS);
+ setField(executor, "driverPool", Option.empty());
+ setField(executor, "metaStoreClient", mock(IMetaStoreClient.class));
+ setField(executor, "metaStoreClientPool", Option.of(pool));
+ setField(executor, "databaseName", "test_db");
+ setField(executor, "config", config);
+ setField(executor, "partitionValueExtractor", recordingExtractor);
+
+ List<String> partitions = new ArrayList<>();
+ for (int i = 0; i < 16; i++) {
+ partitions.add("datestr=2026-08-" + String.format("%02d", i + 1));
+ }
+
+ try {
+ // When the partitions are dropped across the pool.
+ executor.dropPartitionsToTable("table", partitions);
+
+ // Then every extractor call happened on this thread, never on a worker.
+ assertEquals(Collections.singleton(Thread.currentThread().getName()),
extractorThreads,
+ "PartitionValueExtractor must only be invoked on the calling thread;
a pool "
+ + "worker thread here means a custom extractor is being shared
concurrently");
+
+ // Sanity: the drops really did reach the pool, so the assertion above
had something
+ // to catch rather than passing on a path that never fanned out at all.
+ verify(poolClients.get(0), atLeastOnce()).getPartition(anyString(),
anyString(), anyList());
+ } finally {
+ pool.close();
+ }
+ }
+
+ /**
+ * {@link PartitionValueExtractor} does not require returning a fresh list,
so an
+ * implementation may hand back one buffer it clears and refills per call.
Merely wrapping
+ * that list unmodifiable would leave every resolved partition aliasing the
final
+ * extraction, and {@code partitionExists} would then check the wrong
partition — skipping
+ * valid drops. The values must be copied at resolution time.
+ */
+ @Test
+ void extractorReusingOneBufferStillYieldsPerPartitionValues() throws
Exception {
+ List<IMetaStoreClient> poolClients = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ poolClients.add(mock(IMetaStoreClient.class));
+ }
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(poolClients, 2);
+
+ // Given an extractor that recycles a single mutable list across calls.
+ List<String> recycled = new ArrayList<>();
+ PartitionValueExtractor reusingExtractor =
mock(PartitionValueExtractor.class);
+
when(reusingExtractor.extractPartitionValuesInPath(anyString())).thenAnswer(inv
-> {
+ String partition = inv.getArgument(0, String.class);
+ recycled.clear();
+ recycled.add(partition.replace("datestr=", ""));
+ return recycled;
+ });
+
+ HiveSyncConfig config = mock(HiveSyncConfig.class);
+
when(config.getStringOrDefault(META_SYNC_DATABASE_NAME)).thenReturn("test_db");
+ when(config.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM)).thenReturn(2);
+ when(config.getSplitStrings(META_SYNC_PARTITION_FIELDS))
+ .thenReturn(Collections.singletonList("datestr"));
+
+ HiveQueryDDLExecutor executor = mock(HiveQueryDDLExecutor.class,
CALLS_REAL_METHODS);
+ setField(executor, "driverPool", Option.empty());
+ setField(executor, "metaStoreClient", mock(IMetaStoreClient.class));
+ setField(executor, "metaStoreClientPool", Option.of(pool));
+ setField(executor, "databaseName", "test_db");
+ setField(executor, "config", config);
+ setField(executor, "partitionValueExtractor", reusingExtractor);
+
+ Set<String> lookedUp = ConcurrentHashMap.newKeySet();
+ for (IMetaStoreClient client : poolClients) {
+ when(client.getPartition(anyString(), anyString(),
anyList())).thenAnswer(inv -> {
+ lookedUp.add(String.join("/", (List<String>) inv.getArgument(2)));
+ throw new NoSuchObjectException("absent");
+ });
+ }
+
+ try {
+ // When four distinct partitions are dropped.
+ executor.dropPartitionsToTable("table",
+ Arrays.asList("datestr=2026-08-01", "datestr=2026-08-02",
+ "datestr=2026-08-03", "datestr=2026-08-04"));
+
+ // Then each was looked up with its own values, not four copies of the
last one.
+ assertEquals(new HashSet<>(Arrays.asList("2026-08-01", "2026-08-02",
+ "2026-08-03", "2026-08-04")), lookedUp,
+ "each partition must be checked with the values extracted for it;
identical "
+ + "values here mean the resolved list aliased the extractor's
reused buffer");
+ } finally {
+ pool.close();
+ }
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ Class<?> type = target.getClass();
+ while (type != null) {
+ try {
+ Field field = type.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ return;
+ } catch (NoSuchFieldException e) {
+ type = type.getSuperclass();
+ }
+ }
+ throw new NoSuchFieldException(name);
+ }
+}
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 5b946447ba4d..0f5df5a54399 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
@@ -109,7 +109,7 @@ class TestHiveDriverPool {
};
try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
List<String> sqls = Arrays.asList("SELECT 1", "SELECT 2", "SELECT 3",
"SELECT 4");
- HiveDriverPool.Dispatch futures = pool.dispatchAll(sqls);
+ ParallelDispatch futures = pool.dispatchAll(sqls);
pool.awaitAll(futures);
assertEquals(2, seenThreadsByDriver.size(), "Expected exactly 2 worker
Drivers");
int totalCalls =
seenThreadsByDriver.values().stream().mapToInt(Set::size).sum();
@@ -136,7 +136,7 @@ class TestHiveDriverPool {
return d;
};
try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
- HiveDriverPool.Dispatch futures = pool.dispatchAll(Arrays.asList("OK",
"FAIL", "OK"));
+ ParallelDispatch futures = pool.dispatchAll(Arrays.asList("OK", "FAIL",
"OK"));
HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
() -> pool.awaitAll(futures));
assertNotNull(ex.getCause());
@@ -163,7 +163,7 @@ class TestHiveDriverPool {
};
try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
// 5 SQLs against pool of size 2 → max in-flight should be 2.
- HiveDriverPool.Dispatch futures = pool.dispatchAll(Arrays.asList("a",
"b", "c", "d", "e"));
+ ParallelDispatch futures = pool.dispatchAll(Arrays.asList("a", "b", "c",
"d", "e"));
// Release after a short wait so all SQLs progress.
Thread.sleep(150);
hold.countDown();
@@ -214,7 +214,7 @@ class TestHiveDriverPool {
};
try (HiveDriverPool pool = new HiveDriverPool(config, 3, factory)) {
pool.runOnEachWorker(Arrays.asList("USE `db1`"));
- HiveDriverPool.Dispatch futures = pool.dispatchAll(Arrays.asList("ALTER
1", "ALTER 2", "ALTER 3"));
+ ParallelDispatch futures = pool.dispatchAll(Arrays.asList("ALTER 1",
"ALTER 2", "ALTER 3"));
pool.awaitAll(futures);
assertEquals(3, sqlsByDriver.size(), "Expected one Driver per worker");
@@ -251,7 +251,7 @@ class TestHiveDriverPool {
return d;
};
try (HiveDriverPool pool = new HiveDriverPool(config, 1, factory)) {
- HiveDriverPool.Dispatch dispatch =
pool.dispatchAll(Arrays.asList("FAIL", "PENDING_A", "PENDING_B"));
+ ParallelDispatch dispatch = pool.dispatchAll(Arrays.asList("FAIL",
"PENDING_A", "PENDING_B"));
HoodieHiveSyncException ex = assertThrows(HoodieHiveSyncException.class,
() -> pool.awaitAll(dispatch));
@@ -302,7 +302,7 @@ class TestHiveDriverPool {
};
try (HiveDriverPool pool = new HiveDriverPool(config, 2, factory)) {
// Round-robin over 2 workers: index 0 -> worker 0, indices 1 and 2 ->
worker 1.
- HiveDriverPool.Dispatch dispatch =
+ ParallelDispatch dispatch =
pool.dispatchAll(Arrays.asList("SLOW", "FAIL", "AFTER_FAIL"));
assertTrue(failed.await(5, TimeUnit.SECONDS), "FAIL must have run");
releaseSlow.countDown();
@@ -349,7 +349,7 @@ class TestHiveDriverPool {
return d;
};
try (HiveDriverPool pool = new HiveDriverPool(config, 1, factory)) {
- HiveDriverPool.Dispatch dispatch =
pool.dispatchAll(Collections.singletonList("FAIL"));
+ ParallelDispatch 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");
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveMetaStoreClientPool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveMetaStoreClientPool.java
new file mode 100644
index 000000000000..c491aee80e7e
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveMetaStoreClientPool.java
@@ -0,0 +1,385 @@
+/*
+ * 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.hadoop.hive.metastore.IMetaStoreClient;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+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.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Unit tests for {@link HiveMetaStoreClientPool}: borrow/return bounding,
batch fan-out,
+ * abort-on-first-error, and close semantics. Uses mock clients so no
metastore is needed.
+ */
+class TestHiveMetaStoreClientPool {
+
+ private static List<IMetaStoreClient> mockClients(int n) {
+ return IntStream.range(0, n)
+ .mapToObj(i -> mock(IMetaStoreClient.class))
+ .collect(Collectors.toList());
+ }
+
+ @Test
+ void runBorrowsAndReturnsTheSameClient() throws Exception {
+ List<IMetaStoreClient> clients = mockClients(1);
+ try (HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients,
1)) {
+ // Given a single-client pool, two sequential borrows must both succeed
--
+ // which can only happen if the first borrow returned its client.
+ IMetaStoreClient first = pool.run(c -> c);
+ IMetaStoreClient second = pool.run(c -> c);
+
+ assertEquals(clients.get(0), first);
+ assertEquals(first, second, "Client must be returned to the pool after
each run");
+ }
+ }
+
+ @Test
+ void runReturnsClientEvenWhenActionThrows() throws Exception {
+ List<IMetaStoreClient> clients = mockClients(1);
+ try (HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients,
1)) {
+ assertThrows(IllegalStateException.class, () -> pool.run(c -> {
+ throw new IllegalStateException("boom");
+ }));
+
+ // If the failed borrow had leaked the client, this would block forever.
+ IMetaStoreClient reused = pool.run(c -> c);
+ assertEquals(clients.get(0), reused, "A failed action must still return
its client");
+ }
+ }
+
+ @Test
+ void dispatchAllRunsEveryBatch() throws Exception {
+ List<IMetaStoreClient> clients = mockClients(3);
+ List<String> batches = Arrays.asList("b0", "b1", "b2", "b3", "b4");
+ List<String> applied = Collections.synchronizedList(new ArrayList<>());
+ try (HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients,
3)) {
+ pool.awaitAll(pool.dispatchAll(batches, (client, batch) ->
applied.add(batch)), "test");
+ }
+
+ assertEquals(5, applied.size());
+ assertTrue(applied.containsAll(batches), "Every batch must be applied
exactly once");
+ }
+
+ @Test
+ void concurrentBatchesBoundedByPoolSize() throws Exception {
+ int poolSize = 2;
+ List<IMetaStoreClient> clients = mockClients(poolSize);
+ AtomicInteger inFlight = new AtomicInteger();
+ AtomicInteger maxInFlight = new AtomicInteger();
+ try (HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients,
poolSize)) {
+ pool.awaitAll(pool.dispatchAll(Arrays.asList("a", "b", "c", "d", "e"),
(client, batch) -> {
+ int now = inFlight.incrementAndGet();
+ maxInFlight.accumulateAndGet(now, Math::max);
+ Thread.sleep(20);
+ inFlight.decrementAndGet();
+ }), "test");
+ }
+
+ assertTrue(maxInFlight.get() <= poolSize,
+ "In-flight Thrift calls must never exceed the client count, saw " +
maxInFlight.get());
+ }
+
+ @Test
+ void eachConcurrentBatchGetsADistinctClient() throws Exception {
+ int poolSize = 3;
+ List<IMetaStoreClient> clients = mockClients(poolSize);
+ Set<IMetaStoreClient> seenConcurrently = ConcurrentHashMap.newKeySet();
+ CountDownLatch allBorrowed = new CountDownLatch(poolSize);
+ try (HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients,
poolSize)) {
+ pool.awaitAll(pool.dispatchAll(Arrays.asList("a", "b", "c"), (client,
batch) -> {
+ seenConcurrently.add(client);
+ // Hold every client at once so none can be recycled to another batch.
+ allBorrowed.countDown();
+ assertTrue(allBorrowed.await(5, TimeUnit.SECONDS));
+ }), "test");
+ }
+
+ assertEquals(poolSize, seenConcurrently.size(),
+ "Concurrent batches must not share a Thrift client");
+ }
+
+ @Test
+ void awaitAllThrowsFirstError() {
+ List<IMetaStoreClient> clients = mockClients(2);
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients, 2);
+ try {
+ Exception ex = assertThrows(Exception.class, () ->
+ pool.awaitAll(pool.dispatchAll(Arrays.asList("ok", "boom"), (client,
batch) -> {
+ if (batch.equals("boom")) {
+ throw new IllegalStateException("drop failed");
+ }
+ }), "test"));
+
+ assertEquals("drop failed", ex.getMessage(),
+ "The original cause must surface unwrapped, not as an
ExecutionException");
+ } finally {
+ pool.close();
+ }
+ }
+
+ /**
+ * Regression for the in-order-await bug: waiting on futures in submission
order lets
+ * the executor keep starting queued batches after a sibling has already
failed.
+ *
+ * <p>Given a single-client pool so batches run strictly in order, when the
first batch
+ * fails, then no later batch may reach the metastore -- the task-side abort
flag has to
+ * stop them, since {@code Future.cancel} from the awaiting thread is
inherently late.
+ */
+ @Test
+ void abortStopsQueuedBatchesAfterFirstFailure() {
+ List<IMetaStoreClient> clients = mockClients(1);
+ List<String> applied = Collections.synchronizedList(new ArrayList<>());
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients, 1);
+ try {
+ assertThrows(Exception.class, () ->
+ pool.awaitAll(pool.dispatchAll(Arrays.asList("FAIL", "AFTER_A",
"AFTER_B"),
+ (client, batch) -> {
+ applied.add(batch);
+ if (batch.equals("FAIL")) {
+ throw new IllegalStateException("drop failed");
+ }
+ }), "test"));
+
+ assertEquals(Collections.singletonList("FAIL"), applied,
+ "Batches queued behind the failure must never reach the metastore");
+ } finally {
+ pool.close();
+ }
+ }
+
+ /**
+ * The same abort guarantee, but with the failure landing on a <i>later</i>
future than
+ * a still-running one. This is the interleaving Future-order waiting cannot
handle: the
+ * awaiting thread is parked on SLOW's future while the failing batch races
ahead.
+ */
+ @Test
+ void abortStopsQueuedBatchesWhenEarlierBatchIsSlow() throws Exception {
+ List<IMetaStoreClient> clients = mockClients(2);
+ List<String> applied = Collections.synchronizedList(new ArrayList<>());
+ CountDownLatch failed = new CountDownLatch(1);
+ CountDownLatch releaseSlow = new CountDownLatch(1);
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients, 2);
+ try {
+ // SLOW occupies one client and blocks until FAIL has thrown, pinning the
+ // interleaving where awaitAll is still parked on future 0.
+ List<String> batches = Arrays.asList("SLOW", "FAIL", "AFTER_FAIL",
"AFTER_FAIL_2");
+ ParallelDispatch dispatch = pool.dispatchAll(batches, (client, batch) ->
{
+ applied.add(batch);
+ if (batch.equals("FAIL")) {
+ failed.countDown();
+ throw new IllegalStateException("drop failed");
+ }
+ if (batch.equals("SLOW")) {
+ releaseSlow.await(5, TimeUnit.SECONDS);
+ }
+ });
+ assertTrue(failed.await(5, TimeUnit.SECONDS), "FAIL must have run");
+ releaseSlow.countDown();
+
+ assertThrows(Exception.class, () -> pool.awaitAll(dispatch, "test"));
+
+ assertFalse(applied.contains("AFTER_FAIL_2"),
+ "Batches queued behind a failure must not be applied while an
earlier "
+ + "batch on another client is still running");
+ } finally {
+ pool.close();
+ }
+ }
+
+ @Test
+ void firstErrorIsTheFailureThatAbortedTheBatch() throws Exception {
+ List<IMetaStoreClient> clients = mockClients(2);
+ AtomicReference<ParallelDispatch> dispatchRef = new AtomicReference<>();
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients, 2);
+ try {
+ // Given a slow batch submitted first that fails only after a later
batch has
+ // already failed and tripped the abort. Selecting the error by
submission order
+ // would report "slow-later"; the batch was actually stopped by
"fast-first".
+ //
+ // The slow batch waits on the abort flag itself rather than on a latch
counted
+ // down before the throw: only the flag proves fast-first has already
reached
+ // abort(), so this pins the interleaving instead of merely making it
likely.
+ List<String> batches = Arrays.asList("SLOW_LATER", "FAST_FIRST");
+ ParallelDispatch dispatch = pool.dispatchAll(batches, (client, batch) ->
{
+ if (batch.equals("FAST_FIRST")) {
+ throw new IllegalStateException("fast-first");
+ }
+ ParallelDispatch inFlight;
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while ((inFlight = dispatchRef.get()) == null || !inFlight.aborted()) {
+ if (System.nanoTime() > deadline) {
+ throw new IllegalStateException("timed-out-waiting-for-abort");
+ }
+ Thread.sleep(1);
+ }
+ throw new IllegalStateException("slow-later");
+ });
+ dispatchRef.set(dispatch);
+
+ Exception thrown = assertThrows(Exception.class, () ->
pool.awaitAll(dispatch, "test"));
+
+ assertEquals("fast-first", thrown.getMessage(),
+ "The reported root cause must be the failure that aborted the batch,
"
+ + "not whichever failure was submitted earliest");
+ } finally {
+ pool.close();
+ }
+ }
+
+ @Test
+ void anErrorThatAbortsTheBatchIsNotAlsoReportedAsSuppressed() throws
Exception {
+ List<IMetaStoreClient> clients = mockClients(2);
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients, 2);
+ try {
+ // Given the aborting task throws an Error rather than an Exception.
guard() catches
+ // Throwable, so the Error is what trips the abort, but it cannot be
reported as-is:
+ // it gets wrapped in a RuntimeException. An identity check against that
wrapper
+ // would never match the raw cause coming back off the future, so the
one real
+ // failure would be reported twice -- once as the cause, once as
suppressed.
+ ParallelDispatch dispatch = pool.dispatchAll(
+ Collections.singletonList("BOOM"),
+ (client, batch) -> {
+ throw new StackOverflowError("boom");
+ });
+
+ // Asserted on the Outcome rather than on awaitAll(): awaitAll only
*logs* the
+ // suppressed list, so a duplicate there is invisible to the thrown
exception.
+ ParallelDispatch.Outcome outcome = dispatch.awaitOutcome();
+
+ assertTrue(outcome.suppressed().isEmpty(),
+ "The failure that aborted the batch must not also appear in its own
suppressed list");
+ assertEquals("boom", outcome.firstError().getCause().getMessage(),
+ "The wrapped Error must still be reported as the root cause");
+ } finally {
+ pool.close();
+ }
+ }
+
+ @Test
+ void interruptedAwaitIsReportedAsAFailureNotASuccess() throws Exception {
+ List<IMetaStoreClient> clients = mockClients(2);
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients, 2);
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch running = new CountDownLatch(1);
+ try {
+ // Given a batch that is still in flight when the awaiting thread is
interrupted.
+ // The interrupt must surface as a dispatch failure: reporting success
here would let
+ // HiveSyncTool advance the last-synced commit marker over work that
never ran.
+ ParallelDispatch dispatch = pool.dispatchAll(
+ Collections.singletonList("BLOCKED"),
+ (client, batch) -> {
+ running.countDown();
+ release.await(10, TimeUnit.SECONDS);
+ return;
+ });
+ assertTrue(running.await(10, TimeUnit.SECONDS), "the batch must have
started");
+
+ AtomicReference<Throwable> thrown = new AtomicReference<>();
+ AtomicBoolean returnedNormally = new AtomicBoolean(false);
+ Thread awaiter = new Thread(() -> {
+ try {
+ pool.awaitAll(dispatch, "test");
+ returnedNormally.set(true);
+ } catch (Throwable t) {
+ thrown.set(t);
+ }
+ }, "awaiter");
+ awaiter.start();
+
+ // When that thread is interrupted while parked in awaitAll.
+ Thread.sleep(200);
+ awaiter.interrupt();
+ awaiter.join(10_000);
+ release.countDown();
+
+ // Then it reports a failure rather than a clean batch.
+ assertFalse(returnedNormally.get(),
+ "awaitAll must not report success when the wait was interrupted; the
batch's "
+ + "work was cancelled or is still in flight");
+ assertNotNull(thrown.get(), "the interruption must surface as a thrown
failure");
+ } finally {
+ release.countDown();
+ pool.close();
+ }
+ }
+
+ @Test
+ void invalidPoolSizeIsRejectedWithAClearMessage() {
+ // Both zero and negative must report the pool's own message. Negative is
the reason
+ // buildClients keeps its own guard: without it, ArrayList's capacity
check fires first
+ // and the caller sees "Illegal Capacity: -1" instead.
+ for (int badSize : new int[] {0, -1}) {
+ IllegalArgumentException thrown =
assertThrows(IllegalArgumentException.class,
+ () -> new HiveMetaStoreClientPool(Collections.emptyList(), badSize),
+ "pool size " + badSize + " must be rejected");
+ assertTrue(thrown.getMessage().contains("Pool size must be >= 1"),
+ "expected the pool's own size message, got: " + thrown.getMessage());
+ }
+ }
+
+ @Test
+ void closeIsIdempotentAndClosesEveryClient() throws Exception {
+ List<IMetaStoreClient> clients = mockClients(2);
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(clients, 2);
+
+ pool.close();
+ pool.close();
+
+ for (IMetaStoreClient client : clients) {
+ verify(client).close();
+ }
+ assertThrows(IllegalStateException.class, () -> pool.run(c -> c),
+ "Borrowing from a closed pool must fail fast");
+ }
+
+ @Test
+ void dispatchOnClosedPoolFailsFast() {
+ HiveMetaStoreClientPool pool = new HiveMetaStoreClientPool(mockClients(1),
1);
+ pool.close();
+
+ assertThrows(IllegalStateException.class,
+ () -> pool.dispatchAll(Collections.singletonList("a"), (client, batch)
-> { }));
+ }
+
+ @Test
+ void rejectsSizeMismatchAndNonPositiveSize() {
+ assertThrows(IllegalArgumentException.class, () -> new
HiveMetaStoreClientPool(mockClients(1), 2));
+ assertThrows(IllegalArgumentException.class, () -> new
HiveMetaStoreClientPool(mockClients(0), 0));
+ }
+}
