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 1663766b3d1 Revert "fix: retry SyncStatus batch memory reservation
(#18453)"
1663766b3d1 is described below
commit 1663766b3d11d058a41a007c999e26db438c2991
Author: Tian Jiang <[email protected]>
AuthorDate: Mon Aug 31 20:24:39 2026 +0800
Revert "fix: retry SyncStatus batch memory reservation (#18453)"
This reverts commit 2939325ea5ffd465f4e46b3da146697f9df14650.
---
.../consensus/iot/logdispatcher/SyncStatus.java | 13 ++---
.../iot/logdispatcher/SyncStatusTest.java | 64 ----------------------
2 files changed, 4 insertions(+), 73 deletions(-)
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 9e6af375828..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
@@ -53,15 +53,10 @@ public class SyncStatus {
* @throws InterruptedException
*/
public synchronized void addNextBatch(Batch batch) throws
InterruptedException {
- while (true) {
- while (pendingBatches.size() >=
config.getReplication().getMaxPendingBatchesNum()) {
- wait();
- }
- if (iotConsensusMemoryManager.reserve(batch)) {
- break;
- }
- // Memory may be freed by another SyncStatus, which cannot notify this
monitor.
- wait(Math.max(1, config.getReplication().getBasicRetryWaitTimeMs()));
+ while ((pendingBatches.size() >=
config.getReplication().getMaxPendingBatchesNum()
+ || !iotConsensusMemoryManager.reserve(batch))
+ && !Thread.interrupted()) {
+ wait();
}
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatusTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatusTest.java
index c70b97e7131..be81c69f7f7 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatusTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatusTest.java
@@ -21,8 +21,6 @@ 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.memory.AtomicLongMemoryBlock;
-import org.apache.iotdb.commons.memory.IMemoryBlock;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.config.IoTConsensusConfig;
import org.apache.iotdb.consensus.iot.thrift.TLogEntry;
@@ -38,14 +36,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
public class SyncStatusTest {
@@ -251,59 +242,4 @@ public class SyncStatusTest {
Assert.assertEquals(
config.getReplication().getMaxPendingBatchesNum() + 1,
status.getNextSendingIndex());
}
-
- @Test
- public void testFirstBatchRetriesMemoryReservation()
- throws InterruptedException, ExecutionException, TimeoutException {
- IndexController controller =
- new IndexController(storageDir.getAbsolutePath(), peer, 0,
CHECK_POINT_GAP);
- IoTConsensusConfig retryConfig =
- IoTConsensusConfig.newBuilder()
- .setReplication(
-
IoTConsensusConfig.Replication.newBuilder().setBasicRetryWaitTimeMs(10).build())
- .build();
- SyncStatus status = new SyncStatus(controller, retryConfig);
- TLogEntry logEntry = new TLogEntry().setSearchIndex(1).setMemorySize(1);
- Batch batch = new Batch(retryConfig);
- batch.addTLogEntry(logEntry);
- batch.buildIndex();
-
- IoTConsensusMemoryManager memoryManager =
IoTConsensusMemoryManager.getInstance();
- IMemoryBlock previousMemoryBlock = memoryManager.getMemoryBlock();
- CountDownLatch firstAllocationFailed = new CountDownLatch(1);
- AtomicBoolean rejectAllocation = new AtomicBoolean(true);
- IMemoryBlock memoryBlock =
- new AtomicLongMemoryBlock("SyncStatusTest", null,
batch.getMemorySize()) {
- @Override
- public boolean allocate(long sizeInByte) {
- if (rejectAllocation.compareAndSet(true, false)) {
- firstAllocationFailed.countDown();
- return false;
- }
- return super.allocate(sizeInByte);
- }
- };
- ExecutorService executor = Executors.newSingleThreadExecutor();
- memoryManager.setMemoryBlock(memoryBlock);
- try {
- Future<?> future =
- executor.submit(
- () -> {
- status.addNextBatch(batch);
- return null;
- });
-
- Assert.assertTrue(firstAllocationFailed.await(5, TimeUnit.SECONDS));
- future.get(5, TimeUnit.SECONDS);
-
- Assert.assertEquals(1, status.getPendingBatches().size());
- status.removeBatch(batch);
- Assert.assertEquals(0, status.getPendingBatches().size());
- } finally {
- executor.shutdownNow();
- executor.awaitTermination(5, TimeUnit.SECONDS);
- status.free();
- memoryManager.setMemoryBlock(previousMemoryBlock);
- }
- }
}
