This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 53d701c433f Fix concurrent access to IoTConsensus configuration
(#18290)
53d701c433f is described below
commit 53d701c433f3c137c66ad1763eba6fa7d33437b4
Author: Jiang Tian <[email protected]>
AuthorDate: Mon Jul 27 10:01:17 2026 +0800
Fix concurrent access to IoTConsensus configuration (#18290)
---
.../apache/iotdb/consensus/iot/IoTConsensus.java | 5 +-
.../consensus/iot/IoTConsensusServerImpl.java | 13 +-
.../consensus/iot/IoTConsensusServerImplTest.java | 162 +++++++++++++++++++++
3 files changed, 172 insertions(+), 8 deletions(-)
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index 1e247ad599e..3cab3498b36 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -80,7 +80,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
@@ -201,7 +200,7 @@ public class IoTConsensus implements IConsensus {
recvSnapshotDirs,
recvFolderStrategyType,
new Peer(consensusGroupId, thisNodeId, thisNode),
- new TreeSet<>(),
+ Collections.emptyList(),
registry.apply(consensusGroupId),
backgroundTaskService,
clientManager,
@@ -316,7 +315,7 @@ public class IoTConsensus implements IConsensus {
recvSnapshotDirs,
recvFolderStrategyType,
new Peer(groupId, thisNodeId, thisNode),
- new TreeSet<>(peers),
+ peers,
registry.apply(groupId),
backgroundTaskService,
clientManager,
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
index 2fd379c2d6c..1d55489f7cb 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
@@ -92,13 +92,14 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.PriorityQueue;
-import java.util.TreeSet;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
@@ -142,7 +143,7 @@ public class IoTConsensusServerImpl {
private final ConcurrentHashMap<String, ConcurrentHashMap<String, String>>
snapshotReceiveFolderMap = new ConcurrentHashMap<>();
- private final TreeSet<Peer> configuration;
+ private final Set<Peer> configuration = ConcurrentHashMap.newKeySet();
private final AtomicLong searchIndex;
private final LogDispatcher logDispatcher;
private IoTConsensusConfig config;
@@ -186,7 +187,7 @@ public class IoTConsensusServerImpl {
List<String> recvSnapshotDirs,
DirectoryStrategyType recvFolderStrategyType,
Peer thisNode,
- TreeSet<Peer> configuration,
+ Collection<Peer> configuration,
IStateMachine stateMachine,
ScheduledExecutorService backgroundTaskService,
IClientManager<TEndPoint, AsyncIoTConsensusServiceClient> clientManager,
@@ -209,7 +210,7 @@ public class IoTConsensusServerImpl {
this.stateMachine = stateMachine;
this.cacheQueueMap = new ConcurrentHashMap<>();
this.syncClientManager = syncClientManager;
- this.configuration = configuration;
+ this.configuration.addAll(configuration);
this.backgroundTaskService = backgroundTaskService;
this.config = config;
this.consensusGroupId = thisNode.getGroupId().toString();
@@ -1117,7 +1118,9 @@ public class IoTConsensusServerImpl {
}
public List<Peer> getConfiguration() {
- return new ArrayList<>(configuration);
+ List<Peer> result = new ArrayList<>(configuration);
+ Collections.sort(result);
+ return result;
}
public long getSearchIndex() {
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java
new file mode 100644
index 00000000000..24f8e8dc6bd
--- /dev/null
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.iot;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
+import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.consensus.config.IoTConsensusConfig;
+import org.apache.iotdb.consensus.iot.util.TestStateMachine;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.Spliterator;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class IoTConsensusServerImplTest {
+
+ private static final int WRITER_COUNT = 4;
+ private static final int READER_COUNT = 4;
+ private static final int PEERS_PER_WRITER = 1_000;
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ /**
+ * Verifies that configuration snapshots can be read while several writers
concurrently add and
+ * remove distinct peers. Every snapshot must remain duplicate-free and
sorted, and the final
+ * configuration must contain exactly the peers that were not removed.
+ */
+ @Test
+ public void testConfigurationConcurrentReadAndWrite() throws Exception {
+ final ScheduledExecutorService backgroundTaskService =
+ Executors.newSingleThreadScheduledExecutor();
+ final ExecutorService executorService =
+ Executors.newFixedThreadPool(WRITER_COUNT + READER_COUNT);
+ try {
+ final Peer thisNode = new Peer(new DataRegionId(1), 0, new
TEndPoint("127.0.0.1", 6667));
+ final IoTConsensusServerImpl server =
+ new IoTConsensusServerImpl(
+ temporaryFolder.getRoot().getAbsolutePath(),
+ null,
+ DirectoryStrategyType.SEQUENCE_STRATEGY,
+ thisNode,
+ Collections.singletonList(thisNode),
+ new TestStateMachine(),
+ backgroundTaskService,
+ null,
+ null,
+ IoTConsensusConfig.newBuilder().build());
+ final Set<Peer> configuration = getConfiguration(server);
+
assertTrue(configuration.spliterator().hasCharacteristics(Spliterator.CONCURRENT));
+
+ final CountDownLatch startLatch = new CountDownLatch(1);
+ final CountDownLatch writersDoneLatch = new CountDownLatch(WRITER_COUNT);
+ final Set<Peer> expectedConfiguration = new HashSet<>();
+ expectedConfiguration.add(thisNode);
+
+ final Future<?>[] writerFutures = new Future<?>[WRITER_COUNT];
+ for (int writerIndex = 0; writerIndex < WRITER_COUNT; writerIndex++) {
+ final int currentWriterIndex = writerIndex;
+ writerFutures[writerIndex] =
+ executorService.submit(
+ () -> {
+ startLatch.await();
+ try {
+ for (int peerIndex = 0; peerIndex < PEERS_PER_WRITER;
peerIndex++) {
+ final Peer peer = createPeer(currentWriterIndex,
peerIndex);
+ configuration.add(peer);
+ if ((peerIndex & 1) == 0) {
+ configuration.remove(peer);
+ }
+ }
+ } finally {
+ writersDoneLatch.countDown();
+ }
+ return null;
+ });
+ for (int peerIndex = 1; peerIndex < PEERS_PER_WRITER; peerIndex += 2) {
+ expectedConfiguration.add(createPeer(writerIndex, peerIndex));
+ }
+ }
+
+ final Future<?>[] readerFutures = new Future<?>[READER_COUNT];
+ for (int readerIndex = 0; readerIndex < READER_COUNT; readerIndex++) {
+ readerFutures[readerIndex] =
+ executorService.submit(
+ () -> {
+ startLatch.await();
+ do {
+ assertValidSnapshot(server.getConfiguration());
+ } while (writersDoneLatch.getCount() > 0);
+ assertValidSnapshot(server.getConfiguration());
+ return null;
+ });
+ }
+
+ startLatch.countDown();
+ for (Future<?> writerFuture : writerFutures) {
+ writerFuture.get(30, TimeUnit.SECONDS);
+ }
+ for (Future<?> readerFuture : readerFutures) {
+ readerFuture.get(30, TimeUnit.SECONDS);
+ }
+
+ assertEquals(expectedConfiguration, new
HashSet<>(server.getConfiguration()));
+ } finally {
+ executorService.shutdownNow();
+ backgroundTaskService.shutdownNow();
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Set<Peer> getConfiguration(IoTConsensusServerImpl server)
throws Exception {
+ final Field configurationField =
IoTConsensusServerImpl.class.getDeclaredField("configuration");
+ configurationField.setAccessible(true);
+ return (Set<Peer>) configurationField.get(server);
+ }
+
+ private static Peer createPeer(int writerIndex, int peerIndex) {
+ final int nodeId = writerIndex * PEERS_PER_WRITER + peerIndex + 1;
+ return new Peer(new DataRegionId(1), nodeId, new TEndPoint("127.0.0.1",
10_000 + nodeId));
+ }
+
+ private static void assertValidSnapshot(List<Peer> snapshot) {
+ assertEquals(snapshot.size(), new HashSet<>(snapshot).size());
+ for (int index = 1; index < snapshot.size(); index++) {
+ assertTrue(snapshot.get(index - 1).compareTo(snapshot.get(index)) < 0);
+ }
+ }
+}
