This is an automated email from the ASF dual-hosted git repository. tanxinyu pushed a commit to branch jira3195 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9e1cc7596f0a4b9cb74ca2fdd3dbd09ba8deca2a Author: LebronAl <[email protected]> AuthorDate: Tue May 31 00:47:18 2022 +0800 abstract interface && add MultiLeaderConfig --- .../iotdb/confignode/manager/ConsensusManager.java | 7 +- .../apache/iotdb/consensus/ConsensusFactory.java | 11 +- .../iotdb/consensus/config/ConsensusConfig.java | 374 +++++++++++++++++++++ .../multileader/MultiLeaderConsensus.java | 17 +- .../multileader/MultiLeaderServerImpl.java | 15 +- .../multileader/client/DispatchLogHandler.java | 7 +- .../client/MultiLeaderConsensusClientPool.java | 16 +- .../conf/MultiLeaderConsensusConfig.java | 40 --- .../multileader/logdispatcher/LogDispatcher.java | 39 ++- .../multileader/logdispatcher/SyncStatus.java | 8 +- .../multileader/service/MultiLeaderRPCService.java | 12 +- .../iotdb/consensus/ratis/RatisConsensus.java | 18 +- .../consensus/standalone/StandAloneConsensus.java | 7 +- .../multileader/MultiLeaderConsensusTest.java | 7 +- .../multileader/logdispatcher/SyncStatusTest.java | 79 ++--- .../iotdb/consensus/ratis/RatisConsensusTest.java | 7 +- .../iotdb/consensus/standalone/RecoveryTest.java | 7 +- .../standalone/StandAloneConsensusTest.java | 7 +- .../db/consensus/DataRegionConsensusImpl.java | 8 +- .../db/consensus/SchemaRegionConsensusImpl.java | 8 +- 20 files changed, 542 insertions(+), 152 deletions(-) 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 702f5cfa05..a43e176f44 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 @@ -35,6 +35,7 @@ import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.response.ConsensusReadResponse; import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.rpc.TSStatusCode; import org.slf4j.Logger; @@ -88,8 +89,10 @@ public class ConsensusManager { consensusImpl = ConsensusFactory.getConsensusImpl( conf.getConfigNodeConsensusProtocolClass(), - new TEndPoint(conf.getRpcAddress(), conf.getConsensusPort()), - new File(conf.getConsensusDir()), + ConsensusConfig.newBuilder() + .setThisNode(new TEndPoint(conf.getRpcAddress(), conf.getConsensusPort())) + .setStorageDir(new File(conf.getConsensusDir())) + .build(), gid -> stateMachine) .orElseThrow( () -> diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java b/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java index 15e2492b2e..8146fcf3a4 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java @@ -19,12 +19,11 @@ package org.apache.iotdb.consensus; -import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.File; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; import java.util.Optional; @@ -42,15 +41,13 @@ public class ConsensusFactory { private static final Logger logger = LoggerFactory.getLogger(ConsensusFactory.class); public static Optional<IConsensus> getConsensusImpl( - String className, TEndPoint endpoint, File storageDir, IStateMachine.Registry registry) { + String className, ConsensusConfig config, IStateMachine.Registry registry) { try { Class<?> executor = Class.forName(className); Constructor<?> executorConstructor = - executor.getDeclaredConstructor( - TEndPoint.class, File.class, IStateMachine.Registry.class); + executor.getDeclaredConstructor(ConsensusConfig.class, IStateMachine.Registry.class); executorConstructor.setAccessible(true); - return Optional.of( - (IConsensus) executorConstructor.newInstance(endpoint, storageDir, registry)); + return Optional.of((IConsensus) executorConstructor.newInstance(config, registry)); } catch (ClassNotFoundException | NoSuchMethodException | InstantiationException diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java b/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java new file mode 100644 index 0000000000..e152126014 --- /dev/null +++ b/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java @@ -0,0 +1,374 @@ +/* + * 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.consensus.config; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; + +import java.io.File; +import java.util.concurrent.TimeUnit; + +public class ConsensusConfig { + + private final TEndPoint thisNode; + private final File storageDir; + private final StandAloneConfig standAloneConfig; + private final RatisConfig ratisConfig; + private final MultiLeaderConfig multiLeaderConfig; + + public ConsensusConfig( + TEndPoint thisNode, + File storageDir, + StandAloneConfig standAloneConfig, + RatisConfig ratisConfig, + MultiLeaderConfig multiLeaderConfig) { + this.thisNode = thisNode; + this.storageDir = storageDir; + this.standAloneConfig = standAloneConfig; + this.ratisConfig = ratisConfig; + this.multiLeaderConfig = multiLeaderConfig; + } + + public TEndPoint getThisNode() { + return thisNode; + } + + public File getStorageDir() { + return storageDir; + } + + public StandAloneConfig getStandAloneConfig() { + return standAloneConfig; + } + + public RatisConfig getRatisConfig() { + return ratisConfig; + } + + public MultiLeaderConfig getMultiLeaderConfig() { + return multiLeaderConfig; + } + + public static ConsensusConfig.Builder newBuilder() { + return new ConsensusConfig.Builder(); + } + + public static class Builder { + + private TEndPoint thisNode; + private File storageDir; + private StandAloneConfig standAloneConfig; + private RatisConfig ratisConfig; + private MultiLeaderConfig multiLeaderConfig; + + public ConsensusConfig build() { + return new ConsensusConfig( + thisNode, + storageDir, + standAloneConfig != null ? standAloneConfig : StandAloneConfig.newBuilder().build(), + ratisConfig != null ? ratisConfig : RatisConfig.newBuilder().build(), + multiLeaderConfig != null ? multiLeaderConfig : MultiLeaderConfig.newBuilder().build()); + } + + public Builder setThisNode(TEndPoint thisNode) { + this.thisNode = thisNode; + return this; + } + + public Builder setStorageDir(File storageDir) { + this.storageDir = storageDir; + return this; + } + + public Builder setStandAloneConfig(StandAloneConfig standAloneConfig) { + this.standAloneConfig = standAloneConfig; + return this; + } + + public Builder setRatisConfig(RatisConfig ratisConfig) { + this.ratisConfig = ratisConfig; + return this; + } + + public Builder setMultiLeaderConfig(MultiLeaderConfig multiLeaderConfig) { + this.multiLeaderConfig = multiLeaderConfig; + return this; + } + } + + public static class StandAloneConfig { + public static StandAloneConfig.Builder newBuilder() { + return new StandAloneConfig.Builder(); + } + + public static class Builder { + public StandAloneConfig build() { + return new StandAloneConfig(); + } + } + } + + public static class RatisConfig { + + public static RatisConfig.Builder newBuilder() { + return new RatisConfig.Builder(); + } + + public static class Builder { + public RatisConfig build() { + return new RatisConfig(); + } + } + } + + public static class MultiLeaderConfig { + + private final RPC rpc; + private final Replication replication; + + private MultiLeaderConfig(RPC rpc, Replication replication) { + this.rpc = rpc; + this.replication = replication; + } + + public RPC getRpc() { + return rpc; + } + + public Replication getReplication() { + return replication; + } + + public static MultiLeaderConfig.Builder newBuilder() { + return new MultiLeaderConfig.Builder(); + } + + public static class Builder { + + private RPC rpc; + private Replication replication; + + public MultiLeaderConfig build() { + return new MultiLeaderConfig( + rpc != null ? rpc : new RPC.Builder().build(), + replication != null ? replication : new Replication.Builder().build()); + } + + public Builder setRpc(RPC rpc) { + this.rpc = rpc; + return this; + } + + public Builder setReplication(Replication replication) { + this.replication = replication; + return this; + } + } + + public static class RPC { + private final int rpcMaxConcurrentClientNum; + private final int thriftServerAwaitTimeForStopService; + private final boolean isRpcThriftCompressionEnabled; + private final int selectorNumOfClientManager; + private final int connectionTimeoutInMs; + + public RPC( + int rpcMaxConcurrentClientNum, + int thriftServerAwaitTimeForStopService, + boolean isRpcThriftCompressionEnabled, + int selectorNumOfClientManager, + int connectionTimeoutInMs) { + this.rpcMaxConcurrentClientNum = rpcMaxConcurrentClientNum; + this.thriftServerAwaitTimeForStopService = thriftServerAwaitTimeForStopService; + this.isRpcThriftCompressionEnabled = isRpcThriftCompressionEnabled; + this.selectorNumOfClientManager = selectorNumOfClientManager; + this.connectionTimeoutInMs = connectionTimeoutInMs; + } + + public int getRpcMaxConcurrentClientNum() { + return rpcMaxConcurrentClientNum; + } + + public int getThriftServerAwaitTimeForStopService() { + return thriftServerAwaitTimeForStopService; + } + + public boolean isRpcThriftCompressionEnabled() { + return isRpcThriftCompressionEnabled; + } + + public int getSelectorNumOfClientManager() { + return selectorNumOfClientManager; + } + + public int getConnectionTimeoutInMs() { + return connectionTimeoutInMs; + } + + public static RPC.Builder newBuilder() { + return new RPC.Builder(); + } + + public static class Builder { + private int rpcMaxConcurrentClientNum = 65535; + private int thriftServerAwaitTimeForStopService = 60; + private boolean isRpcThriftCompressionEnabled = false; + private int selectorNumOfClientManager = 1; + private int connectionTimeoutInMs = (int) TimeUnit.SECONDS.toMillis(20); + + public Builder setRpcMaxConcurrentClientNum(int rpcMaxConcurrentClientNum) { + this.rpcMaxConcurrentClientNum = rpcMaxConcurrentClientNum; + return this; + } + + public Builder setThriftServerAwaitTimeForStopService( + int thriftServerAwaitTimeForStopService) { + this.thriftServerAwaitTimeForStopService = thriftServerAwaitTimeForStopService; + return this; + } + + public Builder setRpcThriftCompressionEnabled(boolean rpcThriftCompressionEnabled) { + isRpcThriftCompressionEnabled = rpcThriftCompressionEnabled; + return this; + } + + public Builder setSelectorNumOfClientManager(int selectorNumOfClientManager) { + this.selectorNumOfClientManager = selectorNumOfClientManager; + return this; + } + + public Builder setConnectionTimeoutInMs(int connectionTimeoutInMs) { + this.connectionTimeoutInMs = connectionTimeoutInMs; + return this; + } + + public RPC build() { + return new RPC( + rpcMaxConcurrentClientNum, + thriftServerAwaitTimeForStopService, + isRpcThriftCompressionEnabled, + selectorNumOfClientManager, + connectionTimeoutInMs); + } + } + } + + public static class Replication { + private final int maxPendingRequestNumPerNode; + private final int maxRequestPerBatch; + private final int maxPendingBatch; + private final int maxWaitingTimeForAccumulatingBatchInMs; + private final long basicRetryWaitTimeMs; + private final long maxRetryWaitTimeMs; + + private Replication( + int maxPendingRequestNumPerNode, + int maxRequestPerBatch, + int maxPendingBatch, + int maxWaitingTimeForAccumulatingBatchInMs, + long basicRetryWaitTimeMs, + long maxRetryWaitTimeMs) { + this.maxPendingRequestNumPerNode = maxPendingRequestNumPerNode; + this.maxRequestPerBatch = maxRequestPerBatch; + this.maxPendingBatch = maxPendingBatch; + this.maxWaitingTimeForAccumulatingBatchInMs = maxWaitingTimeForAccumulatingBatchInMs; + this.basicRetryWaitTimeMs = basicRetryWaitTimeMs; + this.maxRetryWaitTimeMs = maxRetryWaitTimeMs; + } + + public int getMaxPendingRequestNumPerNode() { + return maxPendingRequestNumPerNode; + } + + public int getMaxRequestPerBatch() { + return maxRequestPerBatch; + } + + public int getMaxPendingBatch() { + return maxPendingBatch; + } + + public int getMaxWaitingTimeForAccumulatingBatchInMs() { + return maxWaitingTimeForAccumulatingBatchInMs; + } + + public long getBasicRetryWaitTimeMs() { + return basicRetryWaitTimeMs; + } + + public long getMaxRetryWaitTimeMs() { + return maxRetryWaitTimeMs; + } + + public static Replication.Builder newBuilder() { + return new Replication.Builder(); + } + + public static class Builder { + private int maxPendingRequestNumPerNode = 1000; + private int maxRequestPerBatch = 100; + private int maxPendingBatch = 50; + private int maxWaitingTimeForAccumulatingBatchInMs = 10; + private long basicRetryWaitTimeMs = TimeUnit.MILLISECONDS.toMillis(100); + private long maxRetryWaitTimeMs = TimeUnit.SECONDS.toMillis(20); + + public Builder setMaxPendingRequestNumPerNode(int maxPendingRequestNumPerNode) { + this.maxPendingRequestNumPerNode = maxPendingRequestNumPerNode; + return this; + } + + public Builder setMaxRequestPerBatch(int maxRequestPerBatch) { + this.maxRequestPerBatch = maxRequestPerBatch; + return this; + } + + public Builder setMaxPendingBatch(int maxPendingBatch) { + this.maxPendingBatch = maxPendingBatch; + return this; + } + + public Builder setMaxWaitingTimeForAccumulatingBatchInMs( + int maxWaitingTimeForAccumulatingBatchInMs) { + this.maxWaitingTimeForAccumulatingBatchInMs = maxWaitingTimeForAccumulatingBatchInMs; + return this; + } + + public Builder setBasicRetryWaitTimeMs(long basicRetryWaitTimeMs) { + this.basicRetryWaitTimeMs = basicRetryWaitTimeMs; + return this; + } + + public Builder setMaxRetryWaitTimeMs(long maxRetryWaitTimeMs) { + this.maxRetryWaitTimeMs = maxRetryWaitTimeMs; + return this; + } + + public Replication build() { + return new Replication( + maxPendingRequestNumPerNode, + maxRequestPerBatch, + maxPendingBatch, + maxWaitingTimeForAccumulatingBatchInMs, + basicRetryWaitTimeMs, + maxRetryWaitTimeMs); + } + } + } + } +} diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java index c9393ad642..1f8440ff29 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java @@ -31,6 +31,8 @@ import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse; import org.apache.iotdb.consensus.common.response.ConsensusReadResponse; import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse; +import org.apache.iotdb.consensus.config.ConsensusConfig; +import org.apache.iotdb.consensus.config.ConsensusConfig.MultiLeaderConfig; import org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException; import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException; import org.apache.iotdb.consensus.exception.IllegalPeerEndpointException; @@ -64,12 +66,14 @@ public class MultiLeaderConsensus implements IConsensus { new ConcurrentHashMap<>(); private final MultiLeaderRPCService service; private final RegisterManager registerManager = new RegisterManager(); + private final MultiLeaderConfig config; - public MultiLeaderConsensus(TEndPoint thisNode, File storageDir, Registry registry) { - this.thisNode = thisNode; - this.storageDir = storageDir; + public MultiLeaderConsensus(ConsensusConfig config, Registry registry) { + this.thisNode = config.getThisNode(); + this.storageDir = config.getStorageDir(); + this.config = config.getMultiLeaderConfig(); this.registry = registry; - this.service = new MultiLeaderRPCService(thisNode); + this.service = new MultiLeaderRPCService(thisNode, config.getMultiLeaderConfig()); } @Override @@ -100,7 +104,8 @@ public class MultiLeaderConsensus implements IConsensus { path.toString(), new Peer(consensusGroupId, thisNode), new ArrayList<>(), - registry.apply(consensusGroupId)); + registry.apply(consensusGroupId), + config); stateMachineMap.put(consensusGroupId, consensus); consensus.start(); } @@ -161,7 +166,7 @@ public class MultiLeaderConsensus implements IConsensus { } MultiLeaderServerImpl impl = new MultiLeaderServerImpl( - path, new Peer(groupId, thisNode), peers, registry.apply(groupId)); + path, new Peer(groupId, thisNode), peers, registry.apply(groupId), config); impl.start(); return impl; }); diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java index 017fe05a15..b6db8a8a6c 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java @@ -26,6 +26,7 @@ import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest; import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.consensus.config.ConsensusConfig.MultiLeaderConfig; import org.apache.iotdb.consensus.multileader.logdispatcher.IndexController; import org.apache.iotdb.consensus.multileader.logdispatcher.LogDispatcher; import org.apache.iotdb.consensus.multileader.thrift.TLogType; @@ -55,9 +56,14 @@ public class MultiLeaderServerImpl { private final List<Peer> configuration; private final IndexController controller; private final LogDispatcher logDispatcher; + private final MultiLeaderConfig config; public MultiLeaderServerImpl( - String storageDir, Peer thisNode, List<Peer> configuration, IStateMachine stateMachine) { + String storageDir, + Peer thisNode, + List<Peer> configuration, + IStateMachine stateMachine, + MultiLeaderConfig config) { this.storageDir = storageDir; this.thisNode = thisNode; this.stateMachine = stateMachine; @@ -69,7 +75,8 @@ public class MultiLeaderServerImpl { } else { persistConfiguration(); } - logDispatcher = new LogDispatcher(this); + this.config = config; + this.logDispatcher = new LogDispatcher(this); } public IStateMachine getStateMachine() { @@ -183,4 +190,8 @@ public class MultiLeaderServerImpl { public IndexController getController() { return controller; } + + public MultiLeaderConfig getConfig() { + return config; + } } diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java index 8cc7d10bef..14a2b4ad58 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java @@ -19,7 +19,6 @@ package org.apache.iotdb.consensus.multileader.client; -import org.apache.iotdb.consensus.multileader.conf.MultiLeaderConsensusConfig; import org.apache.iotdb.consensus.multileader.logdispatcher.LogDispatcher.LogDispatcherThread; import org.apache.iotdb.consensus.multileader.logdispatcher.PendingBatch; import org.apache.iotdb.consensus.multileader.thrift.TSyncLogRes; @@ -79,9 +78,11 @@ public class DispatchLogHandler implements AsyncMethodCallback<TSyncLogRes> { try { long defaultSleepTime = (long) - (MultiLeaderConsensusConfig.BASIC_RETRY_WAIT_TIME_MS * Math.pow(2, retryCount)); + (thread.getConfig().getReplication().getBasicRetryWaitTimeMs() + * Math.pow(2, retryCount)); Thread.sleep( - Math.min(defaultSleepTime, MultiLeaderConsensusConfig.MAX_RETRY_WAIT_TIME_MS)); + Math.min( + defaultSleepTime, thread.getConfig().getReplication().getMaxRetryWaitTimeMs())); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.warn("Unexpected interruption during retry pending batch"); diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java index a43a979694..b17e3d5038 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java @@ -24,7 +24,7 @@ import org.apache.iotdb.commons.client.ClientFactoryProperty; import org.apache.iotdb.commons.client.ClientManager; import org.apache.iotdb.commons.client.ClientPoolProperty; import org.apache.iotdb.commons.client.IClientPoolFactory; -import org.apache.iotdb.consensus.multileader.conf.MultiLeaderConsensusConfig; +import org.apache.iotdb.consensus.config.ConsensusConfig.MultiLeaderConfig; import org.apache.commons.pool2.KeyedObjectPool; import org.apache.commons.pool2.impl.GenericKeyedObjectPool; @@ -35,6 +35,13 @@ public class MultiLeaderConsensusClientPool { public static class AsyncMultiLeaderServiceClientPoolFactory implements IClientPoolFactory<TEndPoint, AsyncMultiLeaderServiceClient> { + + private final MultiLeaderConfig config; + + public AsyncMultiLeaderServiceClientPoolFactory(MultiLeaderConfig config) { + this.config = config; + } + @Override public KeyedObjectPool<TEndPoint, AsyncMultiLeaderServiceClient> createClientPool( ClientManager<TEndPoint, AsyncMultiLeaderServiceClient> manager) { @@ -42,11 +49,10 @@ public class MultiLeaderConsensusClientPool { new AsyncMultiLeaderServiceClient.Factory( manager, new ClientFactoryProperty.Builder() - .setConnectionTimeoutMs(MultiLeaderConsensusConfig.CONNECTION_TIMEOUT_IN_MS) - .setRpcThriftCompressionEnabled( - MultiLeaderConsensusConfig.IS_RPC_THRIFT_COMPRESSION_ENABLED) + .setConnectionTimeoutMs(config.getRpc().getConnectionTimeoutInMs()) + .setRpcThriftCompressionEnabled(config.getRpc().isRpcThriftCompressionEnabled()) .setSelectorNumOfAsyncClientManager( - MultiLeaderConsensusConfig.SELECTOR_NUM_OF_CLIENT_MANAGER) + config.getRpc().getSelectorNumOfClientManager()) .build()), new ClientPoolProperty.Builder<AsyncMultiLeaderServiceClient>().build().getConfig()); } diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/conf/MultiLeaderConsensusConfig.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/conf/MultiLeaderConsensusConfig.java deleted file mode 100644 index 10a547728f..0000000000 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/conf/MultiLeaderConsensusConfig.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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.consensus.multileader.conf; - -import java.util.concurrent.TimeUnit; - -// TODO make it configurable -public class MultiLeaderConsensusConfig { - - private MultiLeaderConsensusConfig() {} - - public static final int RPC_MAX_CONCURRENT_CLIENT_NUM = 65535; - public static final int THRIFT_SERVER_AWAIT_TIME_FOR_STOP_SERVICE = 60; - public static final boolean IS_RPC_THRIFT_COMPRESSION_ENABLED = false; - public static final int SELECTOR_NUM_OF_CLIENT_MANAGER = 1; - public static final int CONNECTION_TIMEOUT_IN_MS = (int) TimeUnit.SECONDS.toMillis(20); - public static final int MAX_PENDING_REQUEST_NUM_PER_NODE = 1000; - public static final int MAX_REQUEST_PER_BATCH = 100; - public static final int MAX_PENDING_BATCH = 50; - public static final int MAX_WAITING_TIME_FOR_ACCUMULATE_BATCH_IN_MS = 10; - public static final long BASIC_RETRY_WAIT_TIME_MS = TimeUnit.MILLISECONDS.toMillis(100); - public static final long MAX_RETRY_WAIT_TIME_MS = TimeUnit.SECONDS.toMillis(20); -} diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java index 3e664ce4e0..075349e2f7 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java @@ -25,11 +25,11 @@ import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.consensus.config.ConsensusConfig.MultiLeaderConfig; import org.apache.iotdb.consensus.multileader.MultiLeaderServerImpl; import org.apache.iotdb.consensus.multileader.client.AsyncMultiLeaderServiceClient; import org.apache.iotdb.consensus.multileader.client.DispatchLogHandler; import org.apache.iotdb.consensus.multileader.client.MultiLeaderConsensusClientPool.AsyncMultiLeaderServiceClientPoolFactory; -import org.apache.iotdb.consensus.multileader.conf.MultiLeaderConsensusConfig; import org.apache.iotdb.consensus.multileader.thrift.TLogBatch; import org.apache.iotdb.consensus.multileader.thrift.TLogType; import org.apache.iotdb.consensus.multileader.thrift.TSyncLogReq; @@ -72,14 +72,14 @@ public class LogDispatcher { this.threads = impl.getConfiguration().stream() .filter(x -> !Objects.equals(x, impl.getThisNode())) - .map(LogDispatcherThread::new) + .map(x -> new LogDispatcherThread(x, impl.getConfig())) .collect(Collectors.toList()); if (!threads.isEmpty()) { this.executorService = IoTDBThreadPoolFactory.newFixedThreadPool(threads.size(), "LogDispatcher"); this.clientManager = new IClientManager.Factory<TEndPoint, AsyncMultiLeaderServiceClient>() - .createClientManager(new AsyncMultiLeaderServiceClientPoolFactory()); + .createClientManager(new AsyncMultiLeaderServiceClientPoolFactory(impl.getConfig())); } } @@ -121,26 +121,29 @@ public class LogDispatcher { public class LogDispatcherThread implements Runnable { - private volatile boolean stopped = false; + private final MultiLeaderConfig config; private final Peer peer; private final IndexController controller; // A sliding window class that manages asynchronously pendingBatches private final SyncStatus syncStatus; // A queue used to receive asynchronous replication requests - private final BlockingQueue<IndexedConsensusRequest> pendingRequest = - new ArrayBlockingQueue<>(MultiLeaderConsensusConfig.MAX_PENDING_REQUEST_NUM_PER_NODE); + private final BlockingQueue<IndexedConsensusRequest> pendingRequest; // A container used to cache requests, whose size changes dynamically private final List<IndexedConsensusRequest> bufferedRequest = new LinkedList<>(); // A reader management class that gets requests from the DataRegion private final ConsensusReqReader reader = (ConsensusReqReader) impl.getStateMachine().read(new GetConsensusReqReaderPlan()); + private volatile boolean stopped = false; - public LogDispatcherThread(Peer peer) { + public LogDispatcherThread(Peer peer, MultiLeaderConfig config) { this.peer = peer; + this.config = config; + this.pendingRequest = + new ArrayBlockingQueue<>(config.getReplication().getMaxPendingRequestNumPerNode()); this.controller = new IndexController( impl.getStorageDir(), Utils.fromTEndPointToString(peer.getEndpoint()), false); - this.syncStatus = new SyncStatus(controller); + this.syncStatus = new SyncStatus(controller, config); } public IndexController getController() { @@ -155,6 +158,10 @@ public class LogDispatcher { return peer; } + public MultiLeaderConfig getConfig() { + return config; + } + public boolean offer(IndexedConsensusRequest request) { return pendingRequest.offer(request); } @@ -177,8 +184,8 @@ public class LogDispatcher { // we may block here if there is no requests in the queue bufferedRequest.add(pendingRequest.take()); // If write pressure is low, we simply sleep a little to reduce the number of RPC - if (pendingRequest.size() <= MultiLeaderConsensusConfig.MAX_REQUEST_PER_BATCH) { - Thread.sleep(MultiLeaderConsensusConfig.MAX_WAITING_TIME_FOR_ACCUMULATE_BATCH_IN_MS); + if (pendingRequest.size() <= config.getReplication().getMaxRequestPerBatch()) { + Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs()); } } // we may block here if the synchronization pipeline is full @@ -200,11 +207,11 @@ public class LogDispatcher { long startIndex = syncStatus.getNextSendingIndex(); long maxIndex = impl.getController().getCurrentIndex(); long endIndex; - if (bufferedRequest.size() <= MultiLeaderConsensusConfig.MAX_REQUEST_PER_BATCH) { + if (bufferedRequest.size() <= config.getReplication().getMaxRequestPerBatch()) { // Use drainTo instead of poll to reduce lock overhead pendingRequest.drainTo( bufferedRequest, - MultiLeaderConsensusConfig.MAX_REQUEST_PER_BATCH - bufferedRequest.size()); + config.getReplication().getMaxRequestPerBatch() - bufferedRequest.size()); } if (bufferedRequest.isEmpty()) { // only execute this after a restart @@ -217,7 +224,7 @@ public class LogDispatcher { // Prevents gap between logs. For example, some requests are not written into the queue when // the queue is full. In this case, requests need to be loaded from the WAL endIndex = constructBatchFromWAL(startIndex, prev.getSearchIndex(), logBatches); - if (logBatches.size() == MultiLeaderConsensusConfig.MAX_REQUEST_PER_BATCH) { + if (logBatches.size() == config.getReplication().getMaxRequestPerBatch()) { batch = new PendingBatch(startIndex, endIndex, logBatches); logger.debug("accumulated a {} from wal", batch); return batch; @@ -226,14 +233,14 @@ public class LogDispatcher { endIndex = prev.getSearchIndex(); iterator.remove(); while (iterator.hasNext() - && logBatches.size() <= MultiLeaderConsensusConfig.MAX_REQUEST_PER_BATCH) { + && logBatches.size() <= config.getReplication().getMaxRequestPerBatch()) { IndexedConsensusRequest current = iterator.next(); // Prevents gap between logs. For example, some logs are not written into the queue when // the queue is full. In this case, requests need to be loaded from the WAL if (current.getSearchIndex() != prev.getSearchIndex() + 1) { endIndex = constructBatchFromWAL(prev.getSearchIndex(), current.getSearchIndex(), logBatches); - if (logBatches.size() == MultiLeaderConsensusConfig.MAX_REQUEST_PER_BATCH) { + if (logBatches.size() == config.getReplication().getMaxRequestPerBatch()) { batch = new PendingBatch(startIndex, endIndex, logBatches); logger.debug("accumulated a {} from queue and wal", batch); return batch; @@ -271,7 +278,7 @@ public class LogDispatcher { private long constructBatchFromWAL( long currentIndex, long maxIndex, List<TLogBatch> logBatches) { while (currentIndex < maxIndex - && logBatches.size() < MultiLeaderConsensusConfig.MAX_REQUEST_PER_BATCH) { + && logBatches.size() < config.getReplication().getMaxRequestPerBatch()) { // TODO iterator IConsensusRequest data = reader.getReq(currentIndex++); if (data != null) { diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatus.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatus.java index fb521b9b07..36c8726005 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatus.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatus.java @@ -20,7 +20,7 @@ package org.apache.iotdb.consensus.multileader.logdispatcher; import org.apache.iotdb.commons.utils.TestOnly; -import org.apache.iotdb.consensus.multileader.conf.MultiLeaderConsensusConfig; +import org.apache.iotdb.consensus.config.ConsensusConfig.MultiLeaderConfig; import java.util.Iterator; import java.util.LinkedList; @@ -28,17 +28,19 @@ import java.util.List; public class SyncStatus { + private final MultiLeaderConfig config; private final IndexController controller; private final List<PendingBatch> pendingBatches = new LinkedList<>(); - public SyncStatus(IndexController controller) { + public SyncStatus(IndexController controller, MultiLeaderConfig config) { this.controller = controller; + this.config = config; } /** we may block here if the synchronization pipeline is full */ public void addNextBatch(PendingBatch batch) throws InterruptedException { synchronized (this) { - while (pendingBatches.size() >= MultiLeaderConsensusConfig.MAX_PENDING_BATCH) { + while (pendingBatches.size() >= config.getReplication().getMaxPendingBatch()) { wait(); } pendingBatches.add(batch); diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCService.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCService.java index 694a7e038b..e4a08cf57b 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCService.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCService.java @@ -26,7 +26,7 @@ import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.ThriftService; import org.apache.iotdb.commons.service.ThriftServiceThread; -import org.apache.iotdb.consensus.multileader.conf.MultiLeaderConsensusConfig; +import org.apache.iotdb.consensus.config.ConsensusConfig.MultiLeaderConfig; import org.apache.iotdb.consensus.multileader.thrift.MultiLeaderConsensusIService; import java.lang.reflect.InvocationTargetException; @@ -34,10 +34,12 @@ import java.lang.reflect.InvocationTargetException; public class MultiLeaderRPCService extends ThriftService implements MultiLeaderRPCServiceMBean { private final TEndPoint thisNode; + private final MultiLeaderConfig config; private MultiLeaderRPCServiceProcessor multiLeaderRPCServiceProcessor; - public MultiLeaderRPCService(TEndPoint thisNode) { + public MultiLeaderRPCService(TEndPoint thisNode, MultiLeaderConfig config) { this.thisNode = thisNode; + this.config = config; } @Override @@ -73,10 +75,10 @@ public class MultiLeaderRPCService extends ThriftService implements MultiLeaderR ThreadName.MULTI_LEADER_CONSENSUS_RPC_CLIENT.getName(), getBindIP(), getBindPort(), - MultiLeaderConsensusConfig.RPC_MAX_CONCURRENT_CLIENT_NUM, - MultiLeaderConsensusConfig.THRIFT_SERVER_AWAIT_TIME_FOR_STOP_SERVICE, + config.getRpc().getRpcMaxConcurrentClientNum(), + config.getRpc().getThriftServerAwaitTimeForStopService(), new MultiLeaderRPCServiceHandler(multiLeaderRPCServiceProcessor), - MultiLeaderConsensusConfig.IS_RPC_THRIFT_COMPRESSION_ENABLED); + config.getRpc().isRpcThriftCompressionEnabled()); } catch (RPCServiceException e) { throw new IllegalAccessException(e.getMessage()); } diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java index 0c96034f7c..54383deb6f 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java @@ -36,6 +36,7 @@ import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse; import org.apache.iotdb.consensus.common.response.ConsensusReadResponse; import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException; import org.apache.iotdb.consensus.exception.PeerAlreadyInConsensusGroupException; @@ -67,7 +68,6 @@ import org.apache.thrift.TException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.File; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; @@ -94,6 +94,8 @@ class RatisConsensus implements IConsensus { private final RaftProperties properties = new RaftProperties(); private final RaftClientRpc clientRpc; + private final ConsensusConfig config; + private final IClientManager<RaftGroup, RatisClient> clientManager = new IClientManager.Factory<RaftGroup, RatisClient>() .createClientManager(new RatisClientPoolFactory()); @@ -109,17 +111,15 @@ class RatisConsensus implements IConsensus { // TODO make it configurable private static final int DEFAULT_WAIT_LEADER_READY_TIMEOUT = (int) TimeUnit.SECONDS.toMillis(20); - /** - * @param ratisStorageDir different groups of RatisConsensus Peer all share ratisStorageDir as - * root dir - */ - public RatisConsensus(TEndPoint endpoint, File ratisStorageDir, IStateMachine.Registry registry) + public RatisConsensus(ConsensusConfig config, IStateMachine.Registry registry) throws IOException { - myself = Utils.fromTEndPointAndPriorityToRaftPeer(endpoint, DEFAULT_PRIORITY); + this.config = config; + myself = Utils.fromTEndPointAndPriorityToRaftPeer(config.getThisNode(), DEFAULT_PRIORITY); System.setProperty( "org.apache.ratis.thirdparty.io.netty.allocator.useCacheForAllThreads", "false"); - RaftServerConfigKeys.setStorageDir(properties, Collections.singletonList(ratisStorageDir)); + RaftServerConfigKeys.setStorageDir( + properties, Collections.singletonList(config.getStorageDir())); RaftServerConfigKeys.Snapshot.setAutoTriggerEnabled(properties, true); // TODO make this configurable so that RatisConsensusTest can trigger multiple snapshot process // RaftServerConfigKeys.Snapshot.setAutoTriggerThreshold(properties, 20); @@ -131,7 +131,7 @@ class RatisConsensus implements IConsensus { RaftClientConfigKeys.Rpc.setRequestTimeout( properties, TimeDuration.valueOf(20, TimeUnit.SECONDS)); - GrpcConfigKeys.Server.setPort(properties, endpoint.getPort()); + GrpcConfigKeys.Server.setPort(properties, config.getThisNode().getPort()); clientRpc = new GrpcFactory(new Parameters()).newRaftClientRpc(ClientId.randomId(), properties); server = diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/standalone/StandAloneConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/standalone/StandAloneConsensus.java index 3604d29059..53c0d3996b 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/standalone/StandAloneConsensus.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/standalone/StandAloneConsensus.java @@ -29,6 +29,7 @@ import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse; import org.apache.iotdb.consensus.common.response.ConsensusReadResponse; import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException; import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException; import org.apache.iotdb.consensus.exception.IllegalPeerEndpointException; @@ -63,9 +64,9 @@ class StandAloneConsensus implements IConsensus { private final Map<ConsensusGroupId, StandAloneServerImpl> stateMachineMap = new ConcurrentHashMap<>(); - public StandAloneConsensus(TEndPoint thisNode, File storageDir, Registry registry) { - this.thisNode = thisNode; - this.storageDir = storageDir; + public StandAloneConsensus(ConsensusConfig config, Registry registry) { + this.thisNode = config.getThisNode(); + this.storageDir = config.getStorageDir(); this.registry = registry; } diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java index 770031bee1..264d927cbe 100644 --- a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java +++ b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java @@ -31,6 +31,7 @@ import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest; import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.consensus.multileader.logdispatcher.IndexController; import org.apache.iotdb.consensus.multileader.thrift.TLogType; import org.apache.iotdb.consensus.multileader.wal.ConsensusReqReader; @@ -100,8 +101,10 @@ public class MultiLeaderConsensusTest { (MultiLeaderConsensus) ConsensusFactory.getConsensusImpl( ConsensusFactory.MultiLeaderConsensus, - peers.get(i).getEndpoint(), - peersStorage.get(i), + ConsensusConfig.newBuilder() + .setThisNode(peers.get(i).getEndpoint()) + .setStorageDir(peersStorage.get(i)) + .build(), groupId -> stateMachines.get(finalI)) .orElseThrow( () -> diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java index cba5147872..9bff9000f3 100644 --- a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java +++ b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java @@ -19,7 +19,7 @@ package org.apache.iotdb.consensus.multileader.logdispatcher; -import org.apache.iotdb.consensus.multileader.conf.MultiLeaderConsensusConfig; +import org.apache.iotdb.consensus.config.ConsensusConfig.MultiLeaderConfig; import org.apache.ratis.util.FileUtils; import org.junit.After; @@ -39,6 +39,7 @@ public class SyncStatusTest { private static final File storageDir = new File("target" + java.io.File.separator + "test"); private static final String prefix = "version"; + private static final MultiLeaderConfig config = new MultiLeaderConfig.Builder().build(); @Before public void setUp() throws IOException { @@ -56,22 +57,22 @@ public class SyncStatusTest { IndexController controller = new IndexController(storageDir.getAbsolutePath(), prefix, true); Assert.assertEquals(0, controller.getCurrentIndex()); - SyncStatus status = new SyncStatus(controller); + SyncStatus status = new SyncStatus(controller, config); List<PendingBatch> batchList = new ArrayList<>(); - for (long i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH; i++) { + for (long i = 0; i < config.getReplication().getMaxPendingBatch(); i++) { PendingBatch batch = new PendingBatch(i, i, Collections.emptyList()); batchList.add(batch); status.addNextBatch(batch); } - for (int i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH; i++) { + for (int i = 0; i < config.getReplication().getMaxPendingBatch(); i++) { status.removeBatch(batchList.get(i)); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1 - i, status.getPendingBatches().size()); + config.getReplication().getMaxPendingBatch() - 1 - i, status.getPendingBatches().size()); Assert.assertEquals(i, controller.getCurrentIndex()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch(), status.getNextSendingIndex()); } } @@ -82,29 +83,29 @@ public class SyncStatusTest { Assert.assertEquals(0, controller.getCurrentIndex()); Assert.assertEquals(0, controller.getLastFlushedIndex()); - SyncStatus status = new SyncStatus(controller); + SyncStatus status = new SyncStatus(controller, config); List<PendingBatch> batchList = new ArrayList<>(); - for (long i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH; i++) { + for (long i = 0; i < config.getReplication().getMaxPendingBatch(); i++) { PendingBatch batch = new PendingBatch(i, i, Collections.emptyList()); batchList.add(batch); status.addNextBatch(batch); } - for (int i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1; i++) { - status.removeBatch(batchList.get(MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1 - i)); + for (int i = 0; i < config.getReplication().getMaxPendingBatch() - 1; i++) { + status.removeBatch(batchList.get(config.getReplication().getMaxPendingBatch() - 1 - i)); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getPendingBatches().size()); + config.getReplication().getMaxPendingBatch(), status.getPendingBatches().size()); Assert.assertEquals(0, controller.getCurrentIndex()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch(), status.getNextSendingIndex()); } status.removeBatch(batchList.get(0)); Assert.assertEquals(0, status.getPendingBatches().size()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1, controller.getCurrentIndex()); - Assert.assertEquals(MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch() - 1, controller.getCurrentIndex()); + Assert.assertEquals(config.getReplication().getMaxPendingBatch(), status.getNextSendingIndex()); } /** Confirm success first from front to back, then back to front */ @@ -114,39 +115,39 @@ public class SyncStatusTest { Assert.assertEquals(0, controller.getCurrentIndex()); Assert.assertEquals(0, controller.getLastFlushedIndex()); - SyncStatus status = new SyncStatus(controller); + SyncStatus status = new SyncStatus(controller, config); List<PendingBatch> batchList = new ArrayList<>(); - for (long i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH; i++) { + for (long i = 0; i < config.getReplication().getMaxPendingBatch(); i++) { PendingBatch batch = new PendingBatch(i, i, Collections.emptyList()); batchList.add(batch); status.addNextBatch(batch); } - for (int i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH / 2; i++) { + for (int i = 0; i < config.getReplication().getMaxPendingBatch() / 2; i++) { status.removeBatch(batchList.get(i)); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1 - i, status.getPendingBatches().size()); + config.getReplication().getMaxPendingBatch() - 1 - i, status.getPendingBatches().size()); Assert.assertEquals(i, controller.getCurrentIndex()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch(), status.getNextSendingIndex()); } - for (int i = MultiLeaderConsensusConfig.MAX_PENDING_BATCH / 2 + 1; - i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH; + for (int i = config.getReplication().getMaxPendingBatch() / 2 + 1; + i < config.getReplication().getMaxPendingBatch(); i++) { status.removeBatch(batchList.get(i)); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH / 2, status.getPendingBatches().size()); + config.getReplication().getMaxPendingBatch() / 2, status.getPendingBatches().size()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch(), status.getNextSendingIndex()); } - status.removeBatch(batchList.get(MultiLeaderConsensusConfig.MAX_PENDING_BATCH / 2)); + status.removeBatch(batchList.get(config.getReplication().getMaxPendingBatch() / 2)); Assert.assertEquals(0, status.getPendingBatches().size()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1, controller.getCurrentIndex()); - Assert.assertEquals(MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch() - 1, controller.getCurrentIndex()); + Assert.assertEquals(config.getReplication().getMaxPendingBatch(), status.getNextSendingIndex()); } /** Test Blocking while addNextBatch */ @@ -155,22 +156,22 @@ public class SyncStatusTest { IndexController controller = new IndexController(storageDir.getAbsolutePath(), prefix, true); Assert.assertEquals(0, controller.getCurrentIndex()); - SyncStatus status = new SyncStatus(controller); + SyncStatus status = new SyncStatus(controller, config); List<PendingBatch> batchList = new ArrayList<>(); - for (long i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH; i++) { + for (long i = 0; i < config.getReplication().getMaxPendingBatch(); i++) { PendingBatch batch = new PendingBatch(i, i, Collections.emptyList()); batchList.add(batch); status.addNextBatch(batch); } - for (int i = 0; i < MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1; i++) { - status.removeBatch(batchList.get(MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1 - i)); + for (int i = 0; i < config.getReplication().getMaxPendingBatch() - 1; i++) { + status.removeBatch(batchList.get(config.getReplication().getMaxPendingBatch() - 1 - i)); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getPendingBatches().size()); + config.getReplication().getMaxPendingBatch(), status.getPendingBatches().size()); Assert.assertEquals(0, controller.getCurrentIndex()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch(), status.getNextSendingIndex()); } CompletableFuture<Boolean> future = @@ -178,8 +179,8 @@ public class SyncStatusTest { () -> { PendingBatch batch = new PendingBatch( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, - MultiLeaderConsensusConfig.MAX_PENDING_BATCH, + config.getReplication().getMaxPendingBatch(), + config.getReplication().getMaxPendingBatch(), Collections.emptyList()); batchList.add(batch); try { @@ -198,14 +199,14 @@ public class SyncStatusTest { Assert.assertTrue(future.get()); Assert.assertEquals(1, status.getPendingBatches().size()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH - 1, controller.getCurrentIndex()); + config.getReplication().getMaxPendingBatch() - 1, controller.getCurrentIndex()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH + 1, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch() + 1, status.getNextSendingIndex()); - status.removeBatch(batchList.get(MultiLeaderConsensusConfig.MAX_PENDING_BATCH)); + status.removeBatch(batchList.get(config.getReplication().getMaxPendingBatch())); Assert.assertEquals(0, status.getPendingBatches().size()); - Assert.assertEquals(MultiLeaderConsensusConfig.MAX_PENDING_BATCH, controller.getCurrentIndex()); + Assert.assertEquals(config.getReplication().getMaxPendingBatch(), controller.getCurrentIndex()); Assert.assertEquals( - MultiLeaderConsensusConfig.MAX_PENDING_BATCH + 1, status.getNextSendingIndex()); + config.getReplication().getMaxPendingBatch() + 1, status.getNextSendingIndex()); } } diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java index b114c2305a..d44c212141 100644 --- a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java +++ b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java @@ -28,6 +28,7 @@ import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest; import org.apache.iotdb.consensus.common.response.ConsensusReadResponse; import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.ratis.util.FileUtils; import org.junit.After; @@ -63,8 +64,10 @@ public class RatisConsensusTest { servers.add( ConsensusFactory.getConsensusImpl( ConsensusFactory.RatisConsensus, - peers.get(i).getEndpoint(), - peersStorage.get(i), + ConsensusConfig.newBuilder() + .setThisNode(peers.get(i).getEndpoint()) + .setStorageDir(peersStorage.get(i)) + .build(), groupId -> stateMachines.get(finalI)) .orElseThrow( () -> diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/standalone/RecoveryTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/standalone/RecoveryTest.java index de4442ed79..b133f83939 100644 --- a/consensus/src/test/java/org/apache/iotdb/consensus/standalone/RecoveryTest.java +++ b/consensus/src/test/java/org/apache/iotdb/consensus/standalone/RecoveryTest.java @@ -25,6 +25,7 @@ import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException; import org.apache.ratis.util.FileUtils; @@ -45,8 +46,10 @@ public class RecoveryTest { consensusImpl = ConsensusFactory.getConsensusImpl( ConsensusFactory.StandAloneConsensus, - new TEndPoint("0.0.0.0", 9000), - new File("target" + java.io.File.separator + "recovery"), + ConsensusConfig.newBuilder() + .setThisNode(new TEndPoint("0.0.0.0", 9000)) + .setStorageDir(new File("target" + java.io.File.separator + "recovery")) + .build(), gid -> new EmptyStateMachine()) .orElseThrow( () -> diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java index 5c3ae96453..c04adb98ff 100644 --- a/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java +++ b/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java @@ -34,6 +34,7 @@ import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest; import org.apache.iotdb.consensus.common.request.IConsensusRequest; import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse; import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException; import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException; import org.apache.iotdb.consensus.exception.IllegalPeerEndpointException; @@ -123,8 +124,10 @@ public class StandAloneConsensusTest { consensusImpl = ConsensusFactory.getConsensusImpl( ConsensusFactory.StandAloneConsensus, - new TEndPoint("0.0.0.0", 6667), - new File("target" + java.io.File.separator + "standalone"), + ConsensusConfig.newBuilder() + .setThisNode(new TEndPoint("0.0.0.0", 6667)) + .setStorageDir(new File("target" + java.io.File.separator + "standalone")) + .build(), gid -> { switch (gid.getType()) { case SchemaRegion: diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java b/server/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java index 2007f1af59..1557163ead 100644 --- a/server/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java @@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.IConsensus; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.consensus.statemachine.DataRegionStateMachine; @@ -48,8 +49,11 @@ public class DataRegionConsensusImpl { private static final IConsensus INSTANCE = ConsensusFactory.getConsensusImpl( conf.getDataRegionConsensusProtocolClass(), - new TEndPoint(conf.getInternalIp(), conf.getDataRegionConsensusPort()), - new File(conf.getDataRegionConsensusDir()), + ConsensusConfig.newBuilder() + .setThisNode( + new TEndPoint(conf.getInternalIp(), conf.getDataRegionConsensusPort())) + .setStorageDir(new File(conf.getDataRegionConsensusDir())) + .build(), gid -> new DataRegionStateMachine( StorageEngineV2.getInstance().getDataRegion((DataRegionId) gid))) diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java b/server/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java index 1bde912992..84c9da9b4f 100644 --- a/server/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java @@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.consensus.SchemaRegionId; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.IConsensus; +import org.apache.iotdb.consensus.config.ConsensusConfig; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.consensus.statemachine.SchemaRegionStateMachine; @@ -48,8 +49,11 @@ public class SchemaRegionConsensusImpl { private static final IConsensus INSTANCE = ConsensusFactory.getConsensusImpl( conf.getSchemaRegionConsensusProtocolClass(), - new TEndPoint(conf.getInternalIp(), conf.getSchemaRegionConsensusPort()), - new File(conf.getSchemaRegionConsensusDir()), + ConsensusConfig.newBuilder() + .setThisNode( + new TEndPoint(conf.getInternalIp(), conf.getSchemaRegionConsensusPort())) + .setStorageDir(new File(conf.getSchemaRegionConsensusDir())) + .build(), gid -> new SchemaRegionStateMachine( SchemaEngine.getInstance().getSchemaRegion((SchemaRegionId) gid)))
