This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 6318c4084 [AMORO-4269] Refactor Iceberg thread pool initialization and
scope (#4270)
6318c4084 is described below
commit 6318c408471847e4a6628104dc68764395bf1b19
Author: Xu Bai <[email protected]>
AuthorDate: Mon Jul 20 11:36:07 2026 +0800
[AMORO-4269] Refactor Iceberg thread pool initialization and scope (#4270)
* Refactor Iceberg thread pool initialization and usage for improved
configuration management
* Fix Iceberg thread pool initialization in tests
* Restore Iceberg thread pool API compatibility
* Make Iceberg thread pool methods public for enhanced accessibility
* Document Iceberg thread pool fallback
* Remove unused fallback warning set from Iceberg thread pool implementation
---
.../apache/amoro/server/AmoroServiceContainer.java | 13 +-
.../org/apache/amoro/utils/IcebergThreadPools.java | 153 +++++++++++++++------
.../apache/amoro/utils/TestIcebergThreadPools.java | 65 +++++++++
3 files changed, 186 insertions(+), 45 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
index 638354983..b9eb95f2d 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
@@ -84,7 +84,7 @@ import
org.apache.amoro.shade.thrift.org.apache.thrift.transport.layered.TFramed
import org.apache.amoro.utils.IcebergThreadPools;
import org.apache.amoro.utils.JacksonUtil;
import org.apache.commons.lang3.StringUtils;
-import org.apache.iceberg.SystemProperties;
+import org.apache.iceberg.SystemConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
@@ -582,7 +582,7 @@ public class AmoroServiceContainer {
public void init() throws Exception {
Map<String, Object> envConfig = initEnvConfig();
initServiceConfig(envConfig);
- setIcebergSystemProperties();
+ initIcebergThreadPools();
initContainerConfig();
}
@@ -619,14 +619,17 @@ public class AmoroServiceContainer {
return ConfigHelpers.convertConfigurationKeys(prefix, System.getenv());
}
- /** Override the value of {@link SystemProperties}. */
- private void setIcebergSystemProperties() {
+ /**
+ * Configures Iceberg's global worker pool and initializes self-optimizing
Iceberg I/O pools.
+ */
+ private void initIcebergThreadPools() {
int workerThreadPoolSize =
Math.max(
Runtime.getRuntime().availableProcessors() / 2,
serviceConfig.getInteger(AmoroManagementConf.TABLE_MANIFEST_IO_THREAD_COUNT));
System.setProperty(
- SystemProperties.WORKER_THREAD_POOL_SIZE_PROP,
String.valueOf(workerThreadPoolSize));
+ SystemConfigs.WORKER_THREAD_POOL_SIZE.propertyKey(),
+ String.valueOf(workerThreadPoolSize));
int planningThreadPoolSize =
Math.max(
Runtime.getRuntime().availableProcessors() / 2,
diff --git
a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java
b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java
index 1c19968bc..96c0f8fe5 100644
---
a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java
+++
b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java
@@ -22,60 +22,133 @@ import org.apache.iceberg.util.ThreadPools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
+/** Long-lived Iceberg I/O pools that isolate Amoro maintenance workloads from
one another. */
public class IcebergThreadPools {
private static final Logger LOG =
LoggerFactory.getLogger(IcebergThreadPools.class);
- private static volatile ExecutorService planningExecutor;
- private static volatile ExecutorService commitExecutor;
-
- public static void init(int planningThreadPoolSize, int
commitThreadPoolSize) {
- if (planningExecutor == null) {
- synchronized (IcebergThreadPools.class) {
- if (planningExecutor == null) {
- planningExecutor =
- ThreadPools.newWorkerPool("iceberg-planning-pool",
planningThreadPoolSize);
- }
- }
- }
- if (commitExecutor == null) {
- synchronized (IcebergThreadPools.class) {
- if (commitExecutor == null) {
- commitExecutor = ThreadPools.newWorkerPool("iceberg-commit-pool",
commitThreadPoolSize);
- }
- }
- }
+
+ private static final String PLANNING_POOL_NAME_PREFIX =
"iceberg-planning-pool";
+ private static final String COMMIT_POOL_NAME_PREFIX = "iceberg-commit-pool";
+
+ private static final Map<String, Integer> POOL_SIZES = new
ConcurrentHashMap<>();
+ private static final Map<String, ExecutorService> POOLS = new
ConcurrentHashMap<>();
+
+ /**
+ * Initializes the self-optimizing Iceberg I/O pools.
+ *
+ * <p>Thread pools are process-wide and can only be initialized once.
Repeated initialization is
+ * ignored because existing pools cannot be resized.
+ */
+ public static synchronized void init(int planningThreadPoolSize, int
commitThreadPoolSize) {
+ newThreadPool(PLANNING_POOL_NAME_PREFIX, planningThreadPoolSize);
+ newThreadPool(COMMIT_POOL_NAME_PREFIX, commitThreadPoolSize);
LOG.info(
- "init iceberg thread pool success, planningExecutor
size:{},commitExecutor size:{}",
- planningThreadPoolSize,
- commitThreadPoolSize);
+ "Initialized Iceberg thread pools, self-optimizing planning: {},
self-optimizing commit: {}",
+ POOL_SIZES.get(PLANNING_POOL_NAME_PREFIX),
+ POOL_SIZES.get(COMMIT_POOL_NAME_PREFIX));
}
+ /**
+ * Return an {@link ExecutorService} that uses the self-optimizing planning
pool.
+ *
+ * <p>The size of this pool limits the number of tasks concurrently reading
manifests across all
+ * self-optimizing planning operations.
+ *
+ * <p>The size of this pool is controlled by the AMS configuration {@code
+ * self-optimizing.plan-manifest-io-thread-count}.
+ *
+ * <p>Before the dedicated pool is initialized, this returns Iceberg's
global worker pool.
+ *
+ * @return an {@link ExecutorService} that uses the self-optimizing planning
pool, or Iceberg's
+ * global worker pool if the dedicated pool has not been initialized
+ */
public static ExecutorService getPlanningExecutor() {
- if (planningExecutor == null) {
- synchronized (IcebergThreadPools.class) {
- if (planningExecutor == null) {
- planningExecutor =
- ThreadPools.newWorkerPool(
- "iceberg-planning-pool",
Runtime.getRuntime().availableProcessors());
- }
- }
- }
- return planningExecutor;
+ return getThreadPool(PLANNING_POOL_NAME_PREFIX);
}
+ /**
+ * Return an {@link ExecutorService} that uses the self-optimizing commit
pool.
+ *
+ * <p>The size of this pool limits the number of tasks concurrently
filtering manifests across all
+ * self-optimizing commit operations. It does not replace Iceberg's worker
pool used internally by
+ * {@code SnapshotProducer.writeManifests}.
+ *
+ * <p>The size of this pool is controlled by the AMS configuration {@code
+ * self-optimizing.commit-manifest-io-thread-count}.
+ *
+ * <p>Before the dedicated pool is initialized, this returns Iceberg's
global worker pool.
+ *
+ * @return an {@link ExecutorService} that uses the self-optimizing commit
pool, or Iceberg's
+ * global worker pool if the dedicated pool has not been initialized
+ */
public static ExecutorService getCommitExecutor() {
- if (commitExecutor == null) {
- synchronized (IcebergThreadPools.class) {
- if (commitExecutor == null) {
- commitExecutor =
- ThreadPools.newWorkerPool(
- "iceberg-commit-pool",
Runtime.getRuntime().availableProcessors());
- }
+ return getThreadPool(COMMIT_POOL_NAME_PREFIX);
+ }
+
+ /**
+ * Returns the registered Iceberg thread pool for the given name prefix.
+ *
+ * <p>If the pool has not been initialized, a warning is logged once for the
name prefix and
+ * Iceberg's global worker pool is returned.
+ *
+ * @param namePrefix thread pool name prefix
+ * @return the registered thread pool, or Iceberg's global worker pool if
none is registered
+ * @throws IllegalArgumentException if the name prefix is empty
+ */
+ public static ExecutorService getThreadPool(String namePrefix) {
+ if (namePrefix == null || namePrefix.isEmpty()) {
+ throw new IllegalArgumentException("Thread pool name prefix must not be
empty");
+ }
+
+ ExecutorService executorService = POOLS.get(namePrefix);
+ if (executorService == null) {
+ LOG.warn(
+ "Iceberg thread pool {} has not been initialized; using Iceberg's
global worker pool",
+ namePrefix);
+
+ return ThreadPools.getWorkerPool();
+ }
+ return executorService;
+ }
+
+ /**
+ * Creates and registers a process-wide Iceberg thread pool.
+ *
+ * <p>If a pool with the same name prefix already exists, the existing pool
is retained because
+ * thread pools cannot be resized.
+ *
+ * @param namePrefix thread pool name prefix
+ * @param poolSize number of worker threads
+ * @throws IllegalArgumentException if the name prefix is empty or the pool
size is not positive
+ */
+ public static synchronized void newThreadPool(String namePrefix, int
poolSize) {
+ if (namePrefix == null || namePrefix.isEmpty()) {
+ throw new IllegalArgumentException("Thread pool name prefix must not be
empty");
+ }
+ if (poolSize <= 0) {
+ throw new IllegalArgumentException("Thread pool size must be greater
than 0");
+ }
+
+ ExecutorService existingPool = POOLS.get(namePrefix);
+ if (existingPool != null) {
+ int existingPoolSize = POOL_SIZES.get(namePrefix);
+ if (existingPoolSize != poolSize) {
+ LOG.warn(
+ "Iceberg thread pool {} is already initialized with size {} and
cannot be resized to {}; keeping the existing pool",
+ namePrefix,
+ existingPoolSize,
+ poolSize);
}
+ return;
}
- return commitExecutor;
+
+ ExecutorService executorService =
ThreadPools.newExitingWorkerPool(namePrefix, poolSize);
+ POOL_SIZES.put(namePrefix, poolSize);
+ POOLS.put(namePrefix, executorService);
}
}
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestIcebergThreadPools.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestIcebergThreadPools.java
new file mode 100644
index 000000000..a5bfb75c1
--- /dev/null
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestIcebergThreadPools.java
@@ -0,0 +1,65 @@
+/*
+ * 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.amoro.utils;
+
+import org.apache.iceberg.util.ThreadPools;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class TestIcebergThreadPools {
+
+ @Test
+ public void testDefaultPoolsAndInitialization() throws Exception {
+ ExecutorService workerPool = ThreadPools.getWorkerPool();
+ Assert.assertSame(workerPool, IcebergThreadPools.getPlanningExecutor());
+ Assert.assertSame(workerPool, IcebergThreadPools.getCommitExecutor());
+ Assert.assertSame(workerPool,
IcebergThreadPools.getThreadPool("unregistered-test-pool"));
+
+ IcebergThreadPools.newThreadPool("registered-test-pool", 1);
+ ExecutorService registeredPool =
IcebergThreadPools.getThreadPool("registered-test-pool");
+ Assert.assertNotSame(workerPool, registeredPool);
+ IcebergThreadPools.newThreadPool("registered-test-pool", 2);
+ Assert.assertSame(registeredPool,
IcebergThreadPools.getThreadPool("registered-test-pool"));
+
+ IcebergThreadPools.init(1, 1);
+
+ ExecutorService planningPool = IcebergThreadPools.getPlanningExecutor();
+ ExecutorService commitPool = IcebergThreadPools.getCommitExecutor();
+ Assert.assertNotSame(workerPool, planningPool);
+ Assert.assertNotSame(workerPool, commitPool);
+ Assert.assertNotSame(planningPool, commitPool);
+ Assert.assertTrue(
+ planningPool
+ .submit(() -> Thread.currentThread().getName())
+ .get(10, TimeUnit.SECONDS)
+ .startsWith("iceberg-planning-pool-"));
+ Assert.assertTrue(
+ commitPool
+ .submit(() -> Thread.currentThread().getName())
+ .get(10, TimeUnit.SECONDS)
+ .startsWith("iceberg-commit-pool-"));
+
+ IcebergThreadPools.init(2, 2);
+ Assert.assertSame(planningPool, IcebergThreadPools.getPlanningExecutor());
+ Assert.assertSame(commitPool, IcebergThreadPools.getCommitExecutor());
+ }
+}
