This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 26e1425a903 [To dev/1.3] Fix IoTConsensus batch accumulation and
config reload (#18501) (#18507)
26e1425a903 is described below
commit 26e1425a9034b66d5835cbe2f6190ab68c4d5163
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 31 18:08:04 2026 +0800
[To dev/1.3] Fix IoTConsensus batch accumulation and config reload (#18501)
(#18507)
* Fix IoTConsensus batch accumulation and config reload (#18501)
Backport #18501 to origin/dev/1.3.
* Fix IoTConsensus batch accumulation latency
---
.../apache/iotdb/consensus/iot/IoTConsensus.java | 7 +-
.../consensus/iot/IoTConsensusServerImpl.java | 3 +-
.../logdispatcher/IoTConsensusMemoryManager.java | 4 +-
.../consensus/iot/logdispatcher/LogDispatcher.java | 53 +++-
.../consensus/iot/logdispatcher/SyncStatus.java | 7 +-
.../IoTConsensusMemoryManagerTest.java | 14 +
.../iot/logdispatcher/LogDispatcherTest.java | 286 +++++++++++++++++++++
7 files changed, 363 insertions(+), 11 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 d15d6e365a7..11707fe5634 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
@@ -97,7 +97,7 @@ public class IoTConsensus implements IConsensus {
new ConcurrentHashMap<>();
private final IoTConsensusRPCService service;
private final RegisterManager registerManager = new RegisterManager();
- private IoTConsensusConfig config;
+ private volatile IoTConsensusConfig config;
private final IClientManager<TEndPoint, AsyncIoTConsensusServiceClient>
clientManager;
private final IClientManager<TEndPoint, SyncIoTConsensusServiceClient>
syncClientManager;
private final ScheduledExecutorService backgroundTaskService;
@@ -472,6 +472,11 @@ public class IoTConsensus implements IConsensus {
public void reloadConsensusConfig(ConsensusConfig consensusConfig) {
config = consensusConfig.getIotConsensusConfig();
+ IoTConsensusMemoryManager.getInstance()
+ .init(
+ config.getReplication().getAllocateMemoryForConsensus(),
+ config.getReplication().getAllocateMemoryForQueue());
+
for (IoTConsensusServerImpl impl : stateMachineMap.values()) {
impl.reloadConsensusConfig(config);
}
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 3002b018e3e..7033ddf36d9 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
@@ -116,7 +116,7 @@ public class IoTConsensusServerImpl {
private final TreeSet<Peer> configuration;
private final AtomicLong searchIndex;
private final LogDispatcher logDispatcher;
- private IoTConsensusConfig config;
+ private volatile IoTConsensusConfig config;
private final ConsensusReqReader consensusReqReader;
private volatile boolean active;
private String newSnapshotDirName;
@@ -911,6 +911,7 @@ public class IoTConsensusServerImpl {
/** This method is used for hot reload of IoTConsensusConfig. */
public void reloadConsensusConfig(IoTConsensusConfig config) {
this.config = config;
+ logDispatcher.reloadConfig(config);
}
/**
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java
index 22e5484f5a2..d8adec09a7b 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java
@@ -33,8 +33,8 @@ public class IoTConsensusMemoryManager {
private final AtomicLong memorySizeInByte = new AtomicLong(0);
private final AtomicLong queueMemorySizeInByte = new AtomicLong(0);
private final AtomicLong syncMemorySizeInByte = new AtomicLong(0);
- private Long maxMemorySizeInByte = Runtime.getRuntime().maxMemory() / 10;
- private Long maxMemorySizeForQueueInByte = Runtime.getRuntime().maxMemory()
/ 100 * 6;
+ private volatile long maxMemorySizeInByte = Runtime.getRuntime().maxMemory()
/ 10;
+ private volatile long maxMemorySizeForQueueInByte =
Runtime.getRuntime().maxMemory() / 100 * 6;
private IoTConsensusMemoryManager() {
MetricService.getInstance().addMetricSet(new
IoTConsensusMemoryManagerMetrics(this));
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
index 374691bf38b..6b6edee6683 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
@@ -179,6 +179,10 @@ public class LogDispatcher {
}
}
+ public synchronized void reloadConfig(IoTConsensusConfig config) {
+ threads.forEach(thread -> thread.reloadConfig(config));
+ }
+
public void offer(IndexedConsensusRequest request) {
// we don't need to serialize and offer request when replicaNum is 1.
if (!threads.isEmpty()) {
@@ -215,7 +219,7 @@ public class LogDispatcher {
private static final long PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC = 10;
private static final long START_INDEX = 1;
- private final IoTConsensusConfig config;
+ private volatile IoTConsensusConfig config;
private final Peer peer;
private final IndexController controller;
// A sliding window class that manages asynchronous pendingBatches
@@ -273,6 +277,11 @@ public class LogDispatcher {
return config;
}
+ private void reloadConfig(IoTConsensusConfig config) {
+ this.config = config;
+ syncStatus.reloadConfig(config);
+ }
+
public int getPendingEntriesSize() {
return pendingEntries.size();
}
@@ -358,11 +367,16 @@ public class LogDispatcher {
IndexedConsensusRequest request =
pendingEntries.poll(PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC,
TimeUnit.SECONDS);
if (request != null) {
+ final IoTConsensusConfig currentConfig = config;
+ final boolean shouldWaitForBatchAccumulation =
+ pendingEntries.size()
+ <=
currentConfig.getReplication().getMaxLogEntriesNumPerBatch()
+ && bufferedEntries.isEmpty();
bufferedEntries.add(request);
// If write pressure is low, we simply sleep a little to reduce
the number of RPC
- if (pendingEntries.size() <=
config.getReplication().getMaxLogEntriesNumPerBatch()
- && bufferedEntries.isEmpty()) {
-
Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
+ if (shouldWaitForBatchAccumulation) {
+ waitForBatchAccumulation(
+
currentConfig.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
}
}
// Immediately check for interrupts after poll and sleep
@@ -392,6 +406,32 @@ public class LogDispatcher {
logger.info("{}: Dispatcher for {} exits", impl.getThisNode(), peer);
}
+ void waitForBatchAccumulation(long waitingTimeInMs) throws
InterruptedException {
+ if (waitingTimeInMs <= 0) {
+ return;
+ }
+
+ final long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(waitingTimeInMs);
+ final int maxLogEntriesNumPerBatch =
config.getReplication().getMaxLogEntriesNumPerBatch();
+
+ // Keep collecting while the batch is below its entry limit. A plain
sleep makes the
+ // dispatcher wait for the full accumulation interval even when the
batch becomes full
+ // immediately, which unnecessarily throttles IoTConsensus under
sustained write load.
+ while (bufferedEntries.size() < maxLogEntriesNumPerBatch) {
+ final long remainingNanos = deadlineNanos - System.nanoTime();
+ if (remainingNanos <= 0) {
+ return;
+ }
+
+ final IndexedConsensusRequest request =
+ pendingEntries.poll(remainingNanos, TimeUnit.NANOSECONDS);
+ if (request == null) {
+ return;
+ }
+ bufferedEntries.add(request);
+ }
+ }
+
public void updateSafelyDeletedSearchIndex() {
// update safely deleted search index to delete outdated info,
// indicating that insert nodes whose search index are before this value
can be deleted
@@ -406,6 +446,7 @@ public class LogDispatcher {
}
public Batch getBatch() {
+ final IoTConsensusConfig currentConfig = config;
long startIndex = syncStatus.getNextSendingIndex();
long maxIndex;
synchronized (impl.getIndexObject()) {
@@ -420,7 +461,7 @@ public class LogDispatcher {
// Use drainTo instead of poll to reduce lock overhead
pendingEntries.drainTo(
bufferedEntries,
- config.getReplication().getMaxLogEntriesNumPerBatch() -
bufferedEntries.size());
+ currentConfig.getReplication().getMaxLogEntriesNumPerBatch() -
bufferedEntries.size());
}
// remove all request that searchIndex < startIndex
Iterator<IndexedConsensusRequest> iterator = bufferedEntries.iterator();
@@ -434,7 +475,7 @@ public class LogDispatcher {
}
}
- Batch batches = new Batch(config);
+ Batch batches = new Batch(currentConfig);
// This condition will be executed in several scenarios:
// 1. restart
// 2. The getBatch() is invoked immediately at the moment the
PendingEntries are consumed
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java
index accc9f7667d..a96abfb1a5a 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java
@@ -31,7 +31,7 @@ import java.util.List;
public class SyncStatus {
private static final Logger LOGGER =
LoggerFactory.getLogger(SyncStatus.class);
- private final IoTConsensusConfig config;
+ private IoTConsensusConfig config;
private final IndexController controller;
private final LinkedList<Batch> pendingBatches = new LinkedList<>();
private final IoTConsensusMemoryManager iotConsensusMemoryManager =
@@ -42,6 +42,11 @@ public class SyncStatus {
this.config = config;
}
+ public synchronized void reloadConfig(IoTConsensusConfig config) {
+ this.config = config;
+ notifyAll();
+ }
+
/**
* we may block here if the synchronization pipeline is full.
*
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java
index f87d8cd7f98..6d6bae6165e 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java
@@ -37,6 +37,20 @@ import static org.junit.Assert.assertTrue;
public class IoTConsensusMemoryManagerTest {
+ @Test
+ public void testInitUpdatesMemoryLimits() {
+ IoTConsensusMemoryManager memoryManager =
IoTConsensusMemoryManager.getInstance();
+ long previousMaxMemory = memoryManager.getMaxMemorySizeInByte();
+ long previousMaxQueueMemory =
memoryManager.getMaxMemorySizeForQueueInByte();
+ try {
+ memoryManager.init(1024, 512);
+ assertEquals(1024L, memoryManager.getMaxMemorySizeInByte().longValue());
+ assertEquals(512L,
memoryManager.getMaxMemorySizeForQueueInByte().longValue());
+ } finally {
+ memoryManager.init(previousMaxMemory, previousMaxQueueMemory);
+ }
+ }
+
@Test
public void testAllocateQueue() {
IoTConsensusMemoryManager memoryManager =
IoTConsensusMemoryManager.getInstance();
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java
new file mode 100644
index 00000000000..c018f7034bb
--- /dev/null
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java
@@ -0,0 +1,286 @@
+/*
+ * 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.logdispatcher;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
+import org.apache.iotdb.consensus.config.IoTConsensusConfig;
+import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
+import org.apache.iotdb.consensus.iot.client.DispatchLogHandler;
+import org.apache.iotdb.consensus.iot.thrift.TLogEntry;
+import org.apache.iotdb.consensus.iot.util.TestEntry;
+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.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.TreeSet;
+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 java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public class LogDispatcherTest {
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testWaitForBatchAccumulationAfterFirstRequest() throws Exception
{
+ final Peer localPeer = createPeer(1, 6667);
+ final Peer remotePeer = createPeer(2, 6668);
+ final IoTConsensusConfig config = IoTConsensusConfig.newBuilder().build();
+ final ScheduledExecutorService backgroundTaskService =
+ Executors.newSingleThreadScheduledExecutor();
+ final ExecutorService executorService =
Executors.newSingleThreadExecutor();
+ LogDispatcher.LogDispatcherThread dispatcherThread = null;
+ Future<?> dispatcherFuture = null;
+ try {
+ final IoTConsensusServerImpl server =
+ createServer(
+ localPeer, Collections.singletonList(localPeer), config,
backgroundTaskService);
+ final Batch batch = createBatch(config, 1);
+ final CountDownLatch accumulationWaitInvoked = new CountDownLatch(1);
+ final AtomicInteger getBatchInvocations = new AtomicInteger();
+ dispatcherThread =
+ server.getLogDispatcher().new LogDispatcherThread(remotePeer,
config, 0) {
+ @Override
+ public Batch getBatch() {
+ return getBatchInvocations.getAndIncrement() == 0 ? new
Batch(config) : batch;
+ }
+
+ @Override
+ void waitForBatchAccumulation(long waitingTimeInMs) {
+ accumulationWaitInvoked.countDown();
+ }
+
+ @Override
+ public void sendBatchAsync(Batch sentBatch, DispatchLogHandler
handler) {
+ getSyncStatus().removeBatch(sentBatch);
+ Thread.currentThread().interrupt();
+ }
+ };
+ assertTrue(
+ dispatcherThread.offer(
+ new IndexedConsensusRequest(
+ 1, Collections.singletonList(new TestEntry(1, localPeer)))));
+
+ dispatcherFuture = executorService.submit(dispatcherThread);
+
+ assertTrue(accumulationWaitInvoked.await(5, TimeUnit.SECONDS));
+ dispatcherFuture.get(5, TimeUnit.SECONDS);
+ } finally {
+ if (dispatcherFuture != null) {
+ dispatcherFuture.cancel(true);
+ }
+ executorService.shutdownNow();
+ executorService.awaitTermination(5, TimeUnit.SECONDS);
+ if (dispatcherThread != null) {
+ dispatcherThread.stop();
+ }
+ backgroundTaskService.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testBatchAccumulationStopsWhenBatchIsFull() throws Exception {
+ final Peer localPeer = createPeer(1, 6687);
+ final Peer remotePeer = createPeer(2, 6688);
+ final IoTConsensusConfig config =
+ IoTConsensusConfig.newBuilder()
+ .setReplication(
+ IoTConsensusConfig.Replication.newBuilder()
+ .setMaxLogEntriesNumPerBatch(2)
+ .setMaxWaitingTimeForAccumulatingBatchInMs(10_000)
+ .build())
+ .build();
+ final ScheduledExecutorService backgroundTaskService =
+ Executors.newSingleThreadScheduledExecutor();
+ final ExecutorService executorService =
Executors.newSingleThreadExecutor();
+ LogDispatcher.LogDispatcherThread dispatcherThread = null;
+ Future<?> dispatcherFuture = null;
+ try {
+ final IoTConsensusServerImpl server =
+ createServer(
+ localPeer, Arrays.asList(localPeer, remotePeer), config,
backgroundTaskService);
+ final CountDownLatch batchSent = new CountDownLatch(1);
+ final AtomicInteger getBatchInvocations = new AtomicInteger();
+ dispatcherThread =
+ server.getLogDispatcher().new LogDispatcherThread(remotePeer,
config, 0) {
+ @Override
+ public Batch getBatch() {
+ return getBatchInvocations.getAndIncrement() == 0
+ ? new Batch(config)
+ : createBatch(config, 1);
+ }
+
+ @Override
+ public void sendBatchAsync(Batch sentBatch, DispatchLogHandler
handler) {
+ assertEquals(0, getPendingEntriesSize());
+ batchSent.countDown();
+ Thread.currentThread().interrupt();
+ }
+ };
+ assertTrue(
+ dispatcherThread.offer(
+ new IndexedConsensusRequest(
+ 1, Collections.singletonList(new TestEntry(1, localPeer)))));
+ assertTrue(
+ dispatcherThread.offer(
+ new IndexedConsensusRequest(
+ 2, Collections.singletonList(new TestEntry(2, localPeer)))));
+
+ dispatcherFuture = executorService.submit(dispatcherThread);
+ assertTrue(batchSent.await(2, TimeUnit.SECONDS));
+ dispatcherFuture.get(2, TimeUnit.SECONDS);
+ } finally {
+ if (dispatcherFuture != null) {
+ dispatcherFuture.cancel(true);
+ }
+ executorService.shutdownNow();
+ executorService.awaitTermination(5, TimeUnit.SECONDS);
+ if (dispatcherThread != null) {
+ dispatcherThread.stop();
+ }
+ backgroundTaskService.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testReloadConfigUpdatesExistingDispatcherPipeline() throws
Exception {
+ final Peer localPeer = createPeer(1, 6677);
+ final Peer remotePeer = createPeer(2, 6678);
+ final IoTConsensusConfig initialConfig =
+ IoTConsensusConfig.newBuilder()
+ .setReplication(
+ IoTConsensusConfig.Replication.newBuilder()
+ .setMaxLogEntriesNumPerBatch(1)
+ .setMaxPendingBatchesNum(1)
+ .build())
+ .build();
+ final ScheduledExecutorService backgroundTaskService =
+ Executors.newSingleThreadScheduledExecutor();
+ final ExecutorService executorService =
Executors.newSingleThreadExecutor();
+ LogDispatcher dispatcher = null;
+ Future<?> secondBatchFuture = null;
+ try {
+ final IoTConsensusServerImpl server =
+ createServer(
+ localPeer,
+ Arrays.asList(localPeer, remotePeer),
+ initialConfig,
+ backgroundTaskService);
+ dispatcher = server.getLogDispatcher();
+ final LogDispatcher.LogDispatcherThread dispatcherThread =
getOnlyThread(dispatcher);
+ dispatcher.start();
+
+ final SyncStatus syncStatus = dispatcherThread.getSyncStatus();
+ syncStatus.addNextBatch(createBatch(initialConfig, 1));
+ final CountDownLatch secondBatchAttempted = new CountDownLatch(1);
+ secondBatchFuture =
+ executorService.submit(
+ () -> {
+ secondBatchAttempted.countDown();
+ syncStatus.addNextBatch(createBatch(initialConfig, 2));
+ return null;
+ });
+ assertTrue(secondBatchAttempted.await(5, TimeUnit.SECONDS));
+ Thread.sleep(100);
+ assertFalse(secondBatchFuture.isDone());
+
+ final IoTConsensusConfig reloadedConfig =
+ IoTConsensusConfig.newBuilder()
+ .setReplication(
+ IoTConsensusConfig.Replication.newBuilder()
+ .setMaxLogEntriesNumPerBatch(2)
+ .setMaxPendingBatchesNum(2)
+ .build())
+ .build();
+ server.reloadConsensusConfig(reloadedConfig);
+
+ secondBatchFuture.get(5, TimeUnit.SECONDS);
+ assertSame(reloadedConfig, dispatcherThread.getConfig());
+ assertEquals(2, syncStatus.getPendingBatches().size());
+ } finally {
+ if (secondBatchFuture != null) {
+ secondBatchFuture.cancel(true);
+ }
+ executorService.shutdownNow();
+ executorService.awaitTermination(5, TimeUnit.SECONDS);
+ if (dispatcher != null) {
+ dispatcher.stop();
+ }
+ backgroundTaskService.shutdownNow();
+ }
+ }
+
+ private IoTConsensusServerImpl createServer(
+ Peer localPeer,
+ List<Peer> configuration,
+ IoTConsensusConfig config,
+ ScheduledExecutorService backgroundTaskService)
+ throws Exception {
+ return new IoTConsensusServerImpl(
+ temporaryFolder.newFolder().getAbsolutePath(),
+ localPeer,
+ new TreeSet<>(configuration),
+ new TestStateMachine(),
+ backgroundTaskService,
+ null,
+ null,
+ config);
+ }
+
+ private static Peer createPeer(int nodeId, int port) {
+ return new Peer(new DataRegionId(1), nodeId, new TEndPoint("127.0.0.1",
port));
+ }
+
+ private static Batch createBatch(IoTConsensusConfig config, long
searchIndex) {
+ final Batch batch = new Batch(config);
+ batch.addTLogEntry(new
TLogEntry().setSearchIndex(searchIndex).setMemorySize(1));
+ batch.buildIndex();
+ return batch;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static LogDispatcher.LogDispatcherThread getOnlyThread(LogDispatcher
dispatcher)
+ throws Exception {
+ final Field threadsField = LogDispatcher.class.getDeclaredField("threads");
+ threadsField.setAccessible(true);
+ final List<LogDispatcher.LogDispatcherThread> threads =
+ (List<LogDispatcher.LogDispatcherThread>) threadsField.get(dispatcher);
+ assertEquals(1, threads.size());
+ return threads.get(0);
+ }
+}
