You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2022/10/26 04:38:22 UTC
[iotdb] 01/02: add confignode standaloe for ConsensusManager
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/confignode_standalone
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 23399bd08b2d367924fa5d776deb0b076952f97b
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Oct 26 12:37:26 2022 +0800
add confignode standaloe for ConsensusManager
---
.../iotdb/confignode/conf/ConfigNodeConfig.java | 10 ++
.../confignode/conf/ConfigNodeDescriptor.java | 6 +
.../iotdb/confignode/manager/ConsensusManager.java | 179 ++++++++++++---------
3 files changed, 115 insertions(+), 80 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
index 9ebb791180..0140966019 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
@@ -169,6 +169,7 @@ public class ConfigNodeConfig {
private long partitionRegionRatisLogSegmentSizeMax = 24 * 1024 * 1024L;
private long schemaRegionRatisLogSegmentSizeMax = 24 * 1024 * 1024L;
+ private long partitionRegionStandAloneLogSegmentSizeMax = 24 * 1024 * 1024L;
/** RatisConsensus protocol, flow control window for ratis grpc log appender */
private long dataRegionRatisGrpcFlowControlWindow = 4 * 1024 * 1024L;
@@ -695,6 +696,15 @@ public class ConfigNodeConfig {
this.schemaRegionRatisLogSegmentSizeMax = schemaRegionRatisLogSegmentSizeMax;
}
+ public long getPartitionRegionStandAloneLogSegmentSizeMax() {
+ return partitionRegionStandAloneLogSegmentSizeMax;
+ }
+
+ public void setPartitionRegionStandAloneLogSegmentSizeMax(
+ long partitionRegionStandAloneLogSegmentSizeMax) {
+ this.partitionRegionStandAloneLogSegmentSizeMax = partitionRegionStandAloneLogSegmentSizeMax;
+ }
+
public long getSchemaRegionRatisGrpcFlowControlWindow() {
return schemaRegionRatisGrpcFlowControlWindow;
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
index d3aa2a66cd..ed4142ce97 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
@@ -356,6 +356,12 @@ public class ConfigNodeDescriptor {
"schema_region_ratis_log_segment_size_max",
String.valueOf(conf.getSchemaRegionRatisLogSegmentSizeMax()))));
+ conf.setPartitionRegionStandAloneLogSegmentSizeMax(
+ Long.parseLong(
+ properties.getProperty(
+ "partition_region_standalone_log_segment_size_max",
+ String.valueOf(conf.getPartitionRegionStandAloneLogSegmentSizeMax()))));
+
conf.setDataRegionRatisGrpcFlowControlWindow(
Long.parseLong(
properties.getProperty(
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
index bc266bd89e..0e7c9f9bf5 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConsensusManager.java
@@ -79,86 +79,105 @@ public class ConsensusManager {
// There is only one ConfigNodeGroup
consensusGroupId = new PartitionRegionId(CONF.getPartitionRegionId());
- // Implement local ConsensusLayer by ConfigNodeConfig
- consensusImpl =
- ConsensusFactory.getConsensusImpl(
- CONF.getConfigNodeConsensusProtocolClass(),
- ConsensusConfig.newBuilder()
- .setThisNodeId(CONF.getConfigNodeId())
- .setThisNode(new TEndPoint(CONF.getInternalAddress(), CONF.getConsensusPort()))
- .setRatisConfig(
- RatisConfig.newBuilder()
- .setLeaderLogAppender(
- RatisConfig.LeaderLogAppender.newBuilder()
- .setBufferByteLimit(
- CONF
- .getPartitionRegionRatisConsensusLogAppenderBufferSize())
- .build())
- .setSnapshot(
- RatisConfig.Snapshot.newBuilder()
- .setAutoTriggerThreshold(
- CONF.getPartitionRegionRatisSnapshotTriggerThreshold())
- .build())
- .setLog(
- RatisConfig.Log.newBuilder()
- .setUnsafeFlushEnabled(
- CONF.isPartitionRegionRatisLogUnsafeFlushEnable())
- .setSegmentCacheSizeMax(
- SizeInBytes.valueOf(
- CONF.getPartitionRegionRatisLogSegmentSizeMax()))
- .build())
- .setGrpc(
- RatisConfig.Grpc.newBuilder()
- .setFlowControlWindow(
- SizeInBytes.valueOf(
- CONF.getPartitionRegionRatisGrpcFlowControlWindow()))
- .build())
- .setRpc(
- RatisConfig.Rpc.newBuilder()
- .setTimeoutMin(
- TimeDuration.valueOf(
- CONF
- .getPartitionRegionRatisRpcLeaderElectionTimeoutMinMs(),
- TimeUnit.MILLISECONDS))
- .setTimeoutMax(
- TimeDuration.valueOf(
- CONF
- .getPartitionRegionRatisRpcLeaderElectionTimeoutMaxMs(),
- TimeUnit.MILLISECONDS))
- .setRequestTimeout(
- TimeDuration.valueOf(
- CONF.getPartitionRegionRatisRequestTimeoutMs(),
- TimeUnit.MILLISECONDS))
- .setFirstElectionTimeoutMin(
- TimeDuration.valueOf(
- CONF.getRatisFirstElectionTimeoutMinMs(),
- TimeUnit.MILLISECONDS))
- .setFirstElectionTimeoutMax(
- TimeDuration.valueOf(
- CONF.getRatisFirstElectionTimeoutMaxMs(),
- TimeUnit.MILLISECONDS))
- .build())
- .setRatisConsensus(
- RatisConfig.RatisConsensus.newBuilder()
- .setClientRequestTimeoutMillis(
- CONF.getPartitionRegionRatisRequestTimeoutMs())
- .setClientMaxRetryAttempt(
- CONF.getPartitionRegionRatisMaxRetryAttempts())
- .setClientRetryInitialSleepTimeMs(
- CONF.getPartitionRegionRatisInitialSleepTimeMs())
- .setClientRetryMaxSleepTimeMs(
- CONF.getPartitionRegionRatisMaxSleepTimeMs())
- .build())
- .build())
- .setStorageDir(CONF.getConsensusDir())
- .build(),
- gid -> stateMachine)
- .orElseThrow(
- () ->
- new IllegalArgumentException(
- String.format(
- ConsensusFactory.CONSTRUCT_FAILED_MSG,
- CONF.getConfigNodeConsensusProtocolClass())));
+ if (ConsensusFactory.StandAloneConsensus.equals(CONF.getConfigNodeConsensusProtocolClass())) {
+ consensusImpl =
+ ConsensusFactory.getConsensusImpl(
+ ConsensusFactory.StandAloneConsensus,
+ ConsensusConfig.newBuilder()
+ .setThisNode(
+ new TEndPoint(CONF.getInternalAddress(), CONF.getConsensusPort()))
+ .setStorageDir("target" + java.io.File.separator + "standalone")
+ .build(),
+ gid -> stateMachine)
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ String.format(
+ ConsensusFactory.CONSTRUCT_FAILED_MSG,
+ ConsensusFactory.StandAloneConsensus)));
+ } else {
+
+ // Implement local ConsensusLayer by ConfigNodeConfig
+ consensusImpl =
+ ConsensusFactory.getConsensusImpl(
+ CONF.getConfigNodeConsensusProtocolClass(),
+ ConsensusConfig.newBuilder()
+ .setThisNodeId(CONF.getConfigNodeId())
+ .setThisNode(new TEndPoint(CONF.getInternalAddress(), CONF.getConsensusPort()))
+ .setRatisConfig(
+ RatisConfig.newBuilder()
+ .setLeaderLogAppender(
+ RatisConfig.LeaderLogAppender.newBuilder()
+ .setBufferByteLimit(
+ CONF
+ .getPartitionRegionRatisConsensusLogAppenderBufferSize())
+ .build())
+ .setSnapshot(
+ RatisConfig.Snapshot.newBuilder()
+ .setAutoTriggerThreshold(
+ CONF.getPartitionRegionRatisSnapshotTriggerThreshold())
+ .build())
+ .setLog(
+ RatisConfig.Log.newBuilder()
+ .setUnsafeFlushEnabled(
+ CONF.isPartitionRegionRatisLogUnsafeFlushEnable())
+ .setSegmentCacheSizeMax(
+ SizeInBytes.valueOf(
+ CONF.getPartitionRegionRatisLogSegmentSizeMax()))
+ .build())
+ .setGrpc(
+ RatisConfig.Grpc.newBuilder()
+ .setFlowControlWindow(
+ SizeInBytes.valueOf(
+ CONF.getPartitionRegionRatisGrpcFlowControlWindow()))
+ .build())
+ .setRpc(
+ RatisConfig.Rpc.newBuilder()
+ .setTimeoutMin(
+ TimeDuration.valueOf(
+ CONF
+ .getPartitionRegionRatisRpcLeaderElectionTimeoutMinMs(),
+ TimeUnit.MILLISECONDS))
+ .setTimeoutMax(
+ TimeDuration.valueOf(
+ CONF
+ .getPartitionRegionRatisRpcLeaderElectionTimeoutMaxMs(),
+ TimeUnit.MILLISECONDS))
+ .setRequestTimeout(
+ TimeDuration.valueOf(
+ CONF.getPartitionRegionRatisRequestTimeoutMs(),
+ TimeUnit.MILLISECONDS))
+ .setFirstElectionTimeoutMin(
+ TimeDuration.valueOf(
+ CONF.getRatisFirstElectionTimeoutMinMs(),
+ TimeUnit.MILLISECONDS))
+ .setFirstElectionTimeoutMax(
+ TimeDuration.valueOf(
+ CONF.getRatisFirstElectionTimeoutMaxMs(),
+ TimeUnit.MILLISECONDS))
+ .build())
+ .setRatisConsensus(
+ RatisConfig.RatisConsensus.newBuilder()
+ .setClientRequestTimeoutMillis(
+ CONF.getPartitionRegionRatisRequestTimeoutMs())
+ .setClientMaxRetryAttempt(
+ CONF.getPartitionRegionRatisMaxRetryAttempts())
+ .setClientRetryInitialSleepTimeMs(
+ CONF.getPartitionRegionRatisInitialSleepTimeMs())
+ .setClientRetryMaxSleepTimeMs(
+ CONF.getPartitionRegionRatisMaxSleepTimeMs())
+ .build())
+ .build())
+ .setStorageDir(CONF.getConsensusDir())
+ .build(),
+ gid -> stateMachine)
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ String.format(
+ ConsensusFactory.CONSTRUCT_FAILED_MSG,
+ CONF.getConfigNodeConsensusProtocolClass())));
+ }
consensusImpl.start();
if (SystemPropertiesUtils.isRestarted()) {