From f3e63e215cf8420c1fbf1a175f80744d447ab7b9 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Thu, 13 Aug 2026 16:48:56 +0800 Subject: [PATCH 1/2] [TDB-271][to dev/1.3] Wait for consensus before region requests --- iotdb-core/datanode/pom.xml | 1 + .../db/i18n/ConsensusReadinessMessages.java | 32 ++++ .../db/i18n/ConsensusReadinessMessages.java | 32 ++++ .../impl/DataNodeInternalRPCServiceImpl.java | 90 ++++++++++- .../iotdb/db/service/ConsensusReadiness.java | 30 ++++ .../org/apache/iotdb/db/service/DataNode.java | 48 +++++- .../service/DataNodeInternalRPCService.java | 45 +++++- .../thrift/impl/ConsensusWaitTest.java | 144 ++++++++++++++++++ .../db/service/ConsensusReadinessTest.java | 69 +++++++++ .../DataNodeInternalRPCServiceImplTest.java | 15 +- .../DataNodeInternalRPCServiceTest.java | 48 ++++++ pom.xml | 7 + 12 files changed, 552 insertions(+), 9 deletions(-) create mode 100644 iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java create mode 100644 iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ConsensusReadiness.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/ConsensusWaitTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/ConsensusReadinessTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceTest.java diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml index 299e50f1be93..12be9e60f80c 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -499,6 +499,7 @@ ${project.build.directory}/generated-sources/freemarker + ${project.basedir}/src/main/i18n/${i18n.locale} diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java new file mode 100644 index 000000000000..111629e66383 --- /dev/null +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.i18n; + +public final class ConsensusReadinessMessages { + + public static final String + LOG_CONSENSUS_IS_NOT_INITIALIZED_REJECTING_THE_REGION_TOPOLOGY_REQUEST_AFTER_WAITING_UP_TO_ARG_MS_7035CB1C = + "Consensus is not initialized; rejecting the region topology request after waiting up to {} ms"; + public static final String + MESSAGE_CONSENSUS_IS_NOT_INITIALIZED_REGION_TOPOLOGY_REQUEST_REJECTED_AFTER_WAITING_UP_TO_ARG_MS_30E1CBCC = + "Consensus is not initialized; region topology request rejected after waiting up to %d ms"; + + private ConsensusReadinessMessages() {} +} diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java new file mode 100644 index 000000000000..832055894a89 --- /dev/null +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ConsensusReadinessMessages.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.i18n; + +public final class ConsensusReadinessMessages { + + public static final String + LOG_CONSENSUS_IS_NOT_INITIALIZED_REJECTING_THE_REGION_TOPOLOGY_REQUEST_AFTER_WAITING_UP_TO_ARG_MS_7035CB1C = + "Consensus 尚未初始化；等待最多 {} 毫秒后拒绝 region 拓扑请求"; + public static final String + MESSAGE_CONSENSUS_IS_NOT_INITIALIZED_REGION_TOPOLOGY_REQUEST_REJECTED_AFTER_WAITING_UP_TO_ARG_MS_30E1CBCC = + "Consensus 尚未初始化；等待最多 %d 毫秒后已拒绝 region 拓扑请求"; + + private ConsensusReadinessMessages() {} +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index 99e314affa04..da045e5103a7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -82,6 +82,7 @@ import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; import org.apache.iotdb.db.consensus.SchemaRegionConsensusImpl; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.i18n.ConsensusReadinessMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.protocol.client.cn.DnToCnInternalServiceAsyncRequestManager; @@ -149,6 +150,7 @@ import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.reader.ISchemaReader; import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUpdateType; +import org.apache.iotdb.db.service.ConsensusReadiness; import org.apache.iotdb.db.service.DataNode; import org.apache.iotdb.db.service.RegionMigrateService; import org.apache.iotdb.db.service.metrics.FileMetrics; @@ -319,7 +321,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface private final SchemaEngine schemaEngine = SchemaEngine.getInstance(); private final StorageEngine storageEngine = StorageEngine.getInstance(); - private final DataNodeRegionManager regionManager = DataNodeRegionManager.getInstance(); + private final DataNodeRegionManager regionManager; private final DataNodeSpaceQuotaManager spaceQuotaManager = DataNodeSpaceQuotaManager.getInstance(); @@ -330,8 +332,11 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface private static final long TEST_CONNECTION_TIMEOUT_MS = CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS(); private static final int TEST_CONNECTION_RETRY_NUM = 1; + private static final long DEFAULT_CONSENSUS_WAIT_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30); private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + private final ConsensusReadiness consensusReadiness; + private final long consensusWaitTimeoutMs; private final ExecutorService schemaExecutor = new WrappedThreadPoolExecutor( @@ -347,10 +352,49 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface private static final String SYSTEM = "system"; - public DataNodeInternalRPCServiceImpl() { + DataNodeInternalRPCServiceImpl( + ConsensusReadiness consensusReadiness, long consensusWaitTimeoutMs) { + this(consensusReadiness, consensusWaitTimeoutMs, DataNodeRegionManager.getInstance()); + } + + DataNodeInternalRPCServiceImpl( + ConsensusReadiness consensusReadiness, + long consensusWaitTimeoutMs, + DataNodeRegionManager regionManager) { super(); partitionFetcher = ClusterPartitionFetcher.getInstance(); schemaFetcher = ClusterSchemaFetcher.getInstance(); + this.consensusReadiness = consensusReadiness; + this.consensusWaitTimeoutMs = consensusWaitTimeoutMs; + this.regionManager = regionManager; + } + + public DataNodeInternalRPCServiceImpl(ConsensusReadiness consensusReadiness) { + this(consensusReadiness, DEFAULT_CONSENSUS_WAIT_TIMEOUT_MS); + } + + private TSStatus waitForConsensusStarted() { + if (consensusReadiness.isAllConsensusStarted()) { + return null; + } + try { + if (consensusReadiness.awaitAllConsensusStarted( + consensusWaitTimeoutMs, TimeUnit.MILLISECONDS)) { + return null; + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + LOGGER.warn( + ConsensusReadinessMessages + .LOG_CONSENSUS_IS_NOT_INITIALIZED_REJECTING_THE_REGION_TOPOLOGY_REQUEST_AFTER_WAITING_UP_TO_ARG_MS_7035CB1C, + consensusWaitTimeoutMs); + return RpcUtils.getStatus( + TSStatusCode.CONSENSUS_NOT_INITIALIZED, + String.format( + ConsensusReadinessMessages + .MESSAGE_CONSENSUS_IS_NOT_INITIALIZED_REGION_TOPOLOGY_REQUEST_REJECTED_AFTER_WAITING_UP_TO_ARG_MS_30E1CBCC, + consensusWaitTimeoutMs)); } @Override @@ -525,11 +569,19 @@ private TLoadResp createTLoadResp(TSStatus resultStatus) { @Override public TSStatus createSchemaRegion(TCreateSchemaRegionReq req) { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } return regionManager.createSchemaRegion(req.getRegionReplicaSet(), req.getStorageGroup()); } @Override public TSStatus createDataRegion(TCreateDataRegionReq req) { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } return regionManager.createDataRegion(req.getRegionReplicaSet(), req.getStorageGroup()); } @@ -2097,6 +2149,10 @@ public TSStatus updateTemplate(final TUpdateTemplateReq req) { @Override public TSStatus deleteRegion(TConsensusGroupId tconsensusGroupId) { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } ConsensusGroupId consensusGroupId = ConsensusGroupId.Factory.createFromTConsensusGroupId(tconsensusGroupId); if (consensusGroupId instanceof DataRegionId) { @@ -2125,6 +2181,12 @@ public TRegionLeaderChangeResp changeRegionLeader(TRegionLeaderChangeReq req) { LOGGER.info("[ChangeRegionLeader] {}", req); TRegionLeaderChangeResp resp = new TRegionLeaderChangeResp(); + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + resp.setStatus(consensusStatus); + return resp; + } + TSStatus successStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); TConsensusGroupId tgId = req.getRegionId(); ConsensusGroupId regionId = ConsensusGroupId.Factory.createFromTConsensusGroupId(tgId); @@ -2194,6 +2256,10 @@ private boolean isLeader(ConsensusGroupId regionId) { @Override public TSStatus createNewRegionPeer(TCreatePeerReq req) { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } ConsensusGroupId regionId = ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getRegionId()); List peers = @@ -2214,6 +2280,10 @@ public TSStatus createNewRegionPeer(TCreatePeerReq req) { @Override public TSStatus addRegionPeer(TMaintainPeerReq req) { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } TConsensusGroupId regionId = req.getRegionId(); String selectedDataNodeIP = req.getDestNode().getInternalEndPoint().getIp(); boolean submitSucceed = RegionMigrateService.getInstance().submitAddRegionPeerTask(req); @@ -2232,6 +2302,10 @@ public TSStatus addRegionPeer(TMaintainPeerReq req) { @Override public TSStatus removeRegionPeer(TMaintainPeerReq req) { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } TConsensusGroupId regionId = req.getRegionId(); String selectedDataNodeIP = req.getDestNode().getInternalEndPoint().getIp(); boolean submitSucceed = RegionMigrateService.getInstance().submitRemoveRegionPeerTask(req); @@ -2250,6 +2324,10 @@ public TSStatus removeRegionPeer(TMaintainPeerReq req) { @Override public TSStatus deleteOldRegionPeer(TMaintainPeerReq req) { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } TConsensusGroupId regionId = req.getRegionId(); String selectedDataNodeIP = req.getDestNode().getInternalEndPoint().getIp(); boolean submitSucceed = RegionMigrateService.getInstance().submitDeleteOldRegionPeerTask(req); @@ -2269,6 +2347,10 @@ public TSStatus deleteOldRegionPeer(TMaintainPeerReq req) { // TODO: return which DataNode fail @Override public TSStatus resetPeerList(TResetPeerListReq req) throws TException { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } return RegionMigrateService.getInstance().resetPeerList(req); } @@ -2279,6 +2361,10 @@ public TRegionMigrateResult getRegionMaintainResult(long taskId) throws TExcepti @Override public TSStatus notifyRegionMigration(TNotifyRegionMigrationReq req) throws TException { + TSStatus consensusStatus = waitForConsensusStarted(); + if (consensusStatus != null) { + return consensusStatus; + } RegionMigrateService.getInstance().notifyRegionMigration(req); return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ConsensusReadiness.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ConsensusReadiness.java new file mode 100644 index 000000000000..42ace2f2a313 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ConsensusReadiness.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.service; + +import java.util.concurrent.TimeUnit; + +/** Read-only context for the startup state of SchemaRegion and DataRegion consensus. */ +public interface ConsensusReadiness { + + boolean isAllConsensusStarted(); + + boolean awaitAllConsensusStarted(long timeout, TimeUnit unit) throws InterruptedException; +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java index 8967504e25a0..4c0790e70953 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java @@ -128,6 +128,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -167,8 +168,10 @@ public class DataNode extends ServerCommandLine implements DataNodeMBean { private static final String REGISTER_INTERRUPTION = "Unexpected interruption when waiting to register to the cluster"; - private boolean schemaRegionConsensusStarted = false; - private boolean dataRegionConsensusStarted = false; + private volatile boolean schemaRegionConsensusStarted = false; + private volatile boolean dataRegionConsensusStarted = false; + private final ConsensusReadinessContext consensusReadinessContext = + new ConsensusReadinessContext(); private long schemaEngineRecoveryTimeInMs; private static Thread watcherThread; @@ -292,6 +295,7 @@ protected void start() { "DataRegion consensus start successfully, which takes {} ms.", (dataRegionEndTime - dataRegionStartTime)); dataRegionConsensusStarted = true; + consensusReadinessContext.markDataRegionConsensusStarted(); } } catch (StartupException | IOException e) { @@ -721,6 +725,7 @@ private void active() throws StartupException { "SchemaRegion consensus start successfully, which takes {} ms.", (schemaRegionEndTime - startTime)); schemaRegionConsensusStarted = true; + consensusReadinessContext.markSchemaRegionConsensusStarted(); if (!isUsingPipeConsensus()) { DataRegionConsensusImpl.getInstance().start(); long dataRegionEndTime = System.currentTimeMillis(); @@ -728,6 +733,7 @@ private void active() throws StartupException { "DataRegion consensus start successfully, which takes {} ms.", (dataRegionEndTime - schemaRegionEndTime)); dataRegionConsensusStarted = true; + consensusReadinessContext.markDataRegionConsensusStarted(); } } catch (IOException e) { throw new StartupException(e); @@ -810,7 +816,9 @@ private void setUp() throws StartupException { /** Set up RPC and protocols after DataNode is available */ private void setUpRPCService() throws StartupException { // Start InternalRPCService to indicate that the current DataNode can accept cluster scheduling - registerManager.register(DataNodeInternalRPCService.getInstance()); + DataNodeInternalRPCService internalRPCService = DataNodeInternalRPCService.getInstance(); + internalRPCService.setConsensusReadiness(consensusReadinessContext); + registerManager.register(internalRPCService); // Notice: During the period between starting the internal RPC service // and starting the client RPC service , some requests may fail because @@ -1235,4 +1243,38 @@ private DataNodeHolder() { // Empty constructor } } + + static class ConsensusReadinessContext implements ConsensusReadiness { + + private final CountDownLatch allConsensusStarted = new CountDownLatch(1); + private volatile boolean schemaRegionConsensusStarted; + private volatile boolean dataRegionConsensusStarted; + + void markSchemaRegionConsensusStarted() { + schemaRegionConsensusStarted = true; + signalIfReady(); + } + + void markDataRegionConsensusStarted() { + dataRegionConsensusStarted = true; + signalIfReady(); + } + + private void signalIfReady() { + if (schemaRegionConsensusStarted && dataRegionConsensusStarted) { + allConsensusStarted.countDown(); + } + } + + @Override + public boolean isAllConsensusStarted() { + return allConsensusStarted.getCount() == 0; + } + + @Override + public boolean awaitAllConsensusStarted(long timeout, TimeUnit unit) + throws InterruptedException { + return allConsensusStarted.await(timeout, unit); + } + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java index 080967e3a69e..34a72e1d60dd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java @@ -33,12 +33,43 @@ import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService.Processor; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; +import java.util.Objects; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; public class DataNodeInternalRPCService extends ThriftService implements DataNodeInternalRPCServiceMBean { + private static final ConsensusReadiness NOT_READY = + new ConsensusReadiness() { + @Override + public boolean isAllConsensusStarted() { + return false; + } + + @Override + public boolean awaitAllConsensusStarted(long timeout, TimeUnit unit) { + return false; + } + }; + private final AtomicReference impl = new AtomicReference<>(); + private final AtomicReference consensusReadiness = + new AtomicReference<>(NOT_READY); + + private final ConsensusReadiness delegatingConsensusReadiness = + new ConsensusReadiness() { + @Override + public boolean isAllConsensusStarted() { + return consensusReadiness.get().isAllConsensusStarted(); + } + + @Override + public boolean awaitAllConsensusStarted(long timeout, TimeUnit unit) + throws InterruptedException { + return consensusReadiness.get().awaitAllConsensusStarted(timeout, unit); + } + }; private DataNodeInternalRPCService() {} @@ -49,9 +80,9 @@ public ServiceType getID() { @Override public void initTProcessor() { - impl.compareAndSet(null, new DataNodeInternalRPCServiceImpl()); + DataNodeInternalRPCServiceImpl service = getImpl(); initSyncedServiceImpl(null); - processor = new Processor<>(impl.get()); + processor = new Processor<>(service); } @Override @@ -90,10 +121,18 @@ public int getBindPort() { } public DataNodeInternalRPCServiceImpl getImpl() { - impl.compareAndSet(null, new DataNodeInternalRPCServiceImpl()); + impl.compareAndSet(null, new DataNodeInternalRPCServiceImpl(delegatingConsensusReadiness)); return impl.get(); } + void setConsensusReadiness(ConsensusReadiness consensusReadiness) { + this.consensusReadiness.set(Objects.requireNonNull(consensusReadiness)); + } + + ConsensusReadiness getConsensusReadiness() { + return delegatingConsensusReadiness; + } + private static class DataNodeInternalRPCServiceHolder { private static final DataNodeInternalRPCService INSTANCE = new DataNodeInternalRPCService(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/ConsensusWaitTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/ConsensusWaitTest.java new file mode 100644 index 000000000000..0f71a2699d60 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/ConsensusWaitTest.java @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.protocol.thrift.impl; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.service.ConsensusReadiness; +import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +public class ConsensusWaitTest { + + @BeforeClass + public static void setUp() { + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0); + } + + @Test + public void testRegionRequestBlocksUntilConsensusIsReady() throws Exception { + TestConsensusReadiness readiness = new TestConsensusReadiness(); + DataNodeRegionManager regionManager = Mockito.mock(DataNodeRegionManager.class); + DataNodeInternalRPCServiceImpl service = + new DataNodeInternalRPCServiceImpl(readiness, 500, regionManager); + TCreateSchemaRegionReq req = new TCreateSchemaRegionReq(); + TSStatus success = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + Mockito.when(regionManager.createSchemaRegion(null, null)).thenReturn(success); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> service.createSchemaRegion(req)); + Assert.assertTrue(readiness.awaitEntered.await(1, TimeUnit.SECONDS)); + Assert.assertFalse(result.isDone()); + Mockito.verifyZeroInteractions(regionManager); + + readiness.markReady(); + + Assert.assertSame(success, result.get(1, TimeUnit.SECONDS)); + Mockito.verify(regionManager).createSchemaRegion(null, null); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void testTimeoutRejectsWithoutChangingRegionMetadata() { + TestConsensusReadiness readiness = new TestConsensusReadiness(); + DataNodeRegionManager regionManager = Mockito.mock(DataNodeRegionManager.class); + DataNodeInternalRPCServiceImpl service = + new DataNodeInternalRPCServiceImpl(readiness, 20, regionManager); + + TSStatus status = service.createSchemaRegion(new TCreateSchemaRegionReq()); + + assertConsensusNotInitialized(status); + Assert.assertTrue(status.getMessage().contains("20")); + Mockito.verifyZeroInteractions(regionManager); + } + + @Test + public void testAllRegionTopologyRequestsAreGuarded() throws Exception { + ConsensusReadiness readiness = + new ConsensusReadiness() { + @Override + public boolean isAllConsensusStarted() { + return false; + } + + @Override + public boolean awaitAllConsensusStarted(long timeout, TimeUnit unit) { + return false; + } + }; + DataNodeRegionManager regionManager = Mockito.mock(DataNodeRegionManager.class); + DataNodeInternalRPCServiceImpl service = + new DataNodeInternalRPCServiceImpl(readiness, 1, regionManager); + + assertConsensusNotInitialized(service.createSchemaRegion(null)); + assertConsensusNotInitialized(service.createDataRegion(null)); + assertConsensusNotInitialized(service.deleteRegion(null)); + assertConsensusNotInitialized(service.changeRegionLeader(null).getStatus()); + assertConsensusNotInitialized(service.createNewRegionPeer(null)); + assertConsensusNotInitialized(service.addRegionPeer(null)); + assertConsensusNotInitialized(service.removeRegionPeer(null)); + assertConsensusNotInitialized(service.deleteOldRegionPeer(null)); + assertConsensusNotInitialized(service.resetPeerList(null)); + assertConsensusNotInitialized(service.notifyRegionMigration(null)); + Mockito.verifyZeroInteractions(regionManager); + } + + private static void assertConsensusNotInitialized(TSStatus status) { + Assert.assertEquals(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode(), status.getCode()); + Assert.assertTrue(status.isSetMessage()); + Assert.assertFalse(status.getMessage().isEmpty()); + } + + private static class TestConsensusReadiness implements ConsensusReadiness { + + private final CountDownLatch ready = new CountDownLatch(1); + private final CountDownLatch awaitEntered = new CountDownLatch(1); + + @Override + public boolean isAllConsensusStarted() { + return ready.getCount() == 0; + } + + @Override + public boolean awaitAllConsensusStarted(long timeout, TimeUnit unit) + throws InterruptedException { + awaitEntered.countDown(); + return ready.await(timeout, unit); + } + + private void markReady() { + ready.countDown(); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/ConsensusReadinessTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/ConsensusReadinessTest.java new file mode 100644 index 000000000000..f280d3fac932 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/ConsensusReadinessTest.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.service; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +public class ConsensusReadinessTest { + + @Test + public void testReadyOnlyAfterBothConsensusServicesStart() { + DataNode.ConsensusReadinessContext context = new DataNode.ConsensusReadinessContext(); + + context.markSchemaRegionConsensusStarted(); + Assert.assertFalse(context.isAllConsensusStarted()); + + context.markDataRegionConsensusStarted(); + Assert.assertTrue(context.isAllConsensusStarted()); + } + + @Test + public void testReadinessIsVisibleAcrossThreads() throws Exception { + DataNode.ConsensusReadinessContext context = new DataNode.ConsensusReadinessContext(); + context.markSchemaRegionConsensusStarted(); + CountDownLatch waiterStarted = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = + executor.submit( + () -> { + waiterStarted.countDown(); + return context.awaitAllConsensusStarted(1, TimeUnit.SECONDS); + }); + Assert.assertTrue(waiterStarted.await(1, TimeUnit.SECONDS)); + Assert.assertFalse(context.isAllConsensusStarted()); + Assert.assertFalse(result.isDone()); + + context.markDataRegionConsensusStarted(); + + Assert.assertTrue(result.get(1, TimeUnit.SECONDS)); + Assert.assertTrue(context.isAllConsensusStarted()); + } finally { + executor.shutdownNow(); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java index c8ac2c8880ea..7094c0c588bc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java @@ -77,6 +77,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.concurrent.TimeUnit; public class DataNodeInternalRPCServiceImplTest { @@ -133,7 +134,19 @@ public void setUp() throws Exception { .createLocalPeer( ConsensusGroupId.Factory.createFromTConsensusGroupId(regionReplicaSet.getRegionId()), genSchemaRegionPeerList(regionReplicaSet)); - dataNodeInternalRPCServiceImpl = new DataNodeInternalRPCServiceImpl(); + dataNodeInternalRPCServiceImpl = + new DataNodeInternalRPCServiceImpl( + new ConsensusReadiness() { + @Override + public boolean isAllConsensusStarted() { + return true; + } + + @Override + public boolean awaitAllConsensusStarted(long timeout, TimeUnit unit) { + return true; + } + }); } @After diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceTest.java new file mode 100644 index 000000000000..0b7a098c7d0e --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceTest.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.service; + +import org.junit.Assert; +import org.junit.Test; + +public class DataNodeInternalRPCServiceTest { + + @Test + public void testServiceReadsLatestDataNodeReadinessContext() { + DataNodeInternalRPCService service = DataNodeInternalRPCService.getInstance(); + try { + DataNode.ConsensusReadinessContext oldContext = new DataNode.ConsensusReadinessContext(); + service.setConsensusReadiness(oldContext); + ConsensusReadiness serviceContext = service.getConsensusReadiness(); + + DataNode.ConsensusReadinessContext newContext = new DataNode.ConsensusReadinessContext(); + service.setConsensusReadiness(newContext); + oldContext.markSchemaRegionConsensusStarted(); + oldContext.markDataRegionConsensusStarted(); + Assert.assertFalse(serviceContext.isAllConsensusStarted()); + + newContext.markSchemaRegionConsensusStarted(); + newContext.markDataRegionConsensusStarted(); + Assert.assertTrue(serviceContext.isAllConsensusStarted()); + } finally { + service.setConsensusReadiness(new DataNode.ConsensusReadinessContext()); + } + } +} diff --git a/pom.xml b/pom.xml index c073263bdba3..75bdfa5233b6 100644 --- a/pom.xml +++ b/pom.xml @@ -166,6 +166,7 @@ 1.9 1.5.6-3 1.1.4-260130-SNAPSHOT + en