ports) {
- String cmd = "lsof -iTCP -sTCP:LISTEN -P -n | awk '{print $9}' | grep -E ";
- return cmd + ports.stream().map(String::valueOf).collect(Collectors.joining("|")) + "\"";
+ /**
+ * List occupied port and the associated pid on Unix.
+ *
+ * A line of the lsof output splits into 10 columns, e.g. {@code java 1234 user 77u IPv6 0x1a2b
+ * 0t0 TCP 127.0.0.1:6667 (LISTEN)}. The address is in column 8 and the state in column 9.
+ *
+ * @param ports ports to be checked
+ * @return (occupiedPort, pid) pairs
+ */
+ public static Map listPortOccupationUnix(final List ports)
+ throws IOException {
+ return listPortOccupation(ports, "lsof -iTCP -sTCP:LISTEN,TIME_WAIT -P -n", 10, 8, 1);
}
- private static Pair getClusterNodesNum(int index) {
- String valueStr = System.getProperty(CLUSTER_CONFIGURATIONS);
+ private static Pair getClusterNodesNum(final int index) {
+ final String valueStr = System.getProperty(CLUSTER_CONFIGURATIONS);
if (valueStr == null) {
return null;
}
@@ -154,17 +199,17 @@ private static Pair getClusterNodesNum(int index) {
// Print nothing to avoid polluting test outputs
return null;
}
- } catch (NumberFormatException ignore) {
+ } catch (final NumberFormatException ignore) {
return null;
}
}
- public static String getLockFilePath(int port) {
+ public static String getLockFilePath(final int port) {
return LOCK_FILE_PATH + port;
}
public static Pair getNodeNum() {
- Pair nodesNum = getClusterNodesNum(0);
+ final Pair nodesNum = getClusterNodesNum(0);
if (nodesNum != null) {
return nodesNum;
}
@@ -173,8 +218,8 @@ public static Pair getNodeNum() {
getIntFromSysVar(DEFAULT_DATA_NODE_NUM, 3, 0));
}
- public static Pair getNodeNum(int index) {
- Pair nodesNum = getClusterNodesNum(index);
+ public static Pair getNodeNum(final int index) {
+ final Pair nodesNum = getClusterNodesNum(index);
if (nodesNum != null) {
return nodesNum;
}
@@ -183,38 +228,38 @@ public static Pair getNodeNum(int index) {
getIntFromSysVar(DEFAULT_DATA_NODE_NUM, 3, index));
}
- public static String getFilePathFromSysVar(String key, int index) {
- String valueStr = System.getProperty(key);
+ public static String getFilePathFromSysVar(final String key, final int index) {
+ final String valueStr = System.getProperty(key);
if (valueStr == null) {
return null;
}
return System.getProperty(USER_DIR) + getValueOfIndex(valueStr, index);
}
- public static int getIntFromSysVar(String key, int defaultValue, int index) {
- String valueStr = System.getProperty(key);
+ public static int getIntFromSysVar(final String key, final int defaultValue, final int index) {
+ final String valueStr = System.getProperty(key);
if (valueStr == null) {
return defaultValue;
}
- String value = getValueOfIndex(valueStr, index);
+ final String value = getValueOfIndex(valueStr, index);
try {
return Integer.parseInt(value);
- } catch (NumberFormatException e) {
+ } catch (final NumberFormatException e) {
throw new IllegalArgumentException("Invalid property value: " + value + " of key " + key);
}
}
- public static String getValueOfIndex(String valueStr, int index) {
- String[] values = valueStr.split(DELIMITER);
+ public static String getValueOfIndex(final String valueStr, final int index) {
+ final String[] values = valueStr.split(DELIMITER);
return index <= values.length - 1 ? values[index] : values[values.length - 1];
}
- public static String getTimeForLogDirectory(long startTime) {
+ public static String getTimeForLogDirectory(final long startTime) {
return convertLongToDate(startTime, "ms").replace(":", DIR_TIME_REPLACEMENT);
}
- public static String fromConsensusFullNameToAbbr(String consensus) {
+ public static String fromConsensusFullNameToAbbr(final String consensus) {
switch (consensus) {
case SIMPLE_CONSENSUS:
return SIMPLE_CONSENSUS_STR;
@@ -233,7 +278,7 @@ public static String fromConsensusFullNameToAbbr(String consensus) {
}
}
- public static String fromConsensusAbbrToFullName(String consensus) {
+ public static String fromConsensusAbbrToFullName(final String consensus) {
switch (consensus) {
case SIMPLE_CONSENSUS_STR:
return SIMPLE_CONSENSUS;
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppBaseConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppBaseConfig.java
index 473d2f1b64c7f..e0758de70c952 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppBaseConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppBaseConfig.java
@@ -19,8 +19,9 @@
package org.apache.iotdb.it.env.cluster.config;
+import jakarta.validation.constraints.NotNull;
+
import javax.annotation.Nullable;
-import javax.validation.constraints.NotNull;
import java.io.FileWriter;
import java.io.IOException;
@@ -40,6 +41,8 @@ public abstract class MppBaseConfig {
/** Create an empty MppPersistentConfig. */
protected MppBaseConfig() {
this.properties = new Properties();
+ this.properties.setProperty("cn_selector_thread_nums_of_client_manager", "1");
+ this.properties.setProperty("dn_selector_thread_nums_of_client_manager", "1");
}
/**
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
index 9a5e066d6419b..78e47cfd69ff7 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
@@ -124,11 +124,9 @@ public CommonConfig setEnableCrossSpaceCompaction(boolean enableCrossSpaceCompac
}
@Override
- public CommonConfig setMaxInnerCompactionCandidateFileNum(
- int maxInnerCompactionCandidateFileNum) {
+ public CommonConfig setInnerCompactionCandidateFileNum(int maxInnerCompactionCandidateFileNum) {
setProperty(
- "max_inner_compaction_candidate_file_num",
- String.valueOf(maxInnerCompactionCandidateFileNum));
+ "inner_compaction_candidate_file_num", String.valueOf(maxInnerCompactionCandidateFileNum));
return this;
}
@@ -151,8 +149,8 @@ public CommonConfig setPrimitiveArraySize(int primitiveArraySize) {
}
@Override
- public CommonConfig setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
- setProperty("avg_series_point_number_threshold", String.valueOf(avgSeriesPointNumberThreshold));
+ public CommonConfig setTargetChunkPointNum(int targetChunkPointNum) {
+ setProperty("target_chunk_point_num", String.valueOf(targetChunkPointNum));
return this;
}
@@ -232,6 +230,12 @@ public CommonConfig setTimePartitionInterval(long timePartitionInterval) {
return this;
}
+ @Override
+ public CommonConfig setTTLCheckInterval(long ttlCheckInterval) {
+ setProperty("ttl_check_interval", String.valueOf(ttlCheckInterval));
+ return this;
+ }
+
@Override
public CommonConfig setTimePartitionOrigin(long timePartitionOrigin) {
setProperty("time_partition_origin", String.valueOf(timePartitionOrigin));
@@ -359,6 +363,18 @@ public CommonConfig setSeriesSlotNum(int seriesSlotNum) {
return this;
}
+ @Override
+ public CommonConfig setDataPartitionAllocationStrategy(String dataPartitionAllocationStrategy) {
+ setProperty("data_partition_allocation_strategy", dataPartitionAllocationStrategy);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setSeriesPartitionExecutorClass(String seriesPartitionExecutorClass) {
+ setProperty("series_partition_executor_class", seriesPartitionExecutorClass);
+ return this;
+ }
+
@Override
public CommonConfig setSchemaMemoryAllocate(String schemaMemoryAllocate) {
setProperty("schema_memory_proportion", String.valueOf(schemaMemoryAllocate));
@@ -419,6 +435,36 @@ public CommonConfig setSchemaRegionPerDataNode(double schemaRegionPerDataNode) {
return this;
}
+ @Override
+ public CommonConfig setPipeMemoryManagementEnabled(boolean pipeMemoryManagementEnabled) {
+ setProperty("pipe_memory_management_enabled", String.valueOf(pipeMemoryManagementEnabled));
+ return this;
+ }
+
+ @Override
+ public CommonConfig setPipeDataStructureTabletMemoryBlockAllocationRejectThreshold(
+ double pipeDataStructureTabletMemoryBlockAllocationRejectThreshold) {
+ setProperty(
+ "pipe_data_structure_tablet_memory_block_allocation_reject_threshold",
+ String.valueOf(pipeDataStructureTabletMemoryBlockAllocationRejectThreshold));
+ return this;
+ }
+
+ @Override
+ public CommonConfig setPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold(
+ double pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold) {
+ setProperty(
+ "pipe_data_structure_ts_file_memory_block_allocation_reject_threshold",
+ String.valueOf(pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold));
+ return this;
+ }
+
+ @Override
+ public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) {
+ setProperty("pipe_enable_memory_checked", String.valueOf(isPipeEnableMemoryCheck));
+ return this;
+ }
+
@Override
public CommonConfig setPipeAirGapReceiverEnabled(boolean isPipeAirGapReceiverEnabled) {
setProperty("pipe_air_gap_receiver_enabled", String.valueOf(isPipeAirGapReceiverEnabled));
@@ -445,8 +491,8 @@ public CommonConfig setTagAttributeTotalSize(int tagAttributeTotalSize) {
}
@Override
- public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) {
- setProperty("cn_connection_timeout_ms", String.valueOf(connectionTimeoutMs));
+ public CommonConfig setDnConnectionTimeoutMs(int connectionTimeoutMs) {
+ setProperty("dn_connection_timeout_ms", String.valueOf(connectionTimeoutMs));
return this;
}
@@ -476,6 +522,58 @@ public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSync
return this;
}
+ @Override
+ public CommonConfig setPipeConnectorRequestSliceThresholdBytes(
+ int pipeConnectorRequestSliceThresholdBytes) {
+ setProperty(
+ "pipe_sink_request_slice_threshold_bytes",
+ String.valueOf(pipeConnectorRequestSliceThresholdBytes));
+
+ return this;
+ }
+
+ @Override
+ public CommonConfig setPipeAutoSplitFullEnabled(boolean pipeAutoSplitFullEnabled) {
+ setProperty("pipe_auto_split_full_enabled", String.valueOf(pipeAutoSplitFullEnabled));
+ return this;
+ }
+
+ @Override
+ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
+ setProperty("chunk_timeseriesmeta_free_memory_proportion", queryMemoryProportion);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+ int subscriptionPrefetchTsFileBatchMaxDelayInMs) {
+ setProperty(
+ "subscription_prefetch_ts_file_batch_max_delay_in_ms",
+ String.valueOf(subscriptionPrefetchTsFileBatchMaxDelayInMs));
+ return this;
+ }
+
+ @Override
+ public CommonConfig setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+ int subscriptionPrefetchTsFileBatchMaxSizeInBytes) {
+ setProperty(
+ "subscription_prefetch_ts_file_batch_max_size_in_bytes",
+ String.valueOf(subscriptionPrefetchTsFileBatchMaxSizeInBytes));
+ return this;
+ }
+
+ @Override
+ public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
+ setProperty("default_database_level", String.valueOf(defaultStorageGroupLevel));
+ return this;
+ }
+
+ @Override
+ public CommonConfig setDatanodeMemoryProportion(String datanodeMemoryProportion) {
+ setProperty("datanode_memory_proportion", datanodeMemoryProportion);
+ return this;
+ }
+
// For part of the log directory
public String getClusterConfigStr() {
return fromConsensusFullNameToAbbr(properties.getProperty(CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS))
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
index 62ccbb0aa4ff6..4385097ce045b 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
@@ -55,4 +55,22 @@ public ConfigNodeConfig setMetricReporterType(List metricReporterTypes)
properties.setProperty("cn_metric_reporter_list", String.join(",", metricReporterTypes));
return this;
}
+
+ @Override
+ public ConfigNodeConfig setMetricPrometheusReporterUsername(String username) {
+ properties.setProperty("metric_prometheus_reporter_username", username);
+ return this;
+ }
+
+ @Override
+ public ConfigNodeConfig setMetricPrometheusReporterPassword(String password) {
+ properties.setProperty("metric_prometheus_reporter_password", password);
+ return this;
+ }
+
+ @Override
+ public ConfigNodeConfig setLeaderDistributionPolicy(String policy) {
+ properties.setProperty("leader_distribution_policy", policy);
+ return this;
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java
index 0f22e0d428699..3119b04846f94 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java
@@ -24,6 +24,8 @@
import java.io.IOException;
import java.util.List;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.DN_DATA_DIRS;
+
public class MppDataNodeConfig extends MppBaseConfig implements DataNodeConfig {
public MppDataNodeConfig() {
@@ -56,6 +58,18 @@ public DataNodeConfig setMetricReporterType(List metricReporterTypes) {
return this;
}
+ @Override
+ public DataNodeConfig setMetricPrometheusReporterUsername(String username) {
+ properties.setProperty("metric_prometheus_reporter_username", username);
+ return this;
+ }
+
+ @Override
+ public DataNodeConfig setMetricPrometheusReporterPassword(String password) {
+ properties.setProperty("metric_prometheus_reporter_password", password);
+ return this;
+ }
+
@Override
public DataNodeConfig setEnableRestService(boolean enableRestService) {
properties.setProperty("enable_rest_service", String.valueOf(enableRestService));
@@ -68,6 +82,12 @@ public DataNodeConfig setConnectionTimeoutInMS(int connectionTimeoutInMS) {
return this;
}
+ @Override
+ public DataNodeConfig setDnDataDirs(String dnDataDirs) {
+ setProperty(DN_DATA_DIRS, dnDataDirs);
+ return this;
+ }
+
@Override
public DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes(
long loadTsFileAnalyzeSchemaMemorySizeInBytes) {
@@ -76,4 +96,53 @@ public DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes(
String.valueOf(loadTsFileAnalyzeSchemaMemorySizeInBytes));
return this;
}
+
+ @Override
+ public DataNodeConfig setMaxAllocateMemoryRatioForLoad(double maxAllocateMemoryRatioForLoad) {
+ properties.setProperty(
+ "max_allocate_memory_ratio_for_load", String.valueOf(maxAllocateMemoryRatioForLoad));
+ return this;
+ }
+
+ @Override
+ public DataNodeConfig setLoadTsFileTabletConversionBatchMemorySizeInBytes(
+ long loadTsFileTabletConversionBatchMemorySizeInBytes) {
+ properties.setProperty(
+ "load_tsfile_tablet_conversion_batch_memory_size_in_bytes",
+ String.valueOf(loadTsFileTabletConversionBatchMemorySizeInBytes));
+ return this;
+ }
+
+ @Override
+ public DataNodeConfig setLoadActiveListeningCheckIntervalSeconds(
+ long loadActiveListeningCheckIntervalSeconds) {
+ properties.setProperty(
+ "load_active_listening_check_interval_seconds",
+ String.valueOf(loadActiveListeningCheckIntervalSeconds));
+ return this;
+ }
+
+ @Override
+ public DataNodeConfig setLoadLastCacheStrategy(String strategyName) {
+ setProperty("last_cache_operation_on_load", strategyName);
+ return this;
+ }
+
+ @Override
+ public DataNodeConfig setCacheLastValuesForLoad(boolean cacheLastValuesForLoad) {
+ setProperty("cache_last_values_for_load", String.valueOf(cacheLastValuesForLoad));
+ return this;
+ }
+
+ @Override
+ public DataNodeConfig setWalThrottleSize(long walThrottleSize) {
+ setProperty("wal_throttle_threshold_in_byte", String.valueOf(walThrottleSize));
+ return this;
+ }
+
+ @Override
+ public DataNodeConfig setDeleteWalFilesPeriodInMs(long deleteWalFilesPeriodInMs) {
+ setProperty("delete_wal_files_period_in_ms", String.valueOf(deleteWalFilesPeriodInMs));
+ return this;
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java
index d9c51717cf8e1..7e46af797360b 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java
@@ -29,6 +29,16 @@ public class MppJVMConfig implements JVMConfig {
private int maxHeapSize;
private int maxDirectMemorySize;
private String timezone;
+ private int debugPort;
+
+ /** Enables loopback-only JDWP for tests that need to observe an exact node execution stage. */
+ public void setDebugPort(int debugPort) {
+ this.debugPort = debugPort;
+ }
+
+ public int getDebugPort() {
+ return debugPort;
+ }
@Override
public JVMConfig setInitHeapSize(int initSize) {
@@ -94,6 +104,7 @@ public void override(@Nullable MppJVMConfig config) {
this.setMaxHeapSize(config.getMaxHeapSize());
this.setMaxDirectMemorySize(config.getMaxDirectMemorySize());
this.setTimezone(config.getTimezone());
+ this.setDebugPort(config.getDebugPort());
}
public static Builder builder() {
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
index 2d93a2e0dafc0..ee78782e23a7a 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
@@ -118,10 +118,9 @@ public CommonConfig setEnableCrossSpaceCompaction(boolean enableCrossSpaceCompac
}
@Override
- public CommonConfig setMaxInnerCompactionCandidateFileNum(
- int maxInnerCompactionCandidateFileNum) {
- cnConfig.setMaxInnerCompactionCandidateFileNum(maxInnerCompactionCandidateFileNum);
- dnConfig.setMaxInnerCompactionCandidateFileNum(maxInnerCompactionCandidateFileNum);
+ public CommonConfig setInnerCompactionCandidateFileNum(int maxInnerCompactionCandidateFileNum) {
+ cnConfig.setInnerCompactionCandidateFileNum(maxInnerCompactionCandidateFileNum);
+ dnConfig.setInnerCompactionCandidateFileNum(maxInnerCompactionCandidateFileNum);
return this;
}
@@ -147,9 +146,9 @@ public CommonConfig setPrimitiveArraySize(int primitiveArraySize) {
}
@Override
- public CommonConfig setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
- cnConfig.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
- dnConfig.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
+ public CommonConfig setTargetChunkPointNum(int targetChunkPointNum) {
+ cnConfig.setTargetChunkPointNum(targetChunkPointNum);
+ dnConfig.setTargetChunkPointNum(targetChunkPointNum);
return this;
}
@@ -231,6 +230,13 @@ public CommonConfig setTimePartitionInterval(long timePartitionInterval) {
return this;
}
+ @Override
+ public CommonConfig setTTLCheckInterval(long ttlCheckInterval) {
+ cnConfig.setTTLCheckInterval(ttlCheckInterval);
+ dnConfig.setTTLCheckInterval(ttlCheckInterval);
+ return this;
+ }
+
@Override
public CommonConfig setTimePartitionOrigin(long timePartitionOrigin) {
cnConfig.setTimePartitionOrigin(timePartitionOrigin);
@@ -355,6 +361,19 @@ public CommonConfig setSeriesSlotNum(int seriesSlotNum) {
return this;
}
+ @Override
+ public CommonConfig setDataPartitionAllocationStrategy(String dataPartitionAllocationStrategy) {
+ cnConfig.setDataPartitionAllocationStrategy(dataPartitionAllocationStrategy);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setSeriesPartitionExecutorClass(String seriesPartitionExecutorClass) {
+ cnConfig.setSeriesPartitionExecutorClass(seriesPartitionExecutorClass);
+ dnConfig.setSeriesPartitionExecutorClass(seriesPartitionExecutorClass);
+ return this;
+ }
+
@Override
public CommonConfig setSchemaMemoryAllocate(String schemaMemoryAllocate) {
dnConfig.setSchemaMemoryAllocate(schemaMemoryAllocate);
@@ -425,6 +444,40 @@ public CommonConfig setSchemaRegionPerDataNode(double schemaRegionPerDataNode) {
return this;
}
+ @Override
+ public CommonConfig setPipeMemoryManagementEnabled(boolean pipeMemoryManagementEnabled) {
+ dnConfig.setPipeMemoryManagementEnabled(pipeMemoryManagementEnabled);
+ cnConfig.setPipeMemoryManagementEnabled(pipeMemoryManagementEnabled);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setPipeDataStructureTabletMemoryBlockAllocationRejectThreshold(
+ double pipeDataStructureTabletMemoryBlockAllocationRejectThreshold) {
+ dnConfig.setPipeDataStructureTabletMemoryBlockAllocationRejectThreshold(
+ pipeDataStructureTabletMemoryBlockAllocationRejectThreshold);
+ cnConfig.setPipeDataStructureTabletMemoryBlockAllocationRejectThreshold(
+ pipeDataStructureTabletMemoryBlockAllocationRejectThreshold);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold(
+ double pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold) {
+ dnConfig.setPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold(
+ pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold);
+ cnConfig.setPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold(
+ pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) {
+ dnConfig.setIsPipeEnableMemoryCheck(isPipeEnableMemoryCheck);
+ cnConfig.setIsPipeEnableMemoryCheck(isPipeEnableMemoryCheck);
+ return this;
+ }
+
@Override
public CommonConfig setPipeAirGapReceiverEnabled(boolean isPipeAirGapReceiverEnabled) {
dnConfig.setPipeAirGapReceiverEnabled(isPipeAirGapReceiverEnabled);
@@ -454,9 +507,9 @@ public CommonConfig setTagAttributeTotalSize(int tagAttributeTotalSize) {
}
@Override
- public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) {
- dnConfig.setCnConnectionTimeoutMs(connectionTimeoutMs);
- cnConfig.setCnConnectionTimeoutMs(connectionTimeoutMs);
+ public CommonConfig setDnConnectionTimeoutMs(int connectionTimeoutMs) {
+ dnConfig.setDnConnectionTimeoutMs(connectionTimeoutMs);
+ cnConfig.setDnConnectionTimeoutMs(connectionTimeoutMs);
return this;
}
@@ -484,4 +537,59 @@ public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSync
cnConfig.setPipeMetaSyncerSyncIntervalMinutes(pipeMetaSyncerSyncIntervalMinutes);
return this;
}
+
+ @Override
+ public CommonConfig setPipeConnectorRequestSliceThresholdBytes(
+ int pipeConnectorRequestSliceThresholdBytes) {
+ dnConfig.setPipeConnectorRequestSliceThresholdBytes(pipeConnectorRequestSliceThresholdBytes);
+ cnConfig.setPipeConnectorRequestSliceThresholdBytes(pipeConnectorRequestSliceThresholdBytes);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setPipeAutoSplitFullEnabled(boolean pipeAutoSplitFullEnabled) {
+ dnConfig.setPipeAutoSplitFullEnabled(pipeAutoSplitFullEnabled);
+ cnConfig.setPipeAutoSplitFullEnabled(pipeAutoSplitFullEnabled);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
+ dnConfig.setQueryMemoryProportion(queryMemoryProportion);
+ cnConfig.setQueryMemoryProportion(queryMemoryProportion);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+ int subscriptionPrefetchTsFileBatchMaxDelayInMs) {
+ dnConfig.setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+ subscriptionPrefetchTsFileBatchMaxDelayInMs);
+ cnConfig.setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+ subscriptionPrefetchTsFileBatchMaxDelayInMs);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+ int subscriptionPrefetchTsFileBatchMaxSizeInBytes) {
+ dnConfig.setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+ subscriptionPrefetchTsFileBatchMaxSizeInBytes);
+ cnConfig.setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+ subscriptionPrefetchTsFileBatchMaxSizeInBytes);
+ return this;
+ }
+
+ @Override
+ public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
+ dnConfig.setDefaultStorageGroupLevel(defaultStorageGroupLevel);
+ cnConfig.setDefaultStorageGroupLevel(defaultStorageGroupLevel);
+ return this;
+ }
+
+ public CommonConfig setDatanodeMemoryProportion(String datanodeMemoryProportion) {
+ dnConfig.setDatanodeMemoryProportion(datanodeMemoryProportion);
+ cnConfig.setDatanodeMemoryProportion(datanodeMemoryProportion);
+ return this;
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AIEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AIEnv.java
new file mode 100644
index 0000000000000..916a2ee8cf6cf
--- /dev/null
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AIEnv.java
@@ -0,0 +1,38 @@
+/*
+ * 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.it.env.cluster.env;
+
+public class AIEnv extends AbstractEnv {
+ @Override
+ public void initClusterEnvironment() {
+ initClusterEnvironment(1, 1);
+ }
+
+ @Override
+ public void initClusterEnvironment(int configNodesNum, int dataNodesNum) {
+ super.initEnvironment(configNodesNum, dataNodesNum, 1000, true);
+ }
+
+ @Override
+ public void initClusterEnvironment(
+ int configNodesNum, int dataNodesNum, int testWorkingRetryCount) {
+ super.initEnvironment(configNodesNum, dataNodesNum, testWorkingRetryCount, true);
+ }
+}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
index 54fedec8de0f6..ffbd9b78d70d4 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
@@ -19,10 +19,9 @@
package org.apache.iotdb.it.env.cluster.env;
-import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
-import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.ClientPoolFactory;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.exception.ClientManagerException;
@@ -41,6 +40,7 @@
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.EnvUtils;
import org.apache.iotdb.it.env.cluster.config.*;
+import org.apache.iotdb.it.env.cluster.node.AINodeWrapper;
import org.apache.iotdb.it.env.cluster.node.AbstractNodeWrapper;
import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
@@ -57,7 +57,10 @@
import org.apache.iotdb.session.Session;
import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.thrift.TConfiguration;
import org.apache.thrift.TException;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import java.io.File;
@@ -65,6 +68,7 @@
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
+import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
@@ -76,14 +80,19 @@
public abstract class AbstractEnv implements BaseEnv {
private static final Logger logger = IoTDBTestLogger.logger;
+ private static final int DEFAULT_CLUSTER_READY_RETRY_COUNT = 30;
+ private static final String CLUSTER_READY_RETRY_COUNT_PROPERTY =
+ "integrationTest.clusterReadyRetryCount";
private final Random rand = new Random();
protected List configNodeWrapperList = Collections.emptyList();
protected List dataNodeWrapperList = Collections.emptyList();
+ protected List aiNodeWrapperList = Collections.emptyList();
+ protected String testClassName = null;
protected String testMethodName = null;
protected int index = 0;
protected long startTime;
- protected int retryCount = 30;
+ protected int retryCount = getDefaultClusterReadyRetryCount();
private IClientManager clientManager;
private List configNodeKillPoints = new ArrayList<>();
private List dataNodeKillPoints = new ArrayList<>();
@@ -101,49 +110,66 @@ protected AbstractEnv() {
}
// For multiple environment ITs, time must be consistent across environments.
- protected AbstractEnv(long startTime) {
+ protected AbstractEnv(final long startTime) {
this.startTime = startTime;
this.clusterConfig = new MppClusterConfig();
}
+ private static int getDefaultClusterReadyRetryCount() {
+ final int configuredRetryCount =
+ Integer.getInteger(CLUSTER_READY_RETRY_COUNT_PROPERTY, DEFAULT_CLUSTER_READY_RETRY_COUNT);
+ return configuredRetryCount > 0 ? configuredRetryCount : DEFAULT_CLUSTER_READY_RETRY_COUNT;
+ }
+
@Override
public ClusterConfig getConfig() {
return clusterConfig;
}
@Override
- public List getMetricPrometheusReporterContents() {
- List result = new ArrayList<>();
+ public List getMetricPrometheusReporterContents(String authHeader) {
+ final List result = new ArrayList<>();
// get all report content of confignodes
- for (ConfigNodeWrapper configNode : this.configNodeWrapperList) {
- String configNodeMetricContent =
+ for (final ConfigNodeWrapper configNode : this.configNodeWrapperList) {
+ final String configNodeMetricContent =
getUrlContent(
Config.IOTDB_HTTP_URL_PREFIX
+ configNode.getIp()
+ ":"
+ configNode.getMetricPort()
- + "/metrics");
+ + "/metrics",
+ authHeader);
result.add(configNodeMetricContent);
}
// get all report content of datanodes
- for (DataNodeWrapper dataNode : this.dataNodeWrapperList) {
- String dataNodeMetricContent =
+ for (final DataNodeWrapper dataNode : this.dataNodeWrapperList) {
+ final String dataNodeMetricContent =
getUrlContent(
Config.IOTDB_HTTP_URL_PREFIX
+ dataNode.getIp()
+ ":"
+ dataNode.getMetricPort()
- + "/metrics");
+ + "/metrics",
+ authHeader);
result.add(dataNodeMetricContent);
}
return result;
}
- protected void initEnvironment(int configNodesNum, int dataNodesNum) {
+ protected void initEnvironment(final int configNodesNum, final int dataNodesNum) {
initEnvironment(configNodesNum, dataNodesNum, retryCount);
}
- protected void initEnvironment(int configNodesNum, int dataNodesNum, int retryCount) {
+ protected void initEnvironment(
+ final int configNodesNum, final int dataNodesNum, final int testWorkingRetryCount) {
+ initEnvironment(configNodesNum, dataNodesNum, testWorkingRetryCount, false);
+ }
+
+ protected void initEnvironment(
+ final int configNodesNum,
+ final int dataNodesNum,
+ final int retryCount,
+ final boolean addAINode) {
this.retryCount = retryCount;
this.configNodeWrapperList = new ArrayList<>();
this.dataNodeWrapperList = new ArrayList<>();
@@ -154,7 +180,7 @@ protected void initEnvironment(int configNodesNum, int dataNodesNum, int retryCo
final String testClassName = getTestClassName();
- ConfigNodeWrapper seedConfigNodeWrapper =
+ final ConfigNodeWrapper seedConfigNodeWrapper =
new ConfigNodeWrapper(
true,
"",
@@ -172,22 +198,23 @@ protected void initEnvironment(int configNodesNum, int dataNodesNum, int retryCo
seedConfigNodeWrapper.createLogDir();
seedConfigNodeWrapper.setKillPoints(configNodeKillPoints);
seedConfigNodeWrapper.start();
- String seedConfigNode = seedConfigNodeWrapper.getIpAndPortString();
+ final String seedConfigNode = seedConfigNodeWrapper.getIpAndPortString();
this.configNodeWrapperList.add(seedConfigNodeWrapper);
// Check if the Seed-ConfigNode started successfully
- try (SyncConfigNodeIServiceClient ignored =
+ try (final SyncConfigNodeIServiceClient ignored =
(SyncConfigNodeIServiceClient) getLeaderConfigNodeConnection()) {
// Do nothing
logger.info("The Seed-ConfigNode started successfully!");
- } catch (Exception e) {
+ } catch (final Exception e) {
logger.error("Failed to get connection to the Seed-ConfigNode", e);
}
- List configNodeEndpoints = new ArrayList<>();
- RequestDelegate configNodesDelegate = new SerialRequestDelegate<>(configNodeEndpoints);
+ final List configNodeEndpoints = new ArrayList<>();
+ final RequestDelegate configNodesDelegate =
+ new SerialRequestDelegate<>(configNodeEndpoints);
for (int i = 1; i < configNodesNum; i++) {
- ConfigNodeWrapper configNodeWrapper =
+ final ConfigNodeWrapper configNodeWrapper =
new ConfigNodeWrapper(
false,
seedConfigNode,
@@ -214,16 +241,16 @@ protected void initEnvironment(int configNodesNum, int dataNodesNum, int retryCo
}
try {
configNodesDelegate.requestAll();
- } catch (SQLException e) {
+ } catch (final SQLException e) {
logger.error("Start configNodes failed", e);
throw new AssertionError();
}
- List dataNodeEndpoints = new ArrayList<>();
- RequestDelegate dataNodesDelegate =
- new ParallelRequestDelegate<>(dataNodeEndpoints, NODE_START_TIMEOUT);
+ final List dataNodeEndpoints = new ArrayList<>();
+ final RequestDelegate dataNodesDelegate =
+ new ParallelRequestDelegate<>(dataNodeEndpoints, NODE_START_TIMEOUT, this);
for (int i = 0; i < dataNodesNum; i++) {
- DataNodeWrapper dataNodeWrapper =
+ final DataNodeWrapper dataNodeWrapper =
new DataNodeWrapper(
seedConfigNode,
testClassName,
@@ -250,20 +277,59 @@ protected void initEnvironment(int configNodesNum, int dataNodesNum, int retryCo
try {
dataNodesDelegate.requestAll();
- } catch (SQLException e) {
+ } catch (final SQLException e) {
logger.error("Start dataNodes failed", e);
throw new AssertionError();
}
+ if (addAINode) {
+ this.aiNodeWrapperList = new ArrayList<>();
+ startAINode(seedConfigNode, testClassName);
+ }
+
checkClusterStatusWithoutUnknown();
}
+ private void startAINode(final String seedConfigNode, final String testClassName) {
+ final String aiNodeEndPoint;
+ final AINodeWrapper aiNodeWrapper =
+ new AINodeWrapper(
+ seedConfigNode,
+ testClassName,
+ testMethodName,
+ index,
+ EnvUtils.searchAvailablePorts(),
+ startTime);
+ aiNodeWrapperList.add(aiNodeWrapper);
+ aiNodeEndPoint = aiNodeWrapper.getIpAndPortString();
+ aiNodeWrapper.createNodeDir();
+ aiNodeWrapper.createLogDir();
+ final RequestDelegate aiNodesDelegate =
+ new ParallelRequestDelegate<>(
+ Collections.singletonList(aiNodeEndPoint), NODE_START_TIMEOUT, this);
+
+ aiNodesDelegate.addRequest(
+ () -> {
+ aiNodeWrapper.start();
+ return null;
+ });
+
+ try {
+ aiNodesDelegate.requestAll();
+ } catch (final SQLException e) {
+ logger.error("Start aiNodes failed", e);
+ }
+ }
+
public String getTestClassName() {
- StackTraceElement[] stack = Thread.currentThread().getStackTrace();
- for (StackTraceElement stackTraceElement : stack) {
- String className = stackTraceElement.getClassName();
+ if (testClassName != null) {
+ return testClassName;
+ }
+ final StackTraceElement[] stack = Thread.currentThread().getStackTrace();
+ for (final StackTraceElement stackTraceElement : stack) {
+ final String className = stackTraceElement.getClassName();
if (className.endsWith("IT")) {
- String result = className.substring(className.lastIndexOf(".") + 1);
+ final String result = className.substring(className.lastIndexOf(".") + 1);
if (!result.startsWith("Abstract")) {
return result;
}
@@ -272,15 +338,21 @@ public String getTestClassName() {
return "UNKNOWN-IT";
}
- private Map countNodeStatus(Map nodeStatus) {
- Map result = new HashMap<>();
+ private Map countNodeStatus(final Map nodeStatus) {
+ final Map result = new HashMap<>();
nodeStatus.values().forEach(status -> result.put(status, result.getOrDefault(status, 0) + 1));
return result;
}
+ public void checkNodeInStatus(int nodeId, NodeStatus expectation) {
+ checkClusterStatus(
+ nodeStatusMap -> expectation.getStatus().equals(nodeStatusMap.get(nodeId)), m -> true);
+ }
+
public void checkClusterStatusWithoutUnknown() {
checkClusterStatus(
- nodeStatusMap -> nodeStatusMap.values().stream().noneMatch("Unknown"::equals));
+ nodeStatusMap -> nodeStatusMap.values().stream().noneMatch("Unknown"::equals),
+ processStatus -> processStatus.values().stream().noneMatch(i -> i != 0));
testJDBCConnection();
}
@@ -290,6 +362,10 @@ public void checkClusterStatusOneUnknownOtherRunning() {
Map count = countNodeStatus(nodeStatus);
return count.getOrDefault("Unknown", 0) == 1
&& count.getOrDefault("Running", 0) == nodeStatus.size() - 1;
+ },
+ processStatus -> {
+ long aliveProcessCount = processStatus.values().stream().filter(i -> i == 0).count();
+ return aliveProcessCount == processStatus.size() - 1;
});
testJDBCConnection();
}
@@ -300,44 +376,96 @@ public void checkClusterStatusOneUnknownOtherRunning() {
*
* @param statusCheck the predicate to test the status of nodes
*/
- public void checkClusterStatus(Predicate