From 1486d885d3d274809fca5461d19dc769021c3f6b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 8 Sep 2026 12:31:02 +0800 Subject: [PATCH 1/2] Revert "Fix IoTConsensus batch accumulation by byte size (#18560)" This reverts commit c41838e1386706003125f333ddadb25f6132433d. --- .../consensus/iot/logdispatcher/Batch.java | 6 +- .../iot/logdispatcher/LogDispatcher.java | 18 ++--- .../iot/logdispatcher/LogDispatcherTest.java | 66 ------------------- 3 files changed, 7 insertions(+), 83 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java index 26fcddb5d2a4..55569b8a34fc 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java @@ -69,10 +69,6 @@ public void addTLogEntry(TLogEntry entry, boolean containsUserData) { } public boolean canAccumulate() { - return canAccumulate(config, logEntries.size(), memorySize); - } - - static boolean canAccumulate(IoTConsensusConfig config, int logEntriesSize, long memorySize) { // When reading entries from the WAL, the memory size is calculated based on the serialized // size, which can be significantly smaller than the actual size. // Thus, we add a multiplier to sender's memory size to estimate the receiver's memory cost. @@ -81,7 +77,7 @@ static boolean canAccumulate(IoTConsensusConfig config, int logEntriesSize, long long senderMemSize = LogDispatcher.getSenderMemSizeSum().get(); double multiplier = senderMemSize > 0 ? (double) receiverMemSize / senderMemSize : 1.0; multiplier = Math.max(multiplier, 1.0); - return logEntriesSize < config.getReplication().getMaxLogEntriesNumPerBatch() + return logEntries.size() < config.getReplication().getMaxLogEntriesNumPerBatch() && ((long) (memorySize * multiplier)) < config.getReplication().getMaxSizePerBatch(); } 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 d9b34b9f79db..1ff6579e0ded 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 @@ -432,16 +432,12 @@ void waitForBatchAccumulation(long waitingTimeInMs) throws InterruptedException } final long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(waitingTimeInMs); - final IoTConsensusConfig currentConfig = config; - int accumulatedEntries = bufferedEntries.size(); - long accumulatedMemorySize = - bufferedEntries.stream().mapToLong(IndexedConsensusRequest::getMemorySize).sum(); - - // Keep collecting while the batch is below both its entry and memory limits. A plain sleep, - // or checking only the entry limit, makes the dispatcher wait for the full accumulation - // interval after a batch has already reached its memory limit. This unnecessarily throttles - // IoTConsensus when each request contains a large tablet. - while (Batch.canAccumulate(currentConfig, accumulatedEntries, accumulatedMemorySize)) { + 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; @@ -453,8 +449,6 @@ void waitForBatchAccumulation(long waitingTimeInMs) throws InterruptedException return; } bufferedEntries.add(request); - accumulatedEntries++; - accumulatedMemorySize += request.getMemorySize(); } } 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 index a6e299748d21..fce84147dd5c 100644 --- 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 @@ -177,72 +177,6 @@ public void sendBatchAsync(Batch sentBatch, DispatchLogHandler handler) { } } - @Test - public void testBatchAccumulationStopsWhenMemoryLimitIsReached() throws Exception { - final Peer localPeer = createPeer(1, 6697); - final Peer remotePeer = createPeer(2, 6698); - final IoTConsensusConfig config = - IoTConsensusConfig.newBuilder() - .setReplication( - IoTConsensusConfig.Replication.newBuilder() - .setMaxLogEntriesNumPerBatch(1024) - .setMaxSizePerBatch(1) - .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, Collections.singletonList(localPeer), 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(1, getPendingEntriesSize()); - assertEquals(1, getBufferRequestSize()); - batchSent.countDown(); - Thread.currentThread().interrupt(); - } - }; - final IndexedConsensusRequest firstRequest = - new IndexedConsensusRequest(1, Collections.singletonList(new TestEntry(1, localPeer))); - firstRequest.buildSerializedRequests(); - final IndexedConsensusRequest secondRequest = - new IndexedConsensusRequest(2, Collections.singletonList(new TestEntry(2, localPeer))); - secondRequest.buildSerializedRequests(); - assertTrue(dispatcherThread.offer(firstRequest)); - assertTrue(dispatcherThread.offer(secondRequest)); - - 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); From fe419da78deb4891bf7d85bbed20f2d208722c57 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 8 Sep 2026 14:22:48 +0800 Subject: [PATCH 2/2] Revert "Fix IoTConsensus batch accumulation latency (#18522)" This reverts commit ebd9c2072e3195a510d784e7b0fb4422fa419935. --- .../iotdb/consensus/iot/IoTConsensus.java | 5 +- .../consensus/iot/IoTConsensusServerImpl.java | 3 +- .../IoTConsensusMemoryManager.java | 6 +- .../iot/logdispatcher/LogDispatcher.java | 53 +--- .../iot/logdispatcher/SyncStatus.java | 7 +- .../IoTConsensusMemoryManagerTest.java | 22 -- .../iot/logdispatcher/LogDispatcherTest.java | 288 ------------------ 7 files changed, 10 insertions(+), 374 deletions(-) delete mode 100644 iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java 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 ede52c92ed3b..9c0692132f03 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 @@ -108,7 +108,7 @@ public class IoTConsensus implements IConsensus { private final RegisterManager registerManager = new RegisterManager(); private final UserDataTransferAuditHandler userDataTransferAuditHandler; private final UserDataTransferAuditClassifier userDataTransferAuditClassifier; - private volatile IoTConsensusConfig config; + private IoTConsensusConfig config; /** * Optional callback invoked after a new local peer is created via {@link #createLocalPeer}. Used @@ -549,9 +549,6 @@ public String getRegionDirFromConsensusGroupId(ConsensusGroupId groupId) { public void reloadConsensusConfig(ConsensusConfig consensusConfig) { config = consensusConfig.getIotConsensusConfig(); - IoTConsensusMemoryManager.getInstance() - .updateMaxMemoryRatioForQueue(config.getReplication().getMaxMemoryRatioForQueue()); - 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 6e86ec72a3e3..f4448e81f34b 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 @@ -152,7 +152,7 @@ public class IoTConsensusServerImpl { private final Set configuration = ConcurrentHashMap.newKeySet(); private final AtomicLong searchIndex; private final LogDispatcher logDispatcher; - private volatile IoTConsensusConfig config; + private IoTConsensusConfig config; private final ConsensusReqReader consensusReqReader; private volatile boolean active; private String newSnapshotDirName; @@ -1474,7 +1474,6 @@ public String getConsensusGroupId() { /** 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 1247a45129e1..161494a5fe8e 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 @@ -37,7 +37,7 @@ public class IoTConsensusMemoryManager { private final AtomicLong syncMemorySizeInByte = new AtomicLong(0); private IMemoryBlock memoryBlock = new AtomicLongMemoryBlock("Consensus-Default", null, Runtime.getRuntime().maxMemory() / 10); - private volatile double maxMemoryRatioForQueue = 0.6; + private Double maxMemoryRatioForQueue = 0.6; private IoTConsensusMemoryManager() { MetricService.getInstance().addMetricSet(new IoTConsensusMemoryManagerMetrics(this)); @@ -158,10 +158,6 @@ public void init(IMemoryBlock memoryBlock, double maxMemoryRatioForQueue) { this.maxMemoryRatioForQueue = maxMemoryRatioForQueue; } - public void updateMaxMemoryRatioForQueue(double maxMemoryRatioForQueue) { - this.maxMemoryRatioForQueue = maxMemoryRatioForQueue; - } - @TestOnly public void reset() { this.memoryBlock.release(this.memoryBlock.getUsedMemoryInBytes()); 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 1ff6579e0ded..304badbd8aef 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 @@ -183,10 +183,6 @@ public synchronized void checkAndFlushIndex() { impl.checkAndUpdateSafeDeletedSearchIndex(); } - public synchronized void reloadConfig(IoTConsensusConfig config) { - threads.forEach(thread -> thread.reloadConfig(config)); - } - public void offer(IndexedConsensusRequest request) { offer(request, true); } @@ -234,7 +230,7 @@ public class LogDispatcherThread implements Runnable { private static final long PENDING_REQUEST_TAKING_TIME_OUT_IN_MS = 10_000L; private static final long START_INDEX = 1; - private volatile IoTConsensusConfig config; + private final IoTConsensusConfig config; private final Peer peer; private final IndexController controller; // A sliding window class that manages asynchronous pendingBatches @@ -293,11 +289,6 @@ public IoTConsensusConfig getConfig() { return config; } - private void reloadConfig(IoTConsensusConfig config) { - this.config = config; - syncStatus.reloadConfig(config); - } - public int getPendingEntriesSize() { return pendingEntries.size(); } @@ -384,16 +375,11 @@ public void run() { IndexedConsensusRequest request = pendingEntries.poll(calculateIdlePollTimeoutInMs(), TimeUnit.MILLISECONDS); 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 (shouldWaitForBatchAccumulation) { - waitForBatchAccumulation( - currentConfig.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs()); + if (pendingEntries.size() <= config.getReplication().getMaxLogEntriesNumPerBatch() + && bufferedEntries.isEmpty()) { + Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs()); } } else { maybeSendIdleWriterSafeTimeBarrier(); @@ -426,32 +412,6 @@ public void run() { logger.info(IoTConsensusMessages.DISPATCHER_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 @@ -468,7 +428,6 @@ public void updateSafelyDeletedSearchIndex() { public Batch getBatch() { - final IoTConsensusConfig currentConfig = config; long startIndex = syncStatus.getNextSendingIndex(); long maxIndex; synchronized (impl.getIndexObject()) { @@ -483,7 +442,7 @@ public Batch getBatch() { // Use drainTo instead of poll to reduce lock overhead pendingEntries.drainTo( bufferedEntries, - currentConfig.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size()); + config.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size()); } // remove all request that searchIndex < startIndex Iterator iterator = bufferedEntries.iterator(); @@ -497,7 +456,7 @@ public Batch getBatch() { } } - Batch batches = new Batch(currentConfig); + Batch batches = new Batch(config); // 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 3df8a720614a..1749384f5491 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 @@ -32,7 +32,7 @@ public class SyncStatus { private static final Logger LOGGER = LoggerFactory.getLogger(SyncStatus.class); - private IoTConsensusConfig config; + private final IoTConsensusConfig config; private final IndexController controller; private final LinkedList pendingBatches = new LinkedList<>(); private final IoTConsensusMemoryManager iotConsensusMemoryManager = @@ -43,11 +43,6 @@ public SyncStatus(IndexController controller, IoTConsensusConfig config) { 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 88ea61cb2af6..9bddb2987161 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 @@ -41,14 +41,11 @@ public class IoTConsensusMemoryManagerTest { private IMemoryBlock previousMemoryBlock; - private double previousMaxMemoryRatioForQueue; private long memoryBlockSize = 16 * 1024L; @Before public void setUp() throws Exception { previousMemoryBlock = IoTConsensusMemoryManager.getInstance().getMemoryBlock(); - previousMaxMemoryRatioForQueue = - IoTConsensusMemoryManager.getInstance().getMaxMemoryRatioForQueue(); IoTConsensusMemoryManager.getInstance() .setMemoryBlock(new AtomicLongMemoryBlock("Test", null, memoryBlockSize)); IoTConsensusMemoryManager.getInstance().reset(); @@ -58,8 +55,6 @@ public void setUp() throws Exception { public void tearDown() throws Exception { IoTConsensusMemoryManager.getInstance().reset(); IoTConsensusMemoryManager.getInstance().setMemoryBlock(previousMemoryBlock); - IoTConsensusMemoryManager.getInstance() - .updateMaxMemoryRatioForQueue(previousMaxMemoryRatioForQueue); } @Test @@ -126,23 +121,6 @@ public void testClearUnserializedRequest() { assertEquals(0L, request.getRetainedMemorySize()); } - @Test - public void testUpdateMaxMemoryRatioForQueue() { - final IndexedConsensusRequest request = - new IndexedConsensusRequest( - 1, - Collections.singletonList( - new ByteBufferConsensusRequest(ByteBuffer.allocate((int) (memoryBlockSize / 3))))); - request.buildSerializedRequests(); - - IoTConsensusMemoryManager.getInstance().updateMaxMemoryRatioForQueue(0.25); - assertFalse(IoTConsensusMemoryManager.getInstance().reserve(request)); - - IoTConsensusMemoryManager.getInstance().updateMaxMemoryRatioForQueue(0.5); - assertTrue(IoTConsensusMemoryManager.getInstance().reserve(request)); - IoTConsensusMemoryManager.getInstance().free(request); - } - private void testReserveAndRelease(int numReservation) { int allocationSize = 1; long allocatedSize = 0; 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 deleted file mode 100644 index fce84147dd5c..000000000000 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java +++ /dev/null @@ -1,288 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.iotdb.consensus.iot.logdispatcher; - -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.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.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 configuration, - IoTConsensusConfig config, - ScheduledExecutorService backgroundTaskService) - throws Exception { - return new IoTConsensusServerImpl( - temporaryFolder.newFolder().getAbsolutePath(), - null, - DirectoryStrategyType.SEQUENCE_STRATEGY, - localPeer, - 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 threads = - (List) threadsField.get(dispatcher); - assertEquals(1, threads.size()); - return threads.get(0); - } -}