From 26127cd7d3fb4e7ed69cf571aa3ef4c3848da472 Mon Sep 17 00:00:00 2001 From: Daniel Roberts ddanielr Date: Wed, 9 Sep 2026 04:50:30 +0000 Subject: [PATCH 1/5] Generalize ConfigurationTypeHelper.getNumThreads getNumThreads was only being used by the BulkImport code and have a hardcoded default value. Refactored the method to pass in a default value to allow for more use in the code. Added test cases. --- .../core/clientImpl/bulk/BulkImport.java | 13 +++--- .../core/conf/ConfigurationTypeHelper.java | 11 +++-- .../conf/ConfigurationTypeHelperTest.java | 41 +++++++++++++++++++ 3 files changed, 56 insertions(+), 9 deletions(-) diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java index 575a221b8b6..8fca7b4b1d6 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java @@ -22,6 +22,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.MINUTES; import static java.util.stream.Collectors.groupingBy; +import static org.apache.accumulo.core.conf.ClientProperty.BULK_LOAD_THREADS; import static org.apache.accumulo.core.file.blockfile.impl.CachableBlockFile.pathToCacheId; import static org.apache.accumulo.core.util.Validators.EXISTING_TABLE_NAME; import static org.apache.accumulo.core.util.threads.ThreadPoolNames.BULK_IMPORT_CLIENT_BULK_THREADS_POOL; @@ -63,7 +64,6 @@ import org.apache.accumulo.core.clientImpl.bulk.Bulk.FileInfo; import org.apache.accumulo.core.clientImpl.bulk.Bulk.Files; import org.apache.accumulo.core.conf.AccumuloConfiguration; -import org.apache.accumulo.core.conf.ClientProperty; import org.apache.accumulo.core.conf.ConfigurationTypeHelper; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.crypto.CryptoFactoryLoader; @@ -507,11 +507,12 @@ private SortedMap computeMappingFromFiles(FileSystem fs, T executor = service = context.threadPools().getPoolBuilder(BULK_IMPORT_CLIENT_LOAD_POOL) .numCoreThreads(numThreads).enableThreadPoolMetrics().build(); } else { - String threads = context.getConfiguration().get(ClientProperty.BULK_LOAD_THREADS.getKey()); - executor = - service = context.threadPools().getPoolBuilder(BULK_IMPORT_CLIENT_BULK_THREADS_POOL) - .numCoreThreads(ConfigurationTypeHelper.getNumThreads(threads)) - .enableThreadPoolMetrics().build(); + String threads = context.getConfiguration().get(BULK_LOAD_THREADS.getKey()); + executor = service = context.threadPools() + .getPoolBuilder(BULK_IMPORT_CLIENT_BULK_THREADS_POOL) + .numCoreThreads( + ConfigurationTypeHelper.getNumThreads(threads, BULK_LOAD_THREADS.getDefaultValue())) + .enableThreadPoolMetrics().build(); } try { diff --git a/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java b/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java index ae0d4088e5c..6a6757dc4e5 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java @@ -28,6 +28,7 @@ import java.util.EnumSet; import java.util.HashMap; import java.util.Map; +import java.util.Objects; import java.util.concurrent.TimeUnit; import org.apache.accumulo.core.classloader.ClassLoaderUtil; @@ -211,9 +212,13 @@ public static T getClassInstance(String context, String clazzName, Class * Get the number of threads from string property. If the value ends with C, then it will be * multiplied by the number of cores. */ - public static int getNumThreads(String threads) { - if (threads == null) { - threads = ClientProperty.BULK_LOAD_THREADS.getDefaultValue(); + public static int getNumThreads(String threads, String defaultValue) { + Objects.requireNonNull(defaultValue, "The default thread value cannot be null"); + if (defaultValue.isBlank()) { + throw new IllegalArgumentException("The default thread value cannot be empty or blank"); + } + if (threads == null || threads.isBlank()) { + threads = defaultValue; } int nThreads; if (threads.toUpperCase().endsWith("C")) { diff --git a/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java b/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java index 66fdd51c6b2..cfe5721b8dc 100644 --- a/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java +++ b/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java @@ -149,4 +149,45 @@ public void testGetDropCacheBehindFilePrefixes() { () -> ConfigurationTypeHelper.getDropCacheBehindFilePrefixes("A")); } + + @Test + public void testGetNumberOfThreads() { + assertEquals(1, ConfigurationTypeHelper.getNumThreads("1", "4")); + assertEquals(2, ConfigurationTypeHelper.getNumThreads("", "2")); + assertEquals(3, ConfigurationTypeHelper.getNumThreads(null, "3")); + + // availableProcessors can return different results than physical core count on systems that use + // simultaneous multithreading (SMT). + int cores = Runtime.getRuntime().availableProcessors(); + assertEquals(cores * 2, ConfigurationTypeHelper.getNumThreads("2C", "1")); + assertEquals(cores * 3, ConfigurationTypeHelper.getNumThreads(null, "3C")); + } + + @Test + public void testGetNumberOfThreadsNullDefaultFailure() { + Exception exception = assertThrows(NullPointerException.class, + () -> ConfigurationTypeHelper.getNumThreads("", null)); + assertEquals("The default thread value cannot be null", exception.getMessage()); + } + + @Test + public void testGetNumberOfThreadsEmptyDefaultFailure() { + Exception exception = assertThrows(IllegalArgumentException.class, + () -> ConfigurationTypeHelper.getNumThreads(null, "")); + assertEquals("The default thread value cannot be empty or blank", exception.getMessage()); + } + + @Test + public void testGetNumberOfThreadsBlankDefaultFailure() { + Exception exception = assertThrows(IllegalArgumentException.class, + () -> ConfigurationTypeHelper.getNumThreads(null, " ")); + assertEquals("The default thread value cannot be empty or blank", exception.getMessage()); + } + + @Test + public void testGetNumberOfThreadsDecimalFailure() { + Exception exception = assertThrows(NumberFormatException.class, + () -> ConfigurationTypeHelper.getNumThreads("0.25C", "1")); + assertEquals("For input string: \"0.25\"", exception.getMessage()); + } } From 5914ec51de89ef01c219142f7e2d639519e6c0fd Mon Sep 17 00:00:00 2001 From: Daniel Roberts ddanielr Date: Wed, 9 Sep 2026 05:38:47 +0000 Subject: [PATCH 2/5] Adds property for selector threads Allows the selector thread counts to be set via property value. This property was added under the `general.rpc` prefix as it can be different across instances in the same accumulo cluster. --- .../main/java/org/apache/accumulo/core/conf/Property.java | 5 +++++ .../java/org/apache/accumulo/server/rpc/TServerUtils.java | 8 ++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java index eaaf8572d54..1fca85caf82 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java @@ -261,6 +261,11 @@ public enum Property { + "org.apache.accumulo.server.rpc.ThriftServerType for more information. " + "Only useful for benchmarking thrift servers.", "1.7.0"), + GENERAL_RPC_SERVER_SELECTOR_THREADS("general.rpc.server.selector.threads", "2", + PropertyType.COUNT, + "The number of selector threads to use for the TThreadedSelectorServer. " + + " If the value ends with C, then it will be multiplied by the number of cores on the system.", + "2.1.7"), GENERAL_KERBEROS_KEYTAB("general.kerberos.keytab", "", PropertyType.PATH, "Path to the kerberos keytab to use. Leave blank if not using kerberoized hdfs.", "1.4.1"), GENERAL_KERBEROS_PRINCIPAL("general.kerberos.principal", "", PropertyType.STRING, diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java index 7c34b6601d6..fc439c7c8d2 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java +++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java @@ -20,6 +20,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.accumulo.core.conf.Property.GENERAL_RPC_SERVER_SELECTOR_THREADS; import static org.apache.accumulo.core.util.threads.ThreadPoolNames.ACCUMULO_POOL_PREFIX; import java.io.IOException; @@ -40,6 +41,7 @@ import javax.net.ssl.SSLServerSocket; import org.apache.accumulo.core.conf.AccumuloConfiguration; +import org.apache.accumulo.core.conf.ConfigurationTypeHelper; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.conf.PropertyType; import org.apache.accumulo.core.conf.PropertyType.PortRange; @@ -238,8 +240,10 @@ private static ServerAddress createThreadedSelectorServer(HostAndPort address, TThreadedSelectorServer.Args options = new TThreadedSelectorServer.Args(transport); - options.selectorThreads = Math.max(2, Runtime.getRuntime().availableProcessors() / 4); - log.info("selectorThreads : " + options.selectorThreads); + String threads = conf.get(GENERAL_RPC_SERVER_SELECTOR_THREADS); + options.selectorThreads = ConfigurationTypeHelper.getNumThreads(threads, + GENERAL_RPC_SERVER_SELECTOR_THREADS.getDefaultValue()); + log.info("selectorThreads : {}", options.selectorThreads); options.protocolFactory(protocolFactory); options.transportFactory(ThriftUtil.transportFactory(maxMessageSize)); options.maxReadBufferBytes = maxMessageSize; From 72f453b1f601e4b933103e81806c9884374cbc29 Mon Sep 17 00:00:00 2001 From: Daniel Roberts Date: Wed, 9 Sep 2026 09:33:06 -0400 Subject: [PATCH 3/5] Update core/src/main/java/org/apache/accumulo/core/conf/Property.java Co-authored-by: Dave Marion --- core/src/main/java/org/apache/accumulo/core/conf/Property.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java index 1fca85caf82..7e039e74f03 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java @@ -261,7 +261,7 @@ public enum Property { + "org.apache.accumulo.server.rpc.ThriftServerType for more information. " + "Only useful for benchmarking thrift servers.", "1.7.0"), - GENERAL_RPC_SERVER_SELECTOR_THREADS("general.rpc.server.selector.threads", "2", + GENERAL_RPC_SERVER_SELECTOR_THREADS("general.rpc.server.threaded.selector.threads", "2", PropertyType.COUNT, "The number of selector threads to use for the TThreadedSelectorServer. " + " If the value ends with C, then it will be multiplied by the number of cores on the system.", From 9866cbf42fb11854fc7cf661ebf4abad55fe5b64 Mon Sep 17 00:00:00 2001 From: Daniel Roberts ddanielr Date: Thu, 10 Sep 2026 16:07:43 +0000 Subject: [PATCH 4/5] Rely on behavior of get(). Still keep value checks. Remove the needless default value passing and rely on the behavior of configuration.get(). Keep the value checks so an invalid config results in process failure. --- .../core/clientImpl/bulk/BulkImport.java | 9 +++-- .../core/conf/ConfigurationTypeHelper.java | 16 ++++----- .../conf/ConfigurationTypeHelperTest.java | 34 ++++++++++--------- .../accumulo/server/rpc/TServerUtils.java | 3 +- 4 files changed, 31 insertions(+), 31 deletions(-) diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java index 8fca7b4b1d6..f94a7d37f50 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java @@ -508,11 +508,10 @@ private SortedMap computeMappingFromFiles(FileSystem fs, T .numCoreThreads(numThreads).enableThreadPoolMetrics().build(); } else { String threads = context.getConfiguration().get(BULK_LOAD_THREADS.getKey()); - executor = service = context.threadPools() - .getPoolBuilder(BULK_IMPORT_CLIENT_BULK_THREADS_POOL) - .numCoreThreads( - ConfigurationTypeHelper.getNumThreads(threads, BULK_LOAD_THREADS.getDefaultValue())) - .enableThreadPoolMetrics().build(); + executor = + service = context.threadPools().getPoolBuilder(BULK_IMPORT_CLIENT_BULK_THREADS_POOL) + .numCoreThreads(ConfigurationTypeHelper.getNumThreads(threads)) + .enableThreadPoolMetrics().build(); } try { diff --git a/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java b/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java index 6a6757dc4e5..a15e02f6f18 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java @@ -209,16 +209,13 @@ public static T getClassInstance(String context, String clazzName, Class } /** - * Get the number of threads from string property. If the value ends with C, then it will be + * Get the number of threads from a string property. If the value ends with C, then it will be * multiplied by the number of cores. */ - public static int getNumThreads(String threads, String defaultValue) { - Objects.requireNonNull(defaultValue, "The default thread value cannot be null"); - if (defaultValue.isBlank()) { - throw new IllegalArgumentException("The default thread value cannot be empty or blank"); - } - if (threads == null || threads.isBlank()) { - threads = defaultValue; + public static int getNumThreads(String threads) { + Objects.requireNonNull(threads, "Threads value cannot be null"); + if (threads.isBlank()) { + throw new IllegalArgumentException("Threads value cannot be empty or blank"); } int nThreads; if (threads.toUpperCase().endsWith("C")) { @@ -227,6 +224,9 @@ public static int getNumThreads(String threads, String defaultValue) { } else { nThreads = Integer.parseInt(threads); } + if (nThreads < 1) { + throw new IllegalArgumentException("Threads value cannot be less than 1"); + } return nThreads; } diff --git a/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java b/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java index cfe5721b8dc..be3be89f33e 100644 --- a/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java +++ b/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java @@ -152,42 +152,44 @@ public void testGetDropCacheBehindFilePrefixes() { @Test public void testGetNumberOfThreads() { - assertEquals(1, ConfigurationTypeHelper.getNumThreads("1", "4")); - assertEquals(2, ConfigurationTypeHelper.getNumThreads("", "2")); - assertEquals(3, ConfigurationTypeHelper.getNumThreads(null, "3")); + assertEquals(1, ConfigurationTypeHelper.getNumThreads("1")); + assertEquals(2, ConfigurationTypeHelper.getNumThreads("2")); + assertEquals(20, ConfigurationTypeHelper.getNumThreads("20")); // availableProcessors can return different results than physical core count on systems that use // simultaneous multithreading (SMT). int cores = Runtime.getRuntime().availableProcessors(); - assertEquals(cores * 2, ConfigurationTypeHelper.getNumThreads("2C", "1")); - assertEquals(cores * 3, ConfigurationTypeHelper.getNumThreads(null, "3C")); + assertEquals(cores * 2, ConfigurationTypeHelper.getNumThreads("2C")); } @Test - public void testGetNumberOfThreadsNullDefaultFailure() { - Exception exception = assertThrows(NullPointerException.class, - () -> ConfigurationTypeHelper.getNumThreads("", null)); - assertEquals("The default thread value cannot be null", exception.getMessage()); + public void testGetNumberOfThreadsNullFailure() { + Exception exception = + assertThrows(NullPointerException.class, () -> ConfigurationTypeHelper.getNumThreads(null)); + assertEquals("Threads value cannot be null", exception.getMessage()); } @Test - public void testGetNumberOfThreadsEmptyDefaultFailure() { + public void testGetNumberOfThreadsEmptyFailure() { Exception exception = assertThrows(IllegalArgumentException.class, - () -> ConfigurationTypeHelper.getNumThreads(null, "")); - assertEquals("The default thread value cannot be empty or blank", exception.getMessage()); + () -> ConfigurationTypeHelper.getNumThreads("")); + assertEquals("Threads value cannot be empty or blank", exception.getMessage()); } @Test - public void testGetNumberOfThreadsBlankDefaultFailure() { + public void testGetNumberOfThreadsNegativeOrZeroFailure() { Exception exception = assertThrows(IllegalArgumentException.class, - () -> ConfigurationTypeHelper.getNumThreads(null, " ")); - assertEquals("The default thread value cannot be empty or blank", exception.getMessage()); + () -> ConfigurationTypeHelper.getNumThreads("-1")); + assertEquals("Threads value cannot be less than 1", exception.getMessage()); + exception = assertThrows(IllegalArgumentException.class, + () -> ConfigurationTypeHelper.getNumThreads("0")); + assertEquals("Threads value cannot be less than 1", exception.getMessage()); } @Test public void testGetNumberOfThreadsDecimalFailure() { Exception exception = assertThrows(NumberFormatException.class, - () -> ConfigurationTypeHelper.getNumThreads("0.25C", "1")); + () -> ConfigurationTypeHelper.getNumThreads("0.25C")); assertEquals("For input string: \"0.25\"", exception.getMessage()); } } diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java index fc439c7c8d2..06949a5c9a9 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java +++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java @@ -241,8 +241,7 @@ private static ServerAddress createThreadedSelectorServer(HostAndPort address, TThreadedSelectorServer.Args options = new TThreadedSelectorServer.Args(transport); String threads = conf.get(GENERAL_RPC_SERVER_SELECTOR_THREADS); - options.selectorThreads = ConfigurationTypeHelper.getNumThreads(threads, - GENERAL_RPC_SERVER_SELECTOR_THREADS.getDefaultValue()); + options.selectorThreads = ConfigurationTypeHelper.getNumThreads(threads); log.info("selectorThreads : {}", options.selectorThreads); options.protocolFactory(protocolFactory); options.transportFactory(ThriftUtil.transportFactory(maxMessageSize)); From 0c42ec8049c066c5fb2ceedb4e2ff70466957913 Mon Sep 17 00:00:00 2001 From: Daniel Roberts ddanielr Date: Thu, 10 Sep 2026 17:22:24 +0000 Subject: [PATCH 5/5] Remove unneeded vars --- .../org/apache/accumulo/core/clientImpl/bulk/BulkImport.java | 4 ++-- .../java/org/apache/accumulo/server/rpc/TServerUtils.java | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java index f94a7d37f50..be4963c7631 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java @@ -507,10 +507,10 @@ private SortedMap computeMappingFromFiles(FileSystem fs, T executor = service = context.threadPools().getPoolBuilder(BULK_IMPORT_CLIENT_LOAD_POOL) .numCoreThreads(numThreads).enableThreadPoolMetrics().build(); } else { - String threads = context.getConfiguration().get(BULK_LOAD_THREADS.getKey()); executor = service = context.threadPools().getPoolBuilder(BULK_IMPORT_CLIENT_BULK_THREADS_POOL) - .numCoreThreads(ConfigurationTypeHelper.getNumThreads(threads)) + .numCoreThreads(ConfigurationTypeHelper + .getNumThreads(context.getConfiguration().get(BULK_LOAD_THREADS.getKey()))) .enableThreadPoolMetrics().build(); } diff --git a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java index 06949a5c9a9..48582eb5ab2 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java +++ b/server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java @@ -240,8 +240,8 @@ private static ServerAddress createThreadedSelectorServer(HostAndPort address, TThreadedSelectorServer.Args options = new TThreadedSelectorServer.Args(transport); - String threads = conf.get(GENERAL_RPC_SERVER_SELECTOR_THREADS); - options.selectorThreads = ConfigurationTypeHelper.getNumThreads(threads); + options.selectorThreads = + ConfigurationTypeHelper.getNumThreads(conf.get(GENERAL_RPC_SERVER_SELECTOR_THREADS)); log.info("selectorThreads : {}", options.selectorThreads); options.protocolFactory(protocolFactory); options.transportFactory(ThriftUtil.transportFactory(maxMessageSize));