Make Edge query and load-event thread pools configurable (#18584)

(cherry picked from commit 9d91245)

Author: JackieTien97

integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java

@@ -58,6 +58,10 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -180,11 +184,89 @@ public void testTableModelReadWrite() throws SQLException {
     }
   }
 
+  @Test
+  public void testConcurrentTableQueriesWithSmallThreadPools() throws Exception {
+    try (Connection connection = openTableConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute("CREATE DATABASE edge_it_concurrent");
+      statement.execute("USE edge_it_concurrent");
+      statement.execute("CREATE TABLE sensor(device STRING TAG, value INT32 FIELD)");
+      statement.execute("INSERT INTO sensor(time,device,value) VALUES (1,'d1',42), (2,'d1',84)");
+    }
+
+    ExecutorService executor = Executors.newFixedThreadPool(4);
+    CountDownLatch ready = new CountDownLatch(4);
+    CountDownLatch start = new CountDownLatch(1);
+    List<Future<Void>> queries = new ArrayList<>();
+    try {
+      for (int i = 0; i < 4; i++) {
+        queries.add(
+            executor.submit(
+                () -> {
+                  try (Connection connection = openTableConnection();
+                      Statement statement = connection.createStatement()) {
+                    statement.execute("USE edge_it_concurrent");
+                    ready.countDown();
+                    assertTrue(start.await(30, TimeUnit.SECONDS));
+                    for (int iteration = 0; iteration < 20; iteration++) {
+                      try (ResultSet result =
+                          statement.executeQuery("SELECT sum(value) FROM sensor")) {
+                        assertTrue(result.next());
+                        assertEquals(126.0, result.getDouble(1), 0.0);
+                        assertFalse(result.next());
+                      }
+                      try (ResultSet result =
+                          statement.executeQuery(
+                              "SELECT value FROM sensor WHERE device='d1' ORDER BY time")) {
+                        assertTrue(result.next());
+                        assertEquals(42, result.getInt(1));
+                        assertTrue(result.next());
+                        assertEquals(84, result.getInt(1));
+                        assertFalse(result.next());
+                      }
+                    }
+                  }
+                  return null;
+                }));
+      }
+      assertTrue(ready.await(30, TimeUnit.SECONDS));
+      start.countDown();
+      for (Future<Void> query : queries) {
+        query.get(60, TimeUnit.SECONDS);
+      }
+    } finally {
+      start.countDown();
+      executor.shutdownNow();
+      assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+    }
+  }
+
   private static Connection openTreeConnection() throws SQLException {
     return DriverManager.getConnection(
         jdbcUrl(), SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD);
   }
 
