This is an automated email from the ASF dual-hosted git repository.
ddanielr pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new db73ec6bd6 Adds property for Thrift Selector threads (#6529)
db73ec6bd6 is described below
commit db73ec6bd6a1cbf5263f3178669cd0967d553956
Author: Daniel Roberts <[email protected]>
AuthorDate: Thu Sep 10 15:05:23 2026 -0400
Adds property for Thrift Selector threads (#6529)
* 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.
* 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.
* Update core/src/main/java/org/apache/accumulo/core/conf/Property.java
Co-authored-by: Dave Marion <[email protected]>
* 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.
* Remove unneeded vars
---------
Co-authored-by: Dave Marion <[email protected]>
---
.../accumulo/core/clientImpl/bulk/BulkImport.java | 6 +--
.../core/conf/ConfigurationTypeHelper.java | 11 ++++--
.../org/apache/accumulo/core/conf/Property.java | 5 +++
.../core/conf/ConfigurationTypeHelperTest.java | 43 ++++++++++++++++++++++
.../apache/accumulo/server/rpc/TServerUtils.java | 7 +++-
5 files changed, 64 insertions(+), 8 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 575a221b8b..be4963c763 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.nio.charset.StandardCharsets.UTF_8;
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.TableOperationsImpl;
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,10 +507,10 @@ public class BulkImport implements
ImportDestinationArguments, ImportMappingOpti
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))
+ .numCoreThreads(ConfigurationTypeHelper
+
.getNumThreads(context.getConfiguration().get(BULK_LOAD_THREADS.getKey())))
.enableThreadPoolMetrics().build();
}
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 ae0d4088e5..a15e02f6f1 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.Collections;
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;
@@ -208,12 +209,13 @@ public class ConfigurationTypeHelper {
}
/**
- * 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) {
- if (threads == null) {
- threads = ClientProperty.BULK_LOAD_THREADS.getDefaultValue();
+ 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")) {
@@ -222,6 +224,9 @@ public class ConfigurationTypeHelper {
} 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/main/java/org/apache/accumulo/core/conf/Property.java
b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index eaaf8572d5..7e039e74f0 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.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.",
+ "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/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java
b/core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java
index 66fdd51c6b..be3be89f33 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,47 @@ public class ConfigurationTypeHelperTest {
() -> ConfigurationTypeHelper.getDropCacheBehindFilePrefixes("A"));
}
+
+ @Test
+ public void testGetNumberOfThreads() {
+ 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"));
+ }
+
+ @Test
+ public void testGetNumberOfThreadsNullFailure() {
+ Exception exception =
+ assertThrows(NullPointerException.class, () ->
ConfigurationTypeHelper.getNumThreads(null));
+ assertEquals("Threads value cannot be null", exception.getMessage());
+ }
+
+ @Test
+ public void testGetNumberOfThreadsEmptyFailure() {
+ Exception exception = assertThrows(IllegalArgumentException.class,
+ () -> ConfigurationTypeHelper.getNumThreads(""));
+ assertEquals("Threads value cannot be empty or blank",
exception.getMessage());
+ }
+
+ @Test
+ public void testGetNumberOfThreadsNegativeOrZeroFailure() {
+ Exception exception = assertThrows(IllegalArgumentException.class,
+ () -> 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"));
+ 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 7c34b6601d..48582eb5ab 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 @@ package org.apache.accumulo.server.rpc;
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 java.util.stream.IntStream;
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,9 @@ public class TServerUtils {
TThreadedSelectorServer.Args options = new
TThreadedSelectorServer.Args(transport);
- options.selectorThreads = Math.max(2,
Runtime.getRuntime().availableProcessors() / 4);
- log.info("selectorThreads : " + options.selectorThreads);
+ options.selectorThreads =
+
ConfigurationTypeHelper.getNumThreads(conf.get(GENERAL_RPC_SERVER_SELECTOR_THREADS));
+ log.info("selectorThreads : {}", options.selectorThreads);
options.protocolFactory(protocolFactory);
options.transportFactory(ThriftUtil.transportFactory(maxMessageSize));
options.maxReadBufferBytes = maxMessageSize;
