shangxinli commented on code in PR #18983:
URL: https://github.com/apache/hudi/pull/18983#discussion_r4433996823
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HMSDDLExecutor.java:
##########
@@ -291,21 +312,71 @@ public void close() {
private void registerAlterPartitionEvent(String tableName, List<String>
alteredPartitions) {
try {
+ // Read the StorageDescriptor once on the session client; each worker
deep-copies
+ // it per partition (alter_partitions semantics today).
StorageDescriptor sd = client.getTable(databaseName, tableName).getSd();
- List<Partition> partitionList = alteredPartitions.stream().map(partition
-> {
- Path partitionPath =
HadoopFSUtils.constructAbsolutePathInHadoopPath(syncConfig.getString(META_SYNC_BASE_PATH),
partition);
- String partitionScheme = partitionPath.toUri().getScheme();
- String fullPartitionPath =
StorageSchemes.HDFS.getScheme().equals(partitionScheme)
- ?
HadoopFSUtils.getDFSFullPartitionPath(syncConfig.getHadoopFileSystem(),
partitionPath) : partitionPath.toString();
- List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
- StorageDescriptor partitionSd = sd.deepCopy();
- partitionSd.setLocation(fullPartitionPath);
- return new Partition(partitionValues, databaseName, tableName, 0, 0,
partitionSd, null);
- }).collect(Collectors.toList());
- client.alter_partitions(databaseName, tableName, partitionList, null);
- } catch (TException e) {
+ int batchSyncPartitionNum =
syncConfig.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM);
+ List<List<String>> batches = CollectionUtils.batches(alteredPartitions,
batchSyncPartitionNum);
+ runBatches("alter", tableName, batches, (poolClient, batch) -> {
+ List<Partition> partitionList = batch.stream().map(partition -> {
+ Path partitionPath =
HadoopFSUtils.constructAbsolutePathInHadoopPath(syncConfig.getString(META_SYNC_BASE_PATH),
partition);
+ String partitionScheme = partitionPath.toUri().getScheme();
+ String fullPartitionPath =
StorageSchemes.HDFS.getScheme().equals(partitionScheme)
+ ?
HadoopFSUtils.getDFSFullPartitionPath(syncConfig.getHadoopFileSystem(),
partitionPath) : partitionPath.toString();
+ List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
+ StorageDescriptor partitionSd = sd.deepCopy();
+ partitionSd.setLocation(fullPartitionPath);
+ return new Partition(partitionValues, databaseName, tableName, 0, 0,
partitionSd, null);
+ }).collect(Collectors.toList());
+ poolClient.alter_partitions(databaseName, tableName, partitionList,
null);
+ });
+ } catch (Exception e) {
log.error("{}.{} update partition failed", databaseName, tableName, e);
throw new HoodieHiveSyncException(databaseName + "." + tableName + "
update partition failed", e);
}
}
+
+ /**
+ * Dispatches partition batches either in parallel against the client pool
(if
+ * configured) or sequentially against the session client. The sequential
path
+ * preserves the exact behavior we had before the pool existed, including
failure
+ * semantics where batch N+1 never runs if batch N throws.
+ */
+ @FunctionalInterface
+ private interface BatchAction {
+ void apply(IMetaStoreClient client, List<String> batch) throws Exception;
+ }
+
+ private void runBatches(String opName, String tableName, List<List<String>>
batches, BatchAction action) throws Exception {
+ if (partitionClientPool == null) {
+ for (List<String> batch : batches) {
+ action.apply(client, batch);
+ }
+ return;
+ }
+ List<Future<Void>> futures = new ArrayList<>(batches.size());
+ for (List<String> batch : batches) {
+ futures.add(partitionClientPool.executor().submit(() ->
+ partitionClientPool.run(poolClient -> {
Review Comment:
all batches are submitted before any f.get() is called below, so later
batches still commit against HMS even after an earlier one has failed. the
fail-fast guarantee only holds on the non-pooled sequential path.
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HMSDDLExecutor.java:
##########
@@ -291,21 +312,71 @@ public void close() {
private void registerAlterPartitionEvent(String tableName, List<String>
alteredPartitions) {
try {
+ // Read the StorageDescriptor once on the session client; each worker
deep-copies
+ // it per partition (alter_partitions semantics today).
StorageDescriptor sd = client.getTable(databaseName, tableName).getSd();
- List<Partition> partitionList = alteredPartitions.stream().map(partition
-> {
- Path partitionPath =
HadoopFSUtils.constructAbsolutePathInHadoopPath(syncConfig.getString(META_SYNC_BASE_PATH),
partition);
- String partitionScheme = partitionPath.toUri().getScheme();
- String fullPartitionPath =
StorageSchemes.HDFS.getScheme().equals(partitionScheme)
- ?
HadoopFSUtils.getDFSFullPartitionPath(syncConfig.getHadoopFileSystem(),
partitionPath) : partitionPath.toString();
- List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
- StorageDescriptor partitionSd = sd.deepCopy();
- partitionSd.setLocation(fullPartitionPath);
- return new Partition(partitionValues, databaseName, tableName, 0, 0,
partitionSd, null);
- }).collect(Collectors.toList());
- client.alter_partitions(databaseName, tableName, partitionList, null);
- } catch (TException e) {
+ int batchSyncPartitionNum =
syncConfig.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM);
+ List<List<String>> batches = CollectionUtils.batches(alteredPartitions,
batchSyncPartitionNum);
+ runBatches("alter", tableName, batches, (poolClient, batch) -> {
+ List<Partition> partitionList = batch.stream().map(partition -> {
+ Path partitionPath =
HadoopFSUtils.constructAbsolutePathInHadoopPath(syncConfig.getString(META_SYNC_BASE_PATH),
partition);
+ String partitionScheme = partitionPath.toUri().getScheme();
+ String fullPartitionPath =
StorageSchemes.HDFS.getScheme().equals(partitionScheme)
+ ?
HadoopFSUtils.getDFSFullPartitionPath(syncConfig.getHadoopFileSystem(),
partitionPath) : partitionPath.toString();
+ List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
+ StorageDescriptor partitionSd = sd.deepCopy();
+ partitionSd.setLocation(fullPartitionPath);
+ return new Partition(partitionValues, databaseName, tableName, 0, 0,
partitionSd, null);
+ }).collect(Collectors.toList());
+ poolClient.alter_partitions(databaseName, tableName, partitionList,
null);
+ });
+ } catch (Exception e) {
log.error("{}.{} update partition failed", databaseName, tableName, e);
throw new HoodieHiveSyncException(databaseName + "." + tableName + "
update partition failed", e);
}
}
+
+ /**
+ * Dispatches partition batches either in parallel against the client pool
(if
+ * configured) or sequentially against the session client. The sequential
path
+ * preserves the exact behavior we had before the pool existed, including
failure
+ * semantics where batch N+1 never runs if batch N throws.
+ */
+ @FunctionalInterface
+ private interface BatchAction {
+ void apply(IMetaStoreClient client, List<String> batch) throws Exception;
+ }
+
+ private void runBatches(String opName, String tableName, List<List<String>>
batches, BatchAction action) throws Exception {
+ if (partitionClientPool == null) {
+ for (List<String> batch : batches) {
+ action.apply(client, batch);
+ }
+ return;
+ }
+ List<Future<Void>> futures = new ArrayList<>(batches.size());
+ for (List<String> batch : batches) {
+ futures.add(partitionClientPool.executor().submit(() ->
+ partitionClientPool.run(poolClient -> {
+ action.apply(poolClient, batch);
+ return null;
+ })
+ ));
+ }
+ Exception firstError = null;
+ for (Future<Void> f : futures) {
+ try {
+ f.get();
+ } catch (Exception e) {
+ if (firstError == null) {
Review Comment:
f.get() throws ExecutionException wrapping the real cause, but that wrapper
is what gets stored and rethrown, changing the exception type callers see. also
an InterruptedException here gets swallowed without restoring the thread's
interrupt status.
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/IMetaStoreClientPool.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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. A
worker
+ * thread pool of the same size is exposed via {@link #executor()} so callers
can fan
+ * out their batches to match 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 only for sync mode HMS.
+ */
+public class IMetaStoreClientPool implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IMetaStoreClientPool.class);
+
+ private final ArrayBlockingQueue<IMetaStoreClient> available;
+ private final List<IMetaStoreClient> all;
+ private final ExecutorService executor;
+ private final int size;
+ private volatile boolean closed;
+
+ public IMetaStoreClientPool(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.
+ IMetaStoreClientPool(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) {
+ 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) {
Review Comment:
race with close(): a worker can pass this check, then close() sets closed
and clears the queue, then this offer() runs afterward and re-inserts an
already-closed client into the queue for a future borrower.
--
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]