+  @Test
+  public void testRatisMetadataConsensus() throws SQLException {
+    Map<String, String> variables = new LinkedHashMap<>();
+    try (Connection connection = openTableConnection();
+        Statement statement = connection.createStatement();
+        ResultSet result = statement.executeQuery("SHOW VARIABLES")) {
+      while (result.next()) {
+        variables.put(result.getString(1), result.getString(2));
+      }
+    }
+    assertEquals(
+        "org.apache.iotdb.consensus.ratis.RatisConsensus",
+        variables.get("ConfigNodeConsensusProtocolClass"));
+    assertEquals(
+        "org.apache.iotdb.consensus.ratis.RatisConsensus",
+        variables.get("SchemaRegionConsensusProtocolClass"));
+    assertEquals(
+        "org.apache.iotdb.consensus.iot.IoTConsensus",
+        variables.get("DataRegionConsensusProtocolClass"));
+  }
+
   private static Connection openTableConnection() throws SQLException {
     return DriverManager.getConnection(
         jdbcUrl() + "?sql_dialect=table",
@@ -195,6 +277,10 @@ private static Connection openTableConnection() throws SQLException {
   @Test
   public void testPackagedConfiguration() throws Exception {
     assertFalse(PACKAGED_SYSTEM_PROPERTIES.containsKey("model_inference_execution_thread_count"));
+    assertEdgeProperty("coordinator_read_executor_size", "2");
+    assertEdgeProperty("coordinator_scheduled_executor_size", "2");
+    assertEdgeProperty("fragment_instance_notification_thread_count", "2");
+    assertEdgeProperty("cn_load_statistics_publisher_thread_count", "1");
     assertEdgeProperty("candidate_compaction_task_queue_size", "10");
     assertEdgeProperty("compaction_max_aligned_series_num_in_one_batch", "2");
     assertEdgeProperty("target_compaction_file_size", "33554432");

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java

@@ -202,6 +202,9 @@ public class ConfigNodeConfig {
   private int procedureCoreWorkerThreadsCount =
       Math.max(Runtime.getRuntime().availableProcessors() / 4, 16);
 
+  /** Thread pool size for publishing cluster load statistics changes. */
+  private int loadStatisticsPublisherThreadCount = 5;
+
   /** The heartbeat interval in milliseconds. */
   private volatile long heartbeatIntervalInMs = 1000;
 
@@ -738,6 +741,17 @@ public void setProcedureCoreWorkerThreadsCount(int procedureCoreWorkerThreadsCou
     this.procedureCoreWorkerThreadsCount = procedureCoreWorkerThreadsCount;
   }
 
+  public int getLoadStatisticsPublisherThreadCount() {
+    return loadStatisticsPublisherThreadCount;
+  }
+
+  public void setLoadStatisticsPublisherThreadCount(int loadStatisticsPublisherThreadCount) {
+    if (loadStatisticsPublisherThreadCount <= 0) {
+      throw new IllegalArgumentException(CommonMessages.SIZE_MUST_BE_POSITIVE);
+    }
+    this.loadStatisticsPublisherThreadCount = loadStatisticsPublisherThreadCount;
+  }
+
   public long getHeartbeatIntervalInMs() {
     return heartbeatIntervalInMs;
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java

@@ -402,6 +402,12 @@ private void loadProperties(TrimProperties properties) throws BadNodeUrlExceptio
                 "procedure_core_worker_thread_count",
                 String.valueOf(conf.getProcedureCoreWorkerThreadsCount()))));
 
+    conf.setLoadStatisticsPublisherThreadCount(
+        Integer.parseInt(
+            properties.getProperty(
+                "cn_load_statistics_publisher_thread_count",
+                String.valueOf(conf.getLoadStatisticsPublisherThreadCount()))));
+
     loadRatisConsensusConfig(properties);
     loadCQConfig(properties);
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java

@@ -82,7 +82,10 @@ public EventService(LoadCache loadCache) {
         new AsyncEventBus(
             ThreadName.CONFIG_NODE_LOAD_PUBLISHER.getName(),
             IoTDBThreadPoolFactory.newFixedThreadPool(
-                5, ThreadName.CONFIG_NODE_LOAD_PUBLISHER.getName()));
+                ConfigNodeDescriptor.getInstance()
+                    .getConf()
+                    .getLoadStatisticsPublisherThreadCount(),
+                ThreadName.CONFIG_NODE_LOAD_PUBLISHER.getName()));
   }
 
   public void register(final IClusterStatusSubscriber listener) {

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/LoadStatisticsPublisherConfigTest.java

@@ -0,0 +1,85 @@
+/*
+ * 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.iotdb.confignode.conf;
+
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.ConfigurationFileUtils;
+import org.apache.iotdb.commons.conf.TrimProperties;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.lang.reflect.Constructor;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+public class LoadStatisticsPublisherConfigTest {
+
+  @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Test
+  public void testDefaultsAndPositiveSize() throws Exception {
+    ConfigNodeConfig config = new ConfigNodeConfig();
+    assertEquals(5, config.getLoadStatisticsPublisherThreadCount());
+    assertEquals(
+        "5",
+        ConfigurationFileUtils.getConfigurationDefaultValue(
+            "cn_load_statistics_publisher_thread_count"));
+    for (int invalid : new int[] {0, -1}) {
+      assertThrows(
+          IllegalArgumentException.class,
+          () -> config.setLoadStatisticsPublisherThreadCount(invalid));
+    }
+    assertEquals(5, config.getLoadStatisticsPublisherThreadCount());
+  }
+
+  @Test
+  public void testStartupOverrideIsRestartOnly() throws Exception {
+    String originalConf = System.getProperty(ConfigNodeConstant.CONFIGNODE_CONF);
+    File confDir = temporaryFolder.newFolder();
+    Files.writeString(
+        confDir.toPath().resolve(CommonConfig.SYSTEM_CONFIG_NAME),
+        "cn_seed_config_node=127.0.0.1:10710\ncn_load_statistics_publisher_thread_count=2\n",
+        StandardCharsets.UTF_8);
+    System.setProperty(ConfigNodeConstant.CONFIGNODE_CONF, confDir.getAbsolutePath());
+    try {
+      Constructor<ConfigNodeDescriptor> constructor =
+          ConfigNodeDescriptor.class.getDeclaredConstructor();
+      constructor.setAccessible(true);
+      ConfigNodeDescriptor descriptor = constructor.newInstance();
+      assertEquals(2, descriptor.getConf().getLoadStatisticsPublisherThreadCount());
+
+      TrimProperties properties = new TrimProperties();
+      properties.setProperty("cn_load_statistics_publisher_thread_count", "3");
+      descriptor.loadHotModifiedProps(properties);
+      assertEquals(2, descriptor.getConf().getLoadStatisticsPublisherThreadCount());
+    } finally {
+      if (originalConf == null) {
+        System.clearProperty(ConfigNodeConstant.CONFIGNODE_CONF);
+      } else {
+        System.setProperty(ConfigNodeConstant.CONFIGNODE_CONF, originalConf);
+      }
+    }
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.conf.IoTDBConstant;
 import org.apache.iotdb.commons.enums.ReadConsistencyLevel;
+import org.apache.iotdb.commons.i18n.CommonMessages;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.consensus.ConsensusFactory;
@@ -1002,6 +1003,12 @@ public class IoTDBConfig {
   /** ThreadPool size for read operation in coordinator */
   private int coordinatorReadExecutorSize = 20;
 
+  /** Thread pool size for scheduling query state checks and termination. */
+  private int coordinatorScheduledExecutorSize = 10;
+
+  /** Thread pool size for fragment instance state change notifications. */
+  private int fragmentInstanceNotificationThreadCount = 4;
+
   /** Policy of DataNodeSchemaCache eviction */
   private String dataNodeSchemaCacheEvictionPolicy = "FIFO";
 
@@ -3479,9 +3486,35 @@ public int getCoordinatorReadExecutorSize() {
   }
 
   public void setCoordinatorReadExecutorSize(int coordinatorReadExecutorSize) {
+    if (coordinatorReadExecutorSize <= 0) {
+      throw new IllegalArgumentException(CommonMessages.SIZE_MUST_BE_POSITIVE);
+    }
     this.coordinatorReadExecutorSize = coordinatorReadExecutorSize;
   }
 
+  public int getCoordinatorScheduledExecutorSize() {
+    return coordinatorScheduledExecutorSize;
+  }
+
+  public void setCoordinatorScheduledExecutorSize(int coordinatorScheduledExecutorSize) {
+    if (coordinatorScheduledExecutorSize <= 0) {
+      throw new IllegalArgumentException(CommonMessages.SIZE_MUST_BE_POSITIVE);
+    }
+    this.coordinatorScheduledExecutorSize = coordinatorScheduledExecutorSize;
+  }
+
+  public int getFragmentInstanceNotificationThreadCount() {
+    return fragmentInstanceNotificationThreadCount;
+  }
+
+  public void setFragmentInstanceNotificationThreadCount(
+      int fragmentInstanceNotificationThreadCount) {
+    if (fragmentInstanceNotificationThreadCount <= 0) {
+      throw new IllegalArgumentException(CommonMessages.SIZE_MUST_BE_POSITIVE);
+    }
+    this.fragmentInstanceNotificationThreadCount = fragmentInstanceNotificationThreadCount;
+  }
+
   public TEndPoint getAddressAndPort() {
     return new TEndPoint(rpcAddress, rpcPort);
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -988,6 +988,16 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException
             properties.getProperty(
                 "coordinator_read_executor_size",
                 Integer.toString(conf.getCoordinatorReadExecutorSize()))));
+    conf.setCoordinatorScheduledExecutorSize(
+        Integer.parseInt(
+            properties.getProperty(
+                "coordinator_scheduled_executor_size",
+                Integer.toString(conf.getCoordinatorScheduledExecutorSize()))));
+    conf.setFragmentInstanceNotificationThreadCount(
+        Integer.parseInt(
+            properties.getProperty(
+                "fragment_instance_notification_thread_count",
+                Integer.toString(conf.getFragmentInstanceNotificationThreadCount()))));
     conf.setDataNodeTableSchemaCacheSize(
         Long.parseLong(
             properties.getProperty(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java

@@ -110,7 +110,8 @@ private FragmentInstanceManager() {
             1, ThreadName.FRAGMENT_INSTANCE_MANAGEMENT.getName());
     this.instanceNotificationExecutor =
         IoTDBThreadPoolFactory.newFixedThreadPool(
-            4, ThreadName.FRAGMENT_INSTANCE_NOTIFICATION.getName());
+            IoTDBDescriptor.getInstance().getConfig().getFragmentInstanceNotificationThreadCount(),
+            ThreadName.FRAGMENT_INSTANCE_NOTIFICATION.getName());
 
     this.infoCacheTime = new Duration(5, TimeUnit.MINUTES);
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java

@@ -200,7 +200,6 @@
 public class Coordinator {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(Coordinator.class);
-  private static final int COORDINATOR_SCHEDULED_EXECUTOR_SIZE = 10;
   private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
   private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig();
 
@@ -856,7 +855,7 @@ private ExecutorService getQueryExecutor() {
 
   private ScheduledExecutorService getScheduledExecutor() {
     return IoTDBThreadPoolFactory.newScheduledThreadPool(
-        COORDINATOR_SCHEDULED_EXECUTOR_SIZE,
+        CONFIG.getCoordinatorScheduledExecutorSize(),
         ThreadName.MPP_COORDINATOR_SCHEDULED_EXECUTOR.getName());
   }