Adds property for Thrift Selector threads

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<KeyExtent,Bulk.Files> 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();
     }
 

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> T getClassInstance(String context, String clazzName, Class<T>
   }
 
   /**
-   * 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;
   }
 

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,

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());
+  }
 }

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;