You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/04/23 11:06:53 UTC
(iotdb) branch check_consensus_before_answering_region_request created (now 1f5671e3cf7)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a change to branch check_consensus_before_answering_region_request
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 1f5671e3cf7 Add ut
This branch includes the following new commits:
new f8f5a7f5d28 Add Await and DataNodeContext
new 6338581e6be Wait for consensus start before answering region requests
new 1f5671e3cf7 Add ut
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/03: Add Await and DataNodeContext
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch check_consensus_before_answering_region_request
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f8f5a7f5d286d09a97bde91fa14a71067c2a699c
Author: Tian Jiang <jt...@163.com>
AuthorDate: Thu Apr 23 18:02:49 2026 +0800
Add Await and DataNodeContext
---
.../impl/DataNodeInternalRPCServiceImpl.java | 6 +-
.../java/org/apache/iotdb/db/service/DataNode.java | 15 +-
.../db/service/DataNodeInternalRPCService.java | 12 +-
.../DataNodeInternalRPCServiceImplTest.java | 8 +-
.../org/apache/iotdb/commons/concurrent/Await.java | 53 ++++++
.../commons/concurrent/AwaitTimeoutException.java | 31 ++++
.../iotdb/commons/concurrent/ConditionAwaiter.java | 154 +++++++++++++++++
.../java/org/apache/iotdb/commons/AwaitTest.java | 190 +++++++++++++++++++++
8 files changed, 461 insertions(+), 8 deletions(-)
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 9fc5fcb436c..28422bbd837 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
@@ -193,6 +193,7 @@ import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUpdateType;
import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUtil;
import org.apache.iotdb.db.service.DataNode;
+import org.apache.iotdb.db.service.DataNode.DataNodeContext;
import org.apache.iotdb.db.service.RegionMigrateService;
import org.apache.iotdb.db.service.externalservice.ExternalServiceManagementService;
import org.apache.iotdb.db.service.metrics.FileMetrics;
@@ -416,6 +417,8 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
+ private final DataNodeContext dataNodeContext;
+
private final ExecutorService schemaExecutor =
new WrappedThreadPoolExecutor(
0,
@@ -430,10 +433,11 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
private static final String SYSTEM = "system";
- public DataNodeInternalRPCServiceImpl() {
+ public DataNodeInternalRPCServiceImpl(DataNodeContext dataNodeContext) {
super();
partitionFetcher = ClusterPartitionFetcher.getInstance();
schemaFetcher = ClusterSchemaFetcher.getInstance();
+ this.dataNodeContext = dataNodeContext;
}
@Override
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 7004d09872d..568f2c266e2 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
@@ -183,14 +183,16 @@ 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 static Thread watcherThread;
+ private DataNodeContext context;
public DataNode() {
super("DataNode");
// We do not init anything here, so that we can re-initialize the instance in IT.
DataNodeHolder.INSTANCE = this;
+ context = new DataNodeContext();
}
public static void reinitializeStatics() {
@@ -934,7 +936,8 @@ public class DataNode extends ServerCommandLine implements DataNodeMBean {
protected void registerInternalRPCService() throws StartupException {
// Start InternalRPCService to indicate that the current DataNode can accept cluster scheduling
- registerManager.register(DataNodeInternalRPCService.getInstance());
+ DataNodeInternalRPCService instance = DataNodeInternalRPCService.getInstance();
+ registerManager.register();
}
// make it easier for users to extend ClientRPCServiceImpl to export more rpc services
@@ -1373,4 +1376,10 @@ public class DataNode extends ServerCommandLine implements DataNodeMBean {
// Empty constructor
}
}
+
+ public class DataNodeContext {
+ public boolean isAllConsensusStarted() {
+ return dataRegionConsensusStarted && schemaRegionConsensusStarted;
+ }
+ }
}
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 5de1041a9a0..f3bf8e507c2 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
@@ -31,6 +31,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.protocol.thrift.handler.InternalServiceThriftHandler;
import org.apache.iotdb.db.protocol.thrift.impl.DataNodeInternalRPCServiceImpl;
+import org.apache.iotdb.db.service.DataNode.DataNodeContext;
import org.apache.iotdb.db.service.metrics.DataNodeInternalRPCServiceMetrics;
import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService.Processor;
import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory;
@@ -44,6 +45,7 @@ public class DataNodeInternalRPCService extends ThriftService
private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
private final AtomicReference<DataNodeInternalRPCServiceImpl> impl = new AtomicReference<>();
+ private DataNodeContext dataNodeContext;
private DataNodeInternalRPCService() {}
@@ -54,9 +56,9 @@ public class DataNodeInternalRPCService extends ThriftService
@Override
public void initTProcessor() {
- impl.compareAndSet(null, new DataNodeInternalRPCServiceImpl());
+ DataNodeInternalRPCServiceImpl service = getImpl();
initSyncedServiceImpl(null);
- processor = new Processor<>(impl.get());
+ processor = new Processor<>(service);
}
@Override
@@ -109,7 +111,7 @@ public class DataNodeInternalRPCService extends ThriftService
}
public DataNodeInternalRPCServiceImpl getImpl() {
- impl.compareAndSet(null, new DataNodeInternalRPCServiceImpl());
+ impl.compareAndSet(null, new DataNodeInternalRPCServiceImpl(dataNodeContext));
return impl.get();
}
@@ -122,4 +124,8 @@ public class DataNodeInternalRPCService extends ThriftService
public static DataNodeInternalRPCService getInstance() {
return DataNodeInternalRPCServiceHolder.INSTANCE;
}
+
+ public void setDataNodeContext(DataNodeContext dataNodeContext) {
+ this.dataNodeContext = dataNodeContext;
+ }
}
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 22a2d6cdc46..f601c9bd1fc 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
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.service;
+import static org.mockito.Mockito.when;
+
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
@@ -50,6 +52,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.Cre
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateMultiTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateTimeSeriesNode;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
+import org.apache.iotdb.db.service.DataNode.DataNodeContext;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -79,6 +82,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import org.mockito.Mockito;
public class DataNodeInternalRPCServiceImplTest {
@@ -134,7 +138,9 @@ public class DataNodeInternalRPCServiceImplTest {
.createLocalPeer(
ConsensusGroupId.Factory.createFromTConsensusGroupId(regionReplicaSet.getRegionId()),
genSchemaRegionPeerList(regionReplicaSet));
- dataNodeInternalRPCServiceImpl = new DataNodeInternalRPCServiceImpl();
+ DataNodeContext context = Mockito.mock(DataNodeContext.class);
+ when(context.isAllConsensusStarted()).thenReturn(true);
+ dataNodeInternalRPCServiceImpl = new DataNodeInternalRPCServiceImpl(context);
}
@After
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/Await.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/Await.java
new file mode 100644
index 00000000000..f08ccd17495
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/Await.java
@@ -0,0 +1,53 @@
+/*
+ * 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.commons.concurrent;
+
+/**
+ * Lightweight polling utility for production code. Provides a fluent API similar to Awaitility for
+ * waiting until a condition becomes true.
+ *
+ * <pre>{@code
+ * // Wait with timeout
+ * Await.await()
+ * .atMost(5, TimeUnit.SECONDS)
+ * .pollInterval(100, TimeUnit.MILLISECONDS)
+ * .until(() -> isReady());
+ *
+ * // Wait forever (use with caution)
+ * Await.await()
+ * .forever()
+ * .pollInterval(1, TimeUnit.SECONDS)
+ * .until(() -> isReady());
+ *
+ * // Ignore exceptions during polling
+ * Await.await()
+ * .atMost(30, TimeUnit.SECONDS)
+ * .ignoreExceptions()
+ * .until(() -> tryConnect());
+ * }</pre>
+ */
+public final class Await {
+
+ private Await() {}
+
+ public static ConditionAwaiter await() {
+ return new ConditionAwaiter();
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/AwaitTimeoutException.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/AwaitTimeoutException.java
new file mode 100644
index 00000000000..b0d5c98bfe7
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/AwaitTimeoutException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.commons.concurrent;
+
+public class AwaitTimeoutException extends RuntimeException {
+
+ public AwaitTimeoutException(String message) {
+ super(message);
+ }
+
+ public AwaitTimeoutException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ConditionAwaiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ConditionAwaiter.java
new file mode 100644
index 00000000000..f88db57f612
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ConditionAwaiter.java
@@ -0,0 +1,154 @@
+/*
+ * 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.commons.concurrent;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+public class ConditionAwaiter {
+
+ private static final long DEFAULT_POLL_INTERVAL_MS = 100;
+ private static final long DEFAULT_TIMEOUT_MS = 10_000;
+
+ private long timeoutMs = DEFAULT_TIMEOUT_MS;
+ private long pollIntervalMs = DEFAULT_POLL_INTERVAL_MS;
+ private long pollDelayMs = 0;
+ private boolean ignoreAllExceptions = false;
+ private boolean forever = false;
+ private final List<Class<? extends Exception>> ignoredExceptions = new ArrayList<>();
+
+ ConditionAwaiter() {}
+
+ public ConditionAwaiter atMost(long time, TimeUnit unit) {
+ this.timeoutMs = unit.toMillis(time);
+ return this;
+ }
+
+ public ConditionAwaiter pollInterval(long time, TimeUnit unit) {
+ this.pollIntervalMs = unit.toMillis(time);
+ return this;
+ }
+
+ public ConditionAwaiter pollDelay(long time, TimeUnit unit) {
+ this.pollDelayMs = unit.toMillis(time);
+ return this;
+ }
+
+ public ConditionAwaiter ignoreExceptions() {
+ this.ignoreAllExceptions = true;
+ return this;
+ }
+
+ public ConditionAwaiter ignoreException(Class<? extends Exception> exceptionType) {
+ this.ignoredExceptions.add(exceptionType);
+ return this;
+ }
+
+ public ConditionAwaiter forever() {
+ this.forever = true;
+ return this;
+ }
+
+ public void until(Callable<Boolean> conditionEvaluator) {
+ long startTime = System.currentTimeMillis();
+
+ if (pollDelayMs > 0) {
+ sleep(pollDelayMs);
+ }
+
+ Exception lastException = null;
+ while (true) {
+ try {
+ Boolean result = conditionEvaluator.call();
+ if (Boolean.TRUE.equals(result)) {
+ return;
+ }
+ lastException = null;
+ } catch (Exception e) {
+ if (shouldIgnore(e)) {
+ lastException = e;
+ } else if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ throw new AwaitTimeoutException("Interrupted while awaiting condition", e);
+ } else {
+ throw new AwaitTimeoutException("Exception while evaluating condition", e);
+ }
+ }
+
+ if (!forever && System.currentTimeMillis() - startTime >= timeoutMs) {
+ String message = String.format("Condition was not met within %d ms", timeoutMs);
+ if (lastException != null) {
+ throw new AwaitTimeoutException(message, lastException);
+ }
+ throw new AwaitTimeoutException(message);
+ }
+
+ sleep(pollIntervalMs);
+ }
+ }
+
+ public void untilAsserted(Runnable assertion) {
+ final AssertionErrorHolder holder = new AssertionErrorHolder();
+ try {
+ until(
+ () -> {
+ try {
+ assertion.run();
+ return true;
+ } catch (AssertionError e) {
+ holder.error = e;
+ return false;
+ }
+ });
+ } catch (AwaitTimeoutException e) {
+ if (holder.error != null) {
+ throw new AwaitTimeoutException(e.getMessage(), holder.error);
+ }
+ throw e;
+ }
+ }
+
+ private static final class AssertionErrorHolder {
+ AssertionError error;
+ }
+
+ private boolean shouldIgnore(Exception e) {
+ if (ignoreAllExceptions) {
+ return true;
+ }
+ for (Class<? extends Exception> ignoredType : ignoredExceptions) {
+ if (ignoredType.isInstance(e)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void sleep(long millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AwaitTimeoutException("Interrupted while awaiting condition", e);
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/AwaitTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/AwaitTest.java
new file mode 100644
index 00000000000..ae094c1a806
--- /dev/null
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/AwaitTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.commons;
+
+import org.apache.iotdb.commons.concurrent.Await;
+import org.apache.iotdb.commons.concurrent.AwaitTimeoutException;
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class AwaitTest {
+
+ @Test
+ public void testConditionAlreadyTrue() {
+ Await.await().atMost(1, TimeUnit.SECONDS).until(() -> true);
+ }
+
+ @Test
+ public void testConditionBecomesTrue() {
+ AtomicBoolean flag = new AtomicBoolean(false);
+ new Thread(
+ () -> {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ flag.set(true);
+ })
+ .start();
+
+ Await.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .pollInterval(50, TimeUnit.MILLISECONDS)
+ .until(flag::get);
+
+ assertTrue(flag.get());
+ }
+
+ @Test(expected = AwaitTimeoutException.class)
+ public void testTimeout() {
+ Await.await()
+ .atMost(300, TimeUnit.MILLISECONDS)
+ .pollInterval(50, TimeUnit.MILLISECONDS)
+ .until(() -> false);
+ }
+
+ @Test
+ public void testPollDelay() {
+ long start = System.currentTimeMillis();
+
+ Await.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .pollDelay(200, TimeUnit.MILLISECONDS)
+ .until(() -> true);
+
+ long elapsed = System.currentTimeMillis() - start;
+ assertTrue("Expected at least 200ms delay, got " + elapsed, elapsed >= 180);
+ }
+
+ @Test
+ public void testIgnoreAllExceptions() {
+ AtomicInteger counter = new AtomicInteger(0);
+
+ Await.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .pollInterval(50, TimeUnit.MILLISECONDS)
+ .ignoreExceptions()
+ .until(
+ () -> {
+ int val = counter.incrementAndGet();
+ if (val < 3) {
+ throw new RuntimeException("not ready yet");
+ }
+ return true;
+ });
+
+ assertTrue(counter.get() >= 3);
+ }
+
+ @Test
+ public void testIgnoreSpecificException() {
+ AtomicInteger counter = new AtomicInteger(0);
+
+ Await.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .pollInterval(50, TimeUnit.MILLISECONDS)
+ .ignoreException(IllegalStateException.class)
+ .until(
+ () -> {
+ int val = counter.incrementAndGet();
+ if (val < 3) {
+ throw new IllegalStateException("not ready");
+ }
+ return true;
+ });
+
+ assertTrue(counter.get() >= 3);
+ }
+
+ @Test
+ public void testNonIgnoredExceptionPropagates() {
+ try {
+ Await.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .ignoreException(IllegalStateException.class)
+ .until(
+ () -> {
+ throw new IllegalArgumentException("unexpected");
+ });
+ fail("Should have thrown");
+ } catch (AwaitTimeoutException e) {
+ assertTrue(e.getCause() instanceof IllegalArgumentException);
+ }
+ }
+
+ @Test
+ public void testUntilAsserted() {
+ AtomicInteger value = new AtomicInteger(0);
+ new Thread(
+ () -> {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ value.set(42);
+ })
+ .start();
+
+ Await.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .pollInterval(50, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> assertEquals(42, value.get()));
+ }
+
+ @Test
+ public void testForever() {
+ AtomicInteger counter = new AtomicInteger(0);
+
+ Await.await()
+ .forever()
+ .pollInterval(10, TimeUnit.MILLISECONDS)
+ .until(() -> counter.incrementAndGet() >= 5);
+
+ assertTrue(counter.get() >= 5);
+ }
+
+ @Test
+ public void testTimeoutMessageIncludesLastException() {
+ try {
+ Await.await()
+ .atMost(200, TimeUnit.MILLISECONDS)
+ .pollInterval(50, TimeUnit.MILLISECONDS)
+ .ignoreExceptions()
+ .until(
+ () -> {
+ throw new RuntimeException("still failing");
+ });
+ fail("Should have thrown");
+ } catch (AwaitTimeoutException e) {
+ assertTrue(e.getCause() instanceof RuntimeException);
+ assertEquals("still failing", e.getCause().getMessage());
+ }
+ }
+}
(iotdb) 03/03: Add ut
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch check_consensus_before_answering_region_request
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1f5671e3cf72346b053050fc1558fc2d689803c2
Author: Tian Jiang <jt...@163.com>
AuthorDate: Thu Apr 23 19:04:39 2026 +0800
Add ut
---
.../impl/DataNodeInternalRPCServiceImpl.java | 14 ++-
.../java/org/apache/iotdb/db/service/DataNode.java | 3 +-
.../db/protocol/thrift/impl/ConsensusWaitTest.java | 131 +++++++++++++++++++++
3 files changed, 143 insertions(+), 5 deletions(-)
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 bd420fab06a..5075212bf41 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
@@ -442,21 +442,25 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
this.dataNodeContext = dataNodeContext;
}
+ private long consensusWaitTimeoutSeconds = 30;
+
private TSStatus waitForConsensusStarted() {
if (dataNodeContext.isAllConsensusStarted()) {
return null;
}
try {
Await.await()
- .atMost(30, TimeUnit.SECONDS)
+ .atMost(consensusWaitTimeoutSeconds, TimeUnit.SECONDS)
.pollInterval(100, TimeUnit.MILLISECONDS)
.until(dataNodeContext::isAllConsensusStarted);
return null;
} catch (AwaitTimeoutException e) {
- LOGGER.warn("Consensus has not been started after 30 seconds, rejecting region request");
+ LOGGER.warn(
+ "Consensus has not been started after {} seconds, rejecting region request",
+ consensusWaitTimeoutSeconds);
return RpcUtils.getStatus(
TSStatusCode.CONSENSUS_NOT_INITIALIZED,
- "Consensus has not been started after 30 seconds");
+ "Consensus has not been started after " + consensusWaitTimeoutSeconds + " seconds");
}
}
@@ -3505,4 +3509,8 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
return result;
}
+
+ public void setConsensusWaitTimeoutSeconds(long consensusWaitTimeoutSeconds) {
+ this.consensusWaitTimeoutSeconds = consensusWaitTimeoutSeconds;
+ }
}
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 568f2c266e2..00f113cc70f 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
@@ -936,8 +936,7 @@ public class DataNode extends ServerCommandLine implements DataNodeMBean {
protected void registerInternalRPCService() throws StartupException {
// Start InternalRPCService to indicate that the current DataNode can accept cluster scheduling
- DataNodeInternalRPCService instance = DataNodeInternalRPCService.getInstance();
- registerManager.register();
+ registerManager.register(DataNodeInternalRPCService.getInstance());
}
// make it easier for users to extend ClientRPCServiceImpl to export more rpc services
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 00000000000..65fd2b47858
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/ConsensusWaitTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.service.DataNode.DataNodeContext;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeResp;
+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.Collections;
+
+import static org.mockito.Mockito.when;
+
+public class ConsensusWaitTest {
+
+ @BeforeClass
+ public static void setUp() {
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0);
+ }
+
+ private DataNodeInternalRPCServiceImpl createServiceWithConsensusState(boolean started) {
+ DataNodeContext context = Mockito.mock(DataNodeContext.class);
+ when(context.isAllConsensusStarted()).thenReturn(started);
+ DataNodeInternalRPCServiceImpl service = new DataNodeInternalRPCServiceImpl(context);
+ service.setConsensusWaitTimeoutSeconds(1);
+ return service;
+ }
+
+ private TCreateSchemaRegionReq createSchemaRegionReq() {
+ TCreateSchemaRegionReq req = new TCreateSchemaRegionReq();
+ req.setStorageGroup("root.test");
+ TRegionReplicaSet replicaSet = new TRegionReplicaSet();
+ replicaSet.setRegionId(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 0));
+ TDataNodeLocation location = new TDataNodeLocation();
+ location.setDataNodeId(0);
+ location.setClientRpcEndPoint(new TEndPoint("0.0.0.0", 6667));
+ location.setInternalEndPoint(new TEndPoint("0.0.0.0", 10730));
+ location.setMPPDataExchangeEndPoint(new TEndPoint("0.0.0.0", 10740));
+ location.setDataRegionConsensusEndPoint(new TEndPoint("0.0.0.0", 10760));
+ location.setSchemaRegionConsensusEndPoint(new TEndPoint("0.0.0.0", 10750));
+ replicaSet.setDataNodeLocations(Collections.singletonList(location));
+ req.setRegionReplicaSet(replicaSet);
+ return req;
+ }
+
+ private TCreateDataRegionReq createDataRegionReq() {
+ TCreateDataRegionReq req = new TCreateDataRegionReq();
+ req.setStorageGroup("root.test");
+ TRegionReplicaSet replicaSet = new TRegionReplicaSet();
+ replicaSet.setRegionId(new TConsensusGroupId(TConsensusGroupType.DataRegion, 0));
+ TDataNodeLocation location = new TDataNodeLocation();
+ location.setDataNodeId(0);
+ location.setClientRpcEndPoint(new TEndPoint("0.0.0.0", 6667));
+ location.setInternalEndPoint(new TEndPoint("0.0.0.0", 10730));
+ location.setMPPDataExchangeEndPoint(new TEndPoint("0.0.0.0", 10740));
+ location.setDataRegionConsensusEndPoint(new TEndPoint("0.0.0.0", 10760));
+ location.setSchemaRegionConsensusEndPoint(new TEndPoint("0.0.0.0", 10750));
+ replicaSet.setDataNodeLocations(Collections.singletonList(location));
+ req.setRegionReplicaSet(replicaSet);
+ return req;
+ }
+
+ @Test
+ public void testCreateSchemaRegionRejectsWhenConsensusNotStarted() {
+ DataNodeInternalRPCServiceImpl service = createServiceWithConsensusState(false);
+ TSStatus status = service.createSchemaRegion(createSchemaRegionReq());
+ Assert.assertEquals(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode(), status.getCode());
+ }
+
+ @Test
+ public void testCreateDataRegionRejectsWhenConsensusNotStarted() {
+ DataNodeInternalRPCServiceImpl service = createServiceWithConsensusState(false);
+ TSStatus status = service.createDataRegion(createDataRegionReq());
+ Assert.assertEquals(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode(), status.getCode());
+ }
+
+ @Test
+ public void testDeleteRegionRejectsWhenConsensusNotStarted() {
+ DataNodeInternalRPCServiceImpl service = createServiceWithConsensusState(false);
+ TConsensusGroupId groupId = new TConsensusGroupId(TConsensusGroupType.DataRegion, 0);
+ TSStatus status = service.deleteRegion(groupId);
+ Assert.assertEquals(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode(), status.getCode());
+ }
+
+ @Test
+ public void testChangeRegionLeaderRejectsWhenConsensusNotStarted() {
+ DataNodeInternalRPCServiceImpl service = createServiceWithConsensusState(false);
+ TRegionLeaderChangeReq req = new TRegionLeaderChangeReq();
+ req.setRegionId(new TConsensusGroupId(TConsensusGroupType.DataRegion, 0));
+ TDataNodeLocation newLeader = new TDataNodeLocation();
+ newLeader.setDataNodeId(0);
+ newLeader.setInternalEndPoint(new TEndPoint("0.0.0.0", 10730));
+ newLeader.setDataRegionConsensusEndPoint(new TEndPoint("0.0.0.0", 10760));
+ newLeader.setSchemaRegionConsensusEndPoint(new TEndPoint("0.0.0.0", 10750));
+ req.setNewLeaderNode(newLeader);
+ TRegionLeaderChangeResp resp = service.changeRegionLeader(req);
+ Assert.assertEquals(
+ TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode(), resp.getStatus().getCode());
+ }
+}
(iotdb) 02/03: Wait for consensus start before answering region requests
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch check_consensus_before_answering_region_request
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6338581e6be087620ef54a1c14cf0fa17c8fb5a6
Author: Tian Jiang <jt...@163.com>
AuthorDate: Thu Apr 23 18:21:25 2026 +0800
Wait for consensus start before answering region requests
---
.../impl/DataNodeInternalRPCServiceImpl.java | 62 ++++++++++++++++++++++
.../DataNodeInternalRPCServiceImplTest.java | 6 +--
2 files changed, 65 insertions(+), 3 deletions(-)
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 28422bbd837..bd420fab06a 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
@@ -49,6 +49,8 @@ import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
import org.apache.iotdb.commons.client.request.AsyncRequestContext;
import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.concurrent.Await;
+import org.apache.iotdb.commons.concurrent.AwaitTimeoutException;
import org.apache.iotdb.commons.concurrent.IoTThreadFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor;
@@ -440,6 +442,24 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
this.dataNodeContext = dataNodeContext;
}
+ private TSStatus waitForConsensusStarted() {
+ if (dataNodeContext.isAllConsensusStarted()) {
+ return null;
+ }
+ try {
+ Await.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .until(dataNodeContext::isAllConsensusStarted);
+ return null;
+ } catch (AwaitTimeoutException e) {
+ LOGGER.warn("Consensus has not been started after 30 seconds, rejecting region request");
+ return RpcUtils.getStatus(
+ TSStatusCode.CONSENSUS_NOT_INITIALIZED,
+ "Consensus has not been started after 30 seconds");
+ }
+ }
+
@Override
public TSendFragmentInstanceResp sendFragmentInstance(final TSendFragmentInstanceReq req) {
LOGGER.debug("receive FragmentInstance to group[{}]", req.getConsensusGroupId());
@@ -628,11 +648,19 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TSStatus createSchemaRegion(final 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());
}
@@ -2620,6 +2648,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TSStatus deleteRegion(TConsensusGroupId tconsensusGroupId) {
+ TSStatus consensusStatus = waitForConsensusStarted();
+ if (consensusStatus != null) {
+ return consensusStatus;
+ }
ConsensusGroupId consensusGroupId =
ConsensusGroupId.Factory.createFromTConsensusGroupId(tconsensusGroupId);
if (consensusGroupId instanceof DataRegionId) {
@@ -2648,6 +2680,12 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
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);
@@ -2717,6 +2755,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TSStatus createNewRegionPeer(TCreatePeerReq req) {
+ TSStatus consensusStatus = waitForConsensusStarted();
+ if (consensusStatus != null) {
+ return consensusStatus;
+ }
ConsensusGroupId regionId =
ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getRegionId());
List<Peer> peers =
@@ -2737,6 +2779,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@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);
@@ -2755,6 +2801,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@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);
@@ -2773,6 +2823,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@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);
@@ -2792,6 +2846,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
// 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);
}
@@ -2802,6 +2860,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@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/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java
index f601c9bd1fc..63dda4537f9 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
@@ -19,8 +19,6 @@
package org.apache.iotdb.db.service;
-import static org.mockito.Mockito.when;
-
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
@@ -71,6 +69,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
@@ -82,7 +81,8 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
-import org.mockito.Mockito;
+
+import static org.mockito.Mockito.when;
public class DataNodeInternalRPCServiceImplTest {