From 1733dfe982442cafe8fda3c2dd490189b9abe367 Mon Sep 17 00:00:00 2001 From: JackieTien97 Date: Sat, 5 Sep 2026 12:44:59 +0800 Subject: [PATCH] Make Edge query and load-event thread pools configurable --- .../iotdb/edge/it/IoTDBEdgeBasicIT.java | 86 +++++++++++++++++++ .../confignode/conf/ConfigNodeConfig.java | 14 +++ .../confignode/conf/ConfigNodeDescriptor.java | 6 ++ .../manager/load/service/EventService.java | 5 +- .../LoadStatisticsPublisherConfigTest.java | 85 ++++++++++++++++++ .../org/apache/iotdb/db/conf/IoTDBConfig.java | 33 +++++++ .../apache/iotdb/db/conf/IoTDBDescriptor.java | 10 +++ .../fragment/FragmentInstanceManager.java | 3 +- .../db/queryengine/plan/Coordinator.java | 3 +- .../db/conf/QueryThreadPoolConfigTest.java | 81 +++++++++++++++++ .../conf/edge/iotdb-system.properties | 6 +- .../conf/iotdb-system.properties.template | 20 +++++ 12 files changed, 347 insertions(+), 5 deletions(-) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/LoadStatisticsPublisherConfigTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java b/integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java index 003ed743eb325..57a8189aa8425 100644 --- a/integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java +++ b/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> 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 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 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"); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java index c1baa98209c23..16a183ecf1ed6 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java +++ b/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; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java index b4e8b6f9d8914..e9ab7657c50a6 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java +++ b/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); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java index 5974e5ceb58c4..bbb807c51d001 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java +++ b/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) { diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/LoadStatisticsPublisherConfigTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/LoadStatisticsPublisherConfigTest.java new file mode 100644 index 0000000000000..37fa62059b239 --- /dev/null +++ b/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 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); + } + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 2f0f284b5c0ac..b14f2e1b6222b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/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; @@ -1019,6 +1020,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"; @@ -3576,9 +3583,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); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 375c274667312..44971c718a104 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -1002,6 +1002,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( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java index 342dec0a7bab7..fa6a45e837496 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java @@ -111,7 +111,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); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java index 8919182c7ce93..09a35803595f8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java @@ -201,7 +201,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(); @@ -873,7 +872,7 @@ private ExecutorService getQueryExecutor() { private ScheduledExecutorService getScheduledExecutor() { return IoTDBThreadPoolFactory.newScheduledThreadPool( - COORDINATOR_SCHEDULED_EXECUTOR_SIZE, + CONFIG.getCoordinatorScheduledExecutorSize(), ThreadName.MPP_COORDINATOR_SCHEDULED_EXECUTOR.getName()); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java new file mode 100644 index 0000000000000..92b6d0fd45ffa --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java @@ -0,0 +1,81 @@ +/* + * 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.db.conf; + +import org.apache.iotdb.commons.conf.ConfigurationFileUtils; +import org.apache.iotdb.commons.conf.TrimProperties; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +public class QueryThreadPoolConfigTest { + + @Test + public void testDefaultsAndPositiveSizes() throws Exception { + IoTDBConfig config = new IoTDBConfig(); + assertSizes(config, 20, 10, 4); + assertEquals( + "20", + ConfigurationFileUtils.getConfigurationDefaultValue("coordinator_read_executor_size")); + assertEquals( + "10", + ConfigurationFileUtils.getConfigurationDefaultValue("coordinator_scheduled_executor_size")); + assertEquals( + "4", + ConfigurationFileUtils.getConfigurationDefaultValue( + "fragment_instance_notification_thread_count")); + + for (int invalid : new int[] {0, -1}) { + assertThrows( + IllegalArgumentException.class, () -> config.setCoordinatorReadExecutorSize(invalid)); + assertThrows( + IllegalArgumentException.class, + () -> config.setCoordinatorScheduledExecutorSize(invalid)); + assertThrows( + IllegalArgumentException.class, + () -> config.setFragmentInstanceNotificationThreadCount(invalid)); + } + assertSizes(config, 20, 10, 4); + } + + @Test + public void testStartupOverridesAreRestartOnly() throws Exception { + IoTDBDescriptor descriptor = new IoTDBDescriptor(); + TrimProperties properties = new TrimProperties(); + properties.setProperty("coordinator_read_executor_size", "3"); + properties.setProperty("coordinator_scheduled_executor_size", "2"); + properties.setProperty("fragment_instance_notification_thread_count", "1"); + descriptor.loadProperties(properties); + assertSizes(descriptor.getConfig(), 3, 2, 1); + + properties.setProperty("coordinator_read_executor_size", "6"); + properties.setProperty("coordinator_scheduled_executor_size", "5"); + properties.setProperty("fragment_instance_notification_thread_count", "4"); + descriptor.loadHotModifiedProps(properties); + assertSizes(descriptor.getConfig(), 3, 2, 1); + } + + private static void assertSizes(IoTDBConfig config, int read, int scheduled, int notification) { + assertEquals(read, config.getCoordinatorReadExecutorSize()); + assertEquals(scheduled, config.getCoordinatorScheduledExecutorSize()); + assertEquals(notification, config.getFragmentInstanceNotificationThreadCount()); + } +} diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties index 01cac68af5223..47a67d1cf3860 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties +++ b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties @@ -79,6 +79,10 @@ dn_metric_prometheus_reporter_port=9092 # sharing the machine with other processes. Validated on x86 and Raspberry Pi 4B. # ---- thread pools (small fixed sizes instead of CPU-core-based defaults) ---- +coordinator_read_executor_size=2 +coordinator_scheduled_executor_size=2 +fragment_instance_notification_thread_count=2 +cn_load_statistics_publisher_thread_count=1 query_thread_count=2 degree_of_query_parallelism=1 mpp_data_exchange_core_pool_size=2 @@ -192,4 +196,4 @@ pipe_logger_cache_max_size_in_bytes=1048576 # runtime meta (progress, remaining events, degraded/failure status) every # 30s instead of the stock 3s: fewer background heartbeat round trips on a # small node, at the cost of ~30s staleness in pipe status (stock: 3). -pipe_heartbeat_interval_seconds_for_collecting_pipe_meta=30 \ No newline at end of file +pipe_heartbeat_interval_seconds_for_collecting_pipe_meta=30 diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index aef3d0c3c3800..95a8154bba4d4 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -759,6 +759,11 @@ time_partition_interval=604800000 # Datatype: long heartbeat_interval_in_ms=1000 +# Number of ConfigNode workers publishing cluster load statistics changes. Must be positive. +# effectiveMode: restart +# Datatype: int +cn_load_statistics_publisher_thread_count=5 + # Default failure detector, enum from {fixed, phi_accrual} # effectiveMode: restart # Datatype: string @@ -1224,6 +1229,21 @@ max_allowed_concurrent_queries=1000 # Datatype: int query_thread_count=0 +# Number of coordinator workers executing read queries. Must be positive. +# effectiveMode: restart +# Datatype: int +coordinator_read_executor_size=20 + +# Number of coordinator workers scheduling query state checks and termination. Must be positive. +# effectiveMode: restart +# Datatype: int +coordinator_scheduled_executor_size=10 + +# Number of workers notifying fragment instance state changes. Must be positive. +# effectiveMode: restart +# Datatype: int +fragment_instance_notification_thread_count=4 + # How many pipeline drivers will be created for one fragment instance. When <= 0, use CPU core number / 2. # effectiveMode: restart # Datatype: int