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..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 @@ -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,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(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 ae0d4088e5c..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 @@ -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; @@ -208,12 +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) { - 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 static int getNumThreads(String threads) { } 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 eaaf8572d54..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,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 66fdd51c6b2..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 @@ -149,4 +149,47 @@ public void testGetDropCacheBehindFilePrefixes() { () -> 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 7c34b6601d6..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 @@ -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,9 @@ 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); + 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;