You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2026/08/16 01:50:55 UTC
(rocketmq) 04/07: fix: handle DLedger control entries during recovery
This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch codex/dledger-latest-pr336-adapter
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit f3374464396076ce8d88fff9da924ba7c466ee57
Author: 通融 <ro...@alibaba-inc.com>
AuthorDate: Sat Aug 15 19:37:16 2026 +0800
fix: handle DLedger control entries during recovery
---
.../controller/impl/DLedgerControllerTest.java | 46 +-
store/BUILD.bazel | 1 +
.../apache/rocketmq/store/DefaultMessageStore.java | 13 +-
.../rocketmq/store/dledger/DLedgerCommitLog.java | 129 ++++-
.../store/dledger/DLedgerLatestCommitLogTest.java | 592 +++++++++++++++++++++
5 files changed, 748 insertions(+), 33 deletions(-)
diff --git a/controller/src/test/java/org/apache/rocketmq/controller/impl/DLedgerControllerTest.java b/controller/src/test/java/org/apache/rocketmq/controller/impl/DLedgerControllerTest.java
index 32e7859a58..6f448a7075 100644
--- a/controller/src/test/java/org/apache/rocketmq/controller/impl/DLedgerControllerTest.java
+++ b/controller/src/test/java/org/apache/rocketmq/controller/impl/DLedgerControllerTest.java
@@ -40,6 +40,7 @@ import org.junit.Test;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -60,9 +61,15 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class DLedgerControllerTest {
+ private static int port = 30000;
private List<String> baseDirs;
private List<DLedgerController> controllers;
+ private static synchronized int nextPort() {
+ port += 10;
+ return port;
+ }
+
public DLedgerController launchController(final String group, final String peers, final String selfId,
final boolean isEnableElectUncleanMaster) {
String tmpdir = System.getProperty("java.io.tmpdir");
@@ -172,7 +179,9 @@ public class DLedgerControllerTest {
public DLedgerController mockMetaData(boolean enableElectUncleanMaster) throws Exception {
String group = UUID.randomUUID().toString();
- String peers = String.format("n0-localhost:%d;n1-localhost:%d;n2-localhost:%d", 30000, 30001, 30002);
+ int basePort = nextPort();
+ String peers = String.format("n0-localhost:%d;n1-localhost:%d;n2-localhost:%d",
+ basePort, basePort + 1, basePort + 2);
DLedgerController c0 = launchController(group, peers, "n0", enableElectUncleanMaster);
DLedgerController c1 = launchController(group, peers, "n1", enableElectUncleanMaster);
DLedgerController c2 = launchController(group, peers, "n2", enableElectUncleanMaster);
@@ -237,6 +246,36 @@ public class DLedgerControllerTest {
assertNotEquals(DEFAULT_IP[0], response.getMasterAddress());
}
+ @Test
+ public void testRestartAllControllersRecoversStateBeforeNewEvent() throws Exception {
+ DLedgerController originalLeader = mockMetaData(false);
+ String group = originalLeader.getControllerConfig().getControllerDLegerGroup();
+ String peers = originalLeader.getControllerConfig().getControllerDLegerPeers();
+ List<String> selfIds = controllers.stream()
+ .map(controller -> controller.getControllerConfig().getControllerDLegerSelfId())
+ .collect(Collectors.toList());
+
+ List<DLedgerController> originalControllers = new ArrayList<>(controllers);
+ for (DLedgerController controller : originalControllers) {
+ controller.shutdown();
+ }
+ controllers.clear();
+ for (String selfId : selfIds) {
+ controllers.add(launchController(group, peers, selfId, false));
+ }
+ DLedgerController restartedLeader = waitLeader(controllers);
+
+ RemotingCommand response = restartedLeader
+ .getReplicaInfo(new GetReplicaInfoRequestHeader(DEFAULT_BROKER_NAME)).get(10, TimeUnit.SECONDS);
+ assertEquals(ResponseCode.SUCCESS, response.getCode());
+ GetReplicaInfoResponseHeader replicaInfo =
+ (GetReplicaInfoResponseHeader) response.readCustomHeader();
+ SyncStateSet syncStateSet = RemotingSerializable.decode(response.getBody(), SyncStateSet.class);
+ assertEquals(1L, replicaInfo.getMasterBrokerId().longValue());
+ assertEquals(DEFAULT_IP[0], replicaInfo.getMasterAddress());
+ assertEquals(new HashSet<>(Arrays.asList(1L, 2L, 3L)), syncStateSet.getSyncStateSet());
+ }
+
@Test
public void testBrokerLifecycleListener() throws Exception {
final DLedgerController leader = mockMetaData(false);
@@ -250,6 +289,11 @@ public class DLedgerControllerTest {
dLedgerController.shutdown();
controllers.remove(dLedgerController);
}
+ await().atMost(Duration.ofSeconds(10)).until(() ->
+ leader.getMemberState().getPeersLiveTable().size() == leader.getMemberState().peerSize() - 1
+ && leader.getMemberState().getPeersLiveTable().values().stream()
+ .noneMatch(Boolean.TRUE::equals));
+ await().atMost(Duration.ofSeconds(10)).until(() -> !leader.isLeaderState());
final ElectMasterRequestHeader request = ElectMasterRequestHeader.ofControllerTrigger(DEFAULT_BROKER_NAME);
setBrokerElectPolicy(leader, 1L);
diff --git a/store/BUILD.bazel b/store/BUILD.bazel
index 66af7d6b45..510da0e044 100644
--- a/store/BUILD.bazel
+++ b/store/BUILD.bazel
@@ -81,6 +81,7 @@ GenTestRules(
"src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest",
"src/test/java/org/apache/rocketmq/store/HATest",
"src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest",
+ "src/test/java/org/apache/rocketmq/store/dledger/DLedgerLatestCommitLogTest",
"src/test/java/org/apache/rocketmq/store/MappedFileQueueTest",
"src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest",
"src/test/java/org/apache/rocketmq/store/dledger/MixCommitlogTest",
diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index 64ce41e47d..00576f79f4 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -2741,16 +2741,19 @@ public class DefaultMessageStore implements MessageStore {
if (dispatchRequest.isSuccess()) {
if (size > 0) {
- currentReputTimestamp = dispatchRequest.getStoreTimestamp();
- DefaultMessageStore.this.doDispatch(dispatchRequest);
+ if (dispatchRequest.getMsgSize() > 0) {
+ currentReputTimestamp = dispatchRequest.getStoreTimestamp();
+ DefaultMessageStore.this.doDispatch(dispatchRequest);
- if (isNotifyMessageArriveWhenReput()) {
- notifyMessageArriveIfNecessary(dispatchRequest);
+ if (isNotifyMessageArriveWhenReput()) {
+ notifyMessageArriveIfNecessary(dispatchRequest);
+ }
}
this.reputFromOffset += size;
readSize += size;
- if (!DefaultMessageStore.this.getMessageStoreConfig().isDuplicationEnable() &&
+ if (dispatchRequest.getMsgSize() > 0
+ && !DefaultMessageStore.this.getMessageStoreConfig().isDuplicationEnable() &&
DefaultMessageStore.this.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE) {
DefaultMessageStore.this.storeStatsService
.getSinglePutMessageTopicTimesTotal(dispatchRequest.getTopic()).add(dispatchRequest.getBatchSize());
diff --git a/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java b/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
index 06f9e0dc42..58f65579e3 100644
--- a/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
@@ -22,6 +22,7 @@ import io.openmessaging.storage.dledger.common.AppendFuture;
import io.openmessaging.storage.dledger.common.BatchAppendFuture;
import io.openmessaging.storage.dledger.entry.DLedgerEntry;
import io.openmessaging.storage.dledger.entry.DLedgerEntryCoder;
+import io.openmessaging.storage.dledger.entry.DLedgerEntryType;
import io.openmessaging.storage.dledger.entry.DLedgerIndexEntry;
import io.openmessaging.storage.dledger.protocol.AppendEntryRequest;
import io.openmessaging.storage.dledger.protocol.AppendEntryResponse;
@@ -103,12 +104,16 @@ public class DLedgerCommitLog extends CommitLog {
dLedgerConfig.setFileReservedHours(defaultMessageStore.getMessageStoreConfig().getFileReservedTime() + 1);
dLedgerConfig.setPreferredLeaderId(defaultMessageStore.getMessageStoreConfig().getPreferredLeaderId());
dLedgerConfig.setEnableBatchAppend(defaultMessageStore.getMessageStoreConfig().isEnableBatchPush());
+ dLedgerConfig.setEnableFastAdvanceCommitIndex(true);
dLedgerConfig.setDiskSpaceRatioToCheckExpired(defaultMessageStore.getMessageStoreConfig().getDiskMaxUsedSpaceRatio() / 100f);
id = Integer.parseInt(dLedgerConfig.getSelfId().substring(1)) + 1;
dLedgerServer = new DLedgerServer(dLedgerConfig);
dLedgerFileStore = (DLedgerMmapFileStore) dLedgerServer.getdLedgerStore();
DLedgerMmapFileStore.AppendHook appendHook = (entry, buffer, bodyOffset) -> {
+ if (entry.getMagic() != DLedgerEntryType.NORMAL.getMagic()) {
+ return;
+ }
assert bodyOffset == DLedgerEntry.BODY_OFFSET;
buffer.position(buffer.position() + bodyOffset + MessageDecoder.PHY_POS_POSITION);
buffer.putLong(entry.getPos() + bodyOffset);
@@ -406,19 +411,24 @@ public class DLedgerCommitLog extends CommitLog {
long mmapFileOffset = 0;
while (true) {
DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover, checkDupInfo);
- int size = dispatchRequest.getMsgSize();
+ int messageSize = dispatchRequest.getMsgSize();
+ int entrySize = dispatchRequest.getBufferSize() == -1
+ ? messageSize : dispatchRequest.getBufferSize();
if (dispatchRequest.isSuccess()) {
- if (size > 0) {
- mmapFileOffset += size;
- if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {
- if (dispatchRequest.getCommitLogOffset() < this.defaultMessageStore.getConfirmOffset()) {
+ if (entrySize > 0) {
+ mmapFileOffset += entrySize;
+ if (messageSize > 0) {
+ if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {
+ if (dispatchRequest.getCommitLogOffset()
+ < this.defaultMessageStore.getConfirmOffset()) {
+ this.defaultMessageStore.doDispatch(dispatchRequest);
+ }
+ } else {
this.defaultMessageStore.doDispatch(dispatchRequest);
}
- } else {
- this.defaultMessageStore.doDispatch(dispatchRequest);
}
- } else if (size == 0) {
+ } else if (entrySize == 0) {
index++;
if (index >= mmapFiles.size()) {
log.info("dledger recover physics file over, last mapped file " + mmapFile.getFileName());
@@ -486,16 +496,53 @@ public class DLedgerCommitLog extends CommitLog {
log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset);
}
- private boolean isMmapFileMatchedRecover(final MmapFile mmapFile, boolean recoverNormally) throws RocksDBException {
- ByteBuffer byteBuffer = mmapFile.sliceByteBuffer();
+ private ByteBuffer firstNormalEntryBody(ByteBuffer byteBuffer) {
+ int limit = byteBuffer.limit();
+ int position = byteBuffer.position();
+ while (limit - position >= Integer.BYTES * 2) {
+ int magic = byteBuffer.getInt(position);
+ int entrySize = byteBuffer.getInt(position + Integer.BYTES);
+ if (magic == MmapFileList.BLANK_MAGIC_CODE || entrySize < DLedgerEntry.BODY_OFFSET
+ || entrySize > limit - position) {
+ return null;
+ }
+ if (magic == DLedgerEntryType.NOOP.getMagic()) {
+ if (entrySize != DLedgerEntry.BODY_OFFSET) {
+ return null;
+ }
+ position += entrySize;
+ continue;
+ }
+ if (magic != DLedgerEntryType.NORMAL.getMagic()) {
+ return null;
+ }
+ ByteBuffer body = byteBuffer.duplicate();
+ body.position(position + DLedgerEntry.BODY_OFFSET);
+ body.limit(position + entrySize);
+ return body.slice();
+ }
+ return null;
+ }
+
+ private boolean isMmapFileMatchedRecover(final MmapFile mmapFile, boolean recoverNormally)
+ throws RocksDBException {
+ ByteBuffer byteBuffer = firstNormalEntryBody(mmapFile.sliceByteBuffer());
+ if (byteBuffer == null
+ || byteBuffer.limit() < MessageDecoder.MESSAGE_MAGIC_CODE_POSITION + Integer.BYTES) {
+ return false;
+ }
- int magicCode = byteBuffer.getInt(DLedgerEntry.BODY_OFFSET + MessageDecoder.MESSAGE_MAGIC_CODE_POSITION);
- if (magicCode != MESSAGE_MAGIC_CODE) {
+ int magicCode = byteBuffer.getInt(MessageDecoder.MESSAGE_MAGIC_CODE_POSITION);
+ if (magicCode != MESSAGE_MAGIC_CODE
+ && magicCode != MessageDecoder.MESSAGE_MAGIC_CODE_V2) {
return false;
}
+ if (byteBuffer.limit() < MessageDecoder.SYSFLAG_POSITION + Integer.BYTES) {
+ return false;
+ }
int storeTimestampPosition;
- int sysFlag = byteBuffer.getInt(DLedgerEntry.BODY_OFFSET + MessageDecoder.SYSFLAG_POSITION);
+ int sysFlag = byteBuffer.getInt(MessageDecoder.SYSFLAG_POSITION);
if ((sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0) {
storeTimestampPosition = MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSITION;
} else {
@@ -503,11 +550,15 @@ public class DLedgerCommitLog extends CommitLog {
storeTimestampPosition = MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSITION + 12;
}
- long storeTimestamp = byteBuffer.getLong(DLedgerEntry.BODY_OFFSET + storeTimestampPosition);
+ if (byteBuffer.limit() < storeTimestampPosition + Long.BYTES
+ || byteBuffer.limit() < MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION + Long.BYTES) {
+ return false;
+ }
+ long storeTimestamp = byteBuffer.getLong(storeTimestampPosition);
if (storeTimestamp == 0) {
return false;
}
- long phyOffset = byteBuffer.getLong(DLedgerEntry.BODY_OFFSET + MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION);
+ long phyOffset = byteBuffer.getLong(MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION);
if (this.defaultMessageStore.getMessageStoreConfig().isMessageIndexEnable()
&& this.defaultMessageStore.getMessageStoreConfig().isMessageIndexSafe()) {
@@ -537,30 +588,54 @@ public class DLedgerCommitLog extends CommitLog {
if (isInrecoveringOldCommitlog) {
return super.checkMessageAndReturnSize(byteBuffer, checkCRC, checkDupInfo, readBody);
}
+ int position = byteBuffer.position();
try {
- int bodyOffset = DLedgerEntry.BODY_OFFSET;
- int pos = byteBuffer.position();
- int magic = byteBuffer.getInt();
+ if (byteBuffer.remaining() < Integer.BYTES * 2) {
+ return new DispatchRequest(-1, false);
+ }
+ int magic = byteBuffer.getInt(position);
//In dledger, this field is size, it must be gt 0, so it could prevent collision
- int magicOld = byteBuffer.getInt();
- if (magicOld == CommitLog.BLANK_MAGIC_CODE
- || magicOld == MessageDecoder.MESSAGE_MAGIC_CODE
- || magicOld == MessageDecoder.MESSAGE_MAGIC_CODE_V2) {
- byteBuffer.position(pos);
+ int entrySize = byteBuffer.getInt(position + Integer.BYTES);
+ if (entrySize == CommitLog.BLANK_MAGIC_CODE
+ || entrySize == MessageDecoder.MESSAGE_MAGIC_CODE
+ || entrySize == MessageDecoder.MESSAGE_MAGIC_CODE_V2) {
return super.checkMessageAndReturnSize(byteBuffer, checkCRC, checkDupInfo, readBody);
}
if (magic == MmapFileList.BLANK_MAGIC_CODE) {
return new DispatchRequest(0, true);
}
- byteBuffer.position(pos + bodyOffset);
- DispatchRequest dispatchRequest = super.checkMessageAndReturnSize(byteBuffer, checkCRC, checkDupInfo, readBody);
+ if (magic == DLedgerEntryType.NOOP.getMagic()) {
+ if (entrySize != DLedgerEntry.BODY_OFFSET || entrySize > byteBuffer.remaining()) {
+ return new DispatchRequest(-1, false);
+ }
+ byteBuffer.position(position + entrySize);
+ DispatchRequest dispatchRequest = new DispatchRequest(0, true);
+ dispatchRequest.setBufferSize(entrySize);
+ return dispatchRequest;
+ }
+ if (magic != DLedgerEntryType.NORMAL.getMagic() || entrySize < DLedgerEntry.BODY_OFFSET
+ || entrySize > byteBuffer.remaining()) {
+ return new DispatchRequest(-1, false);
+ }
+ int entryEnd = position + entrySize;
+ ByteBuffer messageBuffer = byteBuffer.duplicate();
+ messageBuffer.position(position + DLedgerEntry.BODY_OFFSET);
+ messageBuffer.limit(entryEnd);
+ messageBuffer = messageBuffer.slice();
+ DispatchRequest dispatchRequest = super.checkMessageAndReturnSize(
+ messageBuffer, checkCRC, checkDupInfo, readBody);
if (dispatchRequest.isSuccess()) {
- dispatchRequest.setBufferSize(dispatchRequest.getMsgSize() + bodyOffset);
+ if (dispatchRequest.getMsgSize() + DLedgerEntry.BODY_OFFSET != entrySize) {
+ return new DispatchRequest(-1, false);
+ }
+ byteBuffer.position(entryEnd);
+ dispatchRequest.setBufferSize(entrySize);
} else if (dispatchRequest.getMsgSize() > 0) {
- dispatchRequest.setBufferSize(dispatchRequest.getMsgSize() + bodyOffset);
+ dispatchRequest.setBufferSize(dispatchRequest.getMsgSize() + DLedgerEntry.BODY_OFFSET);
}
return dispatchRequest;
} catch (Throwable ignored) {
+ byteBuffer.position(position);
}
return new DispatchRequest(-1, false /* success */);
diff --git a/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerLatestCommitLogTest.java b/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerLatestCommitLogTest.java
new file mode 100644
index 0000000000..cf16b393a1
--- /dev/null
+++ b/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerLatestCommitLogTest.java
@@ -0,0 +1,592 @@
+/*
+ * 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.rocketmq.store.dledger;
+
+import io.openmessaging.storage.dledger.DLedgerServer;
+import io.openmessaging.storage.dledger.common.ReadClosure;
+import io.openmessaging.storage.dledger.common.ReadMode;
+import io.openmessaging.storage.dledger.common.Status;
+import io.openmessaging.storage.dledger.entry.DLedgerEntry;
+import io.openmessaging.storage.dledger.entry.DLedgerEntryCoder;
+import io.openmessaging.storage.dledger.entry.DLedgerEntryType;
+import io.openmessaging.storage.dledger.store.file.DLedgerMmapFileStore;
+import io.openmessaging.storage.dledger.store.file.MmapFileList;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageExtBatch;
+import org.apache.rocketmq.common.message.MessageExtBrokerInner;
+import org.apache.rocketmq.store.DefaultMessageStore;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.GetMessageResult;
+import org.apache.rocketmq.store.PutMessageResult;
+import org.apache.rocketmq.store.PutMessageStatus;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+
+public class DLedgerLatestCommitLogTest extends MessageStoreTestBase {
+
+ private static final int QUEUE_ID = 0;
+
+ @Test
+ public void testUncommittedTailIsNotReadable() throws Exception {
+ String peers = String.format("n0-localhost:%d;n1-localhost:%d", nextPort(), nextPort());
+ DefaultMessageStore leaderStore = null;
+ try {
+ leaderStore = createDledgerMessageStore(
+ createBaseDir(), UUID.randomUUID().toString(), "n0", peers, "n0", false, 0);
+ String topic = UUID.randomUUID().toString();
+ MessageExtBrokerInner message = buildMessage();
+ message.setTopic(topic);
+ message.setQueueId(QUEUE_ID);
+
+ PutMessageResult result = leaderStore.asyncPutMessage(message).get(5, SECONDS);
+ Assert.assertEquals(PutMessageStatus.IN_SYNC_REPLICAS_NOT_ENOUGH, result.getPutMessageStatus());
+ Assert.assertNotNull(result.getAppendMessageResult());
+ Assert.assertTrue(result.getAppendMessageResult().getWroteOffset() > 0);
+
+ DLedgerCommitLog commitLog = commitLog(leaderStore);
+ Assert.assertEquals(-1, commitLog.getdLedgerServer().getMemberState().getCommittedIndex());
+ Assert.assertEquals(-1, commitLog.getCommittedPos());
+ Assert.assertEquals(0, commitLog.getMaxOffset());
+ Assert.assertEquals(0, leaderStore.getMaxOffsetInQueue(topic, QUEUE_ID));
+ Assert.assertNull(commitLog.getData(0));
+ Assert.assertFalse(commitLog.getData(0, 1, ByteBuffer.allocate(1)));
+ Assert.assertNull(commitLog.getMessage(result.getAppendMessageResult().getWroteOffset(), 1));
+ } finally {
+ shutdownAndDestroy(leaderStore);
+ }
+ }
+
+ @Test
+ public void testSingleAndBatchAppendPositions() throws Exception {
+ String peers = String.format("n0-localhost:%d", nextPort());
+ DefaultMessageStore messageStore = null;
+ try {
+ messageStore = createDledgerMessageStore(
+ createBaseDir(), UUID.randomUUID().toString(), "n0", peers, null, false, 0);
+ awaitLeader(Arrays.asList(messageStore));
+ String topic = UUID.randomUUID().toString();
+
+ PutMessageResult singleResult = putSingle(messageStore, topic, 0);
+ PutMessageResult batchResult = putBatch(messageStore, topic, 3, 1);
+
+ Assert.assertTrue(singleResult.getAppendMessageResult().getWroteOffset() > 0);
+ Assert.assertTrue(batchResult.getAppendMessageResult().getWroteOffset()
+ > singleResult.getAppendMessageResult().getWroteOffset());
+ Assert.assertEquals(3, batchResult.getAppendMessageResult().getMsgNum());
+ Assert.assertNotNull(singleResult.getAppendMessageResult().getMsgId());
+ Assert.assertNotNull(batchResult.getAppendMessageResult().getMsgId());
+ Assert.assertEquals(3, batchResult.getAppendMessageResult().getMsgId().split(",").length);
+ awaitStoreReady(messageStore, topic, 4);
+ Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, QUEUE_ID));
+ Assert.assertTrue(commitLog(messageStore).getCommittedPos()
+ > batchResult.getAppendMessageResult().getWroteOffset());
+ doGetMessages(messageStore, topic, QUEUE_ID, 4, 0);
+ } finally {
+ shutdownAndDestroy(messageStore);
+ }
+ }
+
+ @Test
+ public void testThreeNodeElectionAndFailover() throws Exception {
+ String peers = String.format("n0-localhost:%d;n1-localhost:%d;n2-localhost:%d",
+ nextPort(), nextPort(), nextPort());
+ String group = UUID.randomUUID().toString();
+ List<DefaultMessageStore> allStores = new ArrayList<>();
+ try {
+ allStores.add(createDledgerMessageStore(createBaseDir(), group, "n0", peers, null, false, 0));
+ allStores.add(createDledgerMessageStore(createBaseDir(), group, "n1", peers, null, false, 0));
+ allStores.add(createDledgerMessageStore(createBaseDir(), group, "n2", peers, null, false, 0));
+ List<DefaultMessageStore> activeStores = new ArrayList<>(allStores);
+ DefaultMessageStore firstLeader = awaitLeader(activeStores);
+ String topic = UUID.randomUUID().toString();
+
+ putSingle(firstLeader, topic, 0);
+ putSingle(firstLeader, topic, 1);
+ putSingle(firstLeader, topic, 2);
+ for (DefaultMessageStore store : activeStores) {
+ awaitStoreReady(store, topic, 3);
+ }
+ long committedBeforeFailover = commitLog(firstLeader).getCommittedPos();
+
+ firstLeader.shutdown();
+ activeStores.remove(firstLeader);
+ DefaultMessageStore secondLeader = awaitLeader(activeStores);
+ Assert.assertNotSame(firstLeader, secondLeader);
+ awaitStoreReady(secondLeader, topic, 3);
+ Assert.assertTrue(commitLog(secondLeader).getCommittedPos() >= committedBeforeFailover);
+ doGetMessages(secondLeader, topic, QUEUE_ID, 3, 0);
+
+ // Broker-side DLedgerRoleChangeHandler does this before accepting writes on a new leader.
+ secondLeader.recoverTopicQueueTable();
+ putBatch(secondLeader, topic, 3, 3);
+ for (DefaultMessageStore store : activeStores) {
+ awaitStoreReady(store, topic, 6);
+ }
+ doGetMessages(secondLeader, topic, QUEUE_ID, 6, 0);
+ } finally {
+ for (DefaultMessageStore store : allStores) {
+ shutdownAndDestroy(store);
+ }
+ }
+ }
+
+ @Test
+ public void testRestartRecoversCommittedBoundaryBeforeNewWrite() throws Exception {
+ String base = createBaseDir();
+ String peers = String.format("n0-localhost:%d", nextPort());
+ String group = UUID.randomUUID().toString();
+ String topic = UUID.randomUUID().toString();
+ DefaultMessageStore currentStore = null;
+ try {
+ currentStore = createDledgerMessageStore(base, group, "n0", peers, null, false, 0);
+ awaitLeader(Arrays.asList(currentStore));
+ doPutMessages(currentStore, topic, QUEUE_ID, 10, 0);
+ awaitStoreReady(currentStore, topic, 10);
+ doGetMessages(currentStore, topic, QUEUE_ID, 10, 0);
+ long physicalBeforeRestart = currentStore.getMaxPhyOffset();
+ long maxCqOffsetBeforeRestart = currentStore.getMaxOffsetInQueue(topic, QUEUE_ID);
+ List<byte[]> bodiesBeforeRestart = readMessageBodies(currentStore, topic, QUEUE_ID, 10);
+ long committedIndexBeforeRestart = committedIndex(currentStore);
+ Assert.assertTrue(physicalBeforeRestart > 0);
+ Assert.assertEquals(10, maxCqOffsetBeforeRestart);
+ Assert.assertTrue(committedIndexBeforeRestart >= 9);
+
+ currentStore.shutdown();
+ currentStore = createDledgerMessageStore(base, group, "n0", peers, null, false, 0);
+ awaitLeader(Arrays.asList(currentStore));
+ awaitCommittedPast(currentStore, committedIndexBeforeRestart);
+ assertNoopEntry(currentStore, committedIndexBeforeRestart + 1);
+ awaitStoreReady(currentStore, topic, maxCqOffsetBeforeRestart);
+ Assert.assertTrue(currentStore.getMaxPhyOffset() >= physicalBeforeRestart);
+ Assert.assertEquals(0, currentStore.getMinOffsetInQueue(topic, QUEUE_ID));
+ Assert.assertEquals(maxCqOffsetBeforeRestart,
+ currentStore.getMaxOffsetInQueue(topic, QUEUE_ID));
+ assertMessageBodies(currentStore, topic, QUEUE_ID, bodiesBeforeRestart);
+ Assert.assertEquals(commitLog(currentStore).getCommittedPos(), currentStore.getCommitLog().getMaxOffset());
+ doGetMessages(currentStore, topic, QUEUE_ID, 10, 0);
+
+ putSingle(currentStore, topic, 10);
+ awaitStoreReady(currentStore, topic, 11);
+ doGetMessages(currentStore, topic, QUEUE_ID, 11, 0);
+ long committedPosBeforeSecondRestart = commitLog(currentStore).getCommittedPos();
+ long committedIndexBeforeSecondRestart = committedIndex(currentStore);
+
+ currentStore.shutdown();
+ currentStore = createDledgerMessageStore(base, group, "n0", peers, null, true, 0);
+ awaitLeader(Arrays.asList(currentStore));
+ awaitCommittedPast(currentStore, committedIndexBeforeSecondRestart);
+ assertNoopEntry(currentStore, committedIndexBeforeSecondRestart + 1);
+ awaitStoreReady(currentStore, topic, 11);
+ Assert.assertEquals(0, currentStore.getMinOffsetInQueue(topic, QUEUE_ID));
+ Assert.assertTrue(commitLog(currentStore).getCommittedPos() >= committedPosBeforeSecondRestart);
+ Assert.assertEquals(commitLog(currentStore).getCommittedPos(), currentStore.getCommitLog().getMaxOffset());
+ doGetMessages(currentStore, topic, QUEUE_ID, 11, 0);
+ } finally {
+ shutdownAndDestroy(currentStore);
+ }
+ }
+
+ @Test
+ public void testNoopDispatchContractAndBounds() throws Exception {
+ String peers = String.format("n0-localhost:%d", nextPort());
+ DefaultMessageStore messageStore = null;
+ try {
+ messageStore = createDledgerMessageStore(
+ createBaseDir(), UUID.randomUUID().toString(), "n0", peers, null, false, 0);
+ DLedgerCommitLog commitLog = commitLog(messageStore);
+
+ ByteBuffer noopBuffer = ByteBuffer.allocate(DLedgerEntry.BODY_OFFSET);
+ DLedgerEntryCoder.encode(new DLedgerEntry(DLedgerEntryType.NOOP), noopBuffer);
+ DispatchRequest noop = commitLog.checkMessageAndReturnSize(noopBuffer, true, false, false);
+ Assert.assertTrue(noop.isSuccess());
+ Assert.assertEquals(0, noop.getMsgSize());
+ Assert.assertEquals(DLedgerEntry.BODY_OFFSET, noop.getBufferSize());
+ Assert.assertEquals(DLedgerEntry.BODY_OFFSET, noopBuffer.position());
+
+ ByteBuffer undersized = noopHeader(DLedgerEntry.BODY_OFFSET - 1);
+ DispatchRequest invalidSize = commitLog.checkMessageAndReturnSize(undersized, true, false, false);
+ Assert.assertFalse(invalidSize.isSuccess());
+ Assert.assertEquals(-1, invalidSize.getMsgSize());
+ Assert.assertEquals(0, undersized.position());
+
+ ByteBuffer truncated = noopHeader(DLedgerEntry.BODY_OFFSET + 1);
+ DispatchRequest invalidBounds = commitLog.checkMessageAndReturnSize(truncated, true, false, false);
+ Assert.assertFalse(invalidBounds.isSuccess());
+ Assert.assertEquals(-1, invalidBounds.getMsgSize());
+ Assert.assertEquals(0, truncated.position());
+
+ awaitLeader(Arrays.asList(messageStore));
+ String topic = UUID.randomUUID().toString();
+ putSingle(messageStore, topic, 0);
+ awaitStoreReady(messageStore, topic, 1);
+ DLedgerServer server = commitLog.getdLedgerServer();
+ DLedgerEntry normalEntry = server.getDLedgerStore().get(
+ server.getDLedgerStore().getLedgerEndIndex());
+ Assert.assertEquals(DLedgerEntryType.NORMAL.getMagic(), normalEntry.getMagic());
+ byte[] innerMessage = normalEntry.getBody();
+
+ ByteBuffer legalFollowingEntry = normalEntryBuffer(innerMessage, 0, null);
+ byte[] legalFollowingBytes = new byte[legalFollowingEntry.remaining()];
+ legalFollowingEntry.get(legalFollowingBytes);
+ byte[] oversizedInnerMessage = Arrays.copyOf(innerMessage, innerMessage.length);
+ ByteBuffer.wrap(oversizedInnerMessage).putInt(innerMessage.length + Integer.BYTES);
+ ByteBuffer crossingEntry = normalEntryBuffer(
+ oversizedInnerMessage, 0, legalFollowingBytes);
+ DispatchRequest crossingRequest = commitLog.checkMessageAndReturnSize(
+ crossingEntry, true, false, false);
+
+ ByteBuffer mismatchedEntry = normalEntryBuffer(innerMessage, Integer.BYTES, null);
+ DispatchRequest mismatchedRequest = commitLog.checkMessageAndReturnSize(
+ mismatchedEntry, true, false, false);
+
+ Assert.assertFalse(crossingRequest.isSuccess());
+ Assert.assertFalse(mismatchedRequest.isSuccess());
+ Assert.assertArrayEquals(new int[] {0, 0},
+ new int[] {crossingEntry.position(), mismatchedEntry.position()});
+ } finally {
+ shutdownAndDestroy(messageStore);
+ }
+ }
+
+ @Test
+ public void testRaftLogReadNoopDoesNotBuildConsumeQueue() throws Exception {
+ String peers = String.format("n0-localhost:%d", nextPort());
+ DefaultMessageStore messageStore = null;
+ try {
+ messageStore = createDledgerMessageStore(
+ createBaseDir(), UUID.randomUUID().toString(), "n0", peers, null, false, 0);
+ awaitLeader(Arrays.asList(messageStore));
+ Assert.assertTrue(messageStore.getConsumeQueueTable().isEmpty());
+ DLedgerServer server = commitLog(messageStore).getdLedgerServer();
+ long previousCommittedIndex = committedIndex(messageStore);
+ long previousLedgerEndIndex = server.getDLedgerStore().getLedgerEndIndex();
+
+ Status status = appendRaftLogNoop(messageStore);
+ Assert.assertTrue(status.isOk());
+ awaitCommittedPast(messageStore, previousCommittedIndex);
+ Assert.assertEquals(previousLedgerEndIndex + 1,
+ server.getDLedgerStore().getLedgerEndIndex());
+ assertNoopEntry(messageStore, previousLedgerEndIndex + 1);
+ awaitNoopConsumed(messageStore);
+
+ Assert.assertTrue(messageStore.getConsumeQueueTable().isEmpty());
+ Assert.assertTrue(commitLog(messageStore).getCommittedPos() > 0);
+ Assert.assertEquals(commitLog(messageStore).getCommittedPos(), messageStore.getCommitLog().getMaxOffset());
+ } finally {
+ shutdownAndDestroy(messageStore);
+ }
+ }
+
+ @Test
+ public void testAbnormalRecoveryAcrossLeadingNoop() throws Exception {
+ String base = createBaseDir();
+ String peers = String.format("n0-localhost:%d", nextPort());
+ String group = UUID.randomUUID().toString();
+ String topic = String.format("%s%s%s%s", UUID.randomUUID(), UUID.randomUUID(),
+ UUID.randomUUID(), UUID.randomUUID());
+ DefaultMessageStore currentStore = null;
+ try {
+ currentStore = createDledgerMessageStore(base, group, "n0", peers, null, false, 0);
+ awaitLeader(Arrays.asList(currentStore));
+ Assert.assertTrue(appendRaftLogNoop(currentStore).isOk());
+ awaitCommittedPast(currentStore, -1);
+ assertNoopEntry(currentStore, 0);
+ awaitNoopConsumed(currentStore);
+ Assert.assertTrue(currentStore.getConsumeQueueTable().isEmpty());
+
+ putSingle(currentStore, topic, 0);
+ awaitStoreReady(currentStore, topic, 1);
+ assertMessageMagic(currentStore, topic, QUEUE_ID,
+ MessageDecoder.MESSAGE_MAGIC_CODE_V2);
+ doGetMessages(currentStore, topic, QUEUE_ID, 1, 0);
+ long committedIndexBeforeRestart = committedIndex(currentStore);
+ long committedPosBeforeRestart = commitLog(currentStore).getCommittedPos();
+
+ currentStore.shutdown();
+ currentStore = createDledgerMessageStore(base, group, "n0", peers, null, true, 0);
+ awaitLeader(Arrays.asList(currentStore));
+ awaitCommittedPast(currentStore, committedIndexBeforeRestart);
+ assertNoopEntry(currentStore, 0);
+ assertNoopEntry(currentStore, committedIndexBeforeRestart + 1);
+ awaitStoreReady(currentStore, topic, 1);
+ Assert.assertEquals(1, currentStore.getConsumeQueueTable().size());
+ Assert.assertTrue(currentStore.getConsumeQueueTable().containsKey(topic));
+ Assert.assertTrue(commitLog(currentStore).getCommittedPos() >= committedPosBeforeRestart);
+ doGetMessages(currentStore, topic, QUEUE_ID, 1, 0);
+
+ putSingle(currentStore, topic, 1);
+ awaitStoreReady(currentStore, topic, 2);
+ doGetMessages(currentStore, topic, QUEUE_ID, 2, 0);
+ } finally {
+ shutdownAndDestroy(currentStore);
+ }
+ }
+
+ @Test
+ public void testFixedSizeReadsRespectCommittedBoundary() throws Exception {
+ String peers = String.format("n0-localhost:%d;n1-localhost:%d", nextPort(), nextPort());
+ String group = UUID.randomUUID().toString();
+ DefaultMessageStore leaderStore = null;
+ DefaultMessageStore followerStore = null;
+ try {
+ leaderStore = createDledgerMessageStore(
+ createBaseDir(), group, "n0", peers, "n0", false, 0);
+ followerStore = createDledgerMessageStore(
+ createBaseDir(), group, "n1", peers, "n0", false, 0);
+ String topic = UUID.randomUUID().toString();
+ DLedgerCommitLog leaderCommitLog = commitLog(leaderStore);
+ DLedgerMmapFileStore dLedgerStore =
+ (DLedgerMmapFileStore) leaderCommitLog.getdLedgerServer().getdLedgerStore();
+ MmapFileList dataFileList = dLedgerStore.getDataFileList();
+
+ int messageCount = 0;
+ while (dataFileList.getMappedFiles().size() < 2 && messageCount < 16) {
+ MessageExtBrokerInner message = buildMessage();
+ message.setTopic(topic);
+ message.setQueueId(QUEUE_ID);
+ message.setBody(new byte[16 * 1024]);
+ PutMessageResult result = leaderStore.asyncPutMessage(message).get(5, SECONDS);
+ Assert.assertEquals(PutMessageStatus.PUT_OK, result.getPutMessageStatus());
+ messageCount++;
+ }
+ Assert.assertEquals(2, dataFileList.getMappedFiles().size());
+ awaitStoreReady(leaderStore, topic, messageCount);
+ awaitStoreReady(followerStore, topic, messageCount);
+
+ long selectedBase = dataFileList.getMappedFiles().get(1).getFileFromOffset();
+ long committedPos = leaderCommitLog.getCommittedPos();
+ Assert.assertTrue(committedPos > selectedBase);
+
+ followerStore.shutdown();
+ MessageExtBrokerInner uncommittedMessage = buildMessage();
+ uncommittedMessage.setTopic(topic);
+ uncommittedMessage.setQueueId(QUEUE_ID);
+ PutMessageResult uncommittedResult = leaderStore.asyncPutMessage(uncommittedMessage).get(5, SECONDS);
+ Assert.assertEquals(PutMessageStatus.IN_SYNC_REPLICAS_NOT_ENOUGH,
+ uncommittedResult.getPutMessageStatus());
+ Assert.assertEquals(committedPos, leaderCommitLog.getCommittedPos());
+ Assert.assertTrue(dataFileList.getMaxWrotePosition() > committedPos);
+ Assert.assertEquals(2, dataFileList.getMappedFiles().size());
+
+ ByteBuffer oversizedDestination = ByteBuffer.allocate(64);
+ Assert.assertTrue(leaderCommitLog.getData(committedPos - 1, 1, oversizedDestination));
+ Assert.assertEquals(1, oversizedDestination.position());
+ Assert.assertEquals(64, oversizedDestination.limit());
+
+ Assert.assertEquals(1, dataFileList.deleteExpiredFileByTime(0, 0, 0, true));
+ Assert.assertEquals(1, dataFileList.getMappedFiles().size());
+ long firstSurvivingBase = dataFileList.getFirstMappedFile().getFileFromOffset();
+ Assert.assertEquals(selectedBase, firstSurvivingBase);
+ Assert.assertTrue(firstSurvivingBase > 0);
+ Assert.assertTrue(firstSurvivingBase < committedPos);
+ Assert.assertSame(dataFileList.getFirstMappedFile(),
+ dataFileList.findMappedFileByOffset(0, true));
+
+ int crossSize = (int) (committedPos - firstSurvivingBase + 1);
+ Assert.assertTrue(firstSurvivingBase + crossSize <= dataFileList.getMaxWrotePosition());
+ ByteBuffer crossBoundaryDestination = ByteBuffer.allocate(crossSize);
+ Assert.assertFalse(leaderCommitLog.getData(0, crossSize, crossBoundaryDestination));
+ Assert.assertEquals(0, crossBoundaryDestination.position());
+ Assert.assertNull(leaderCommitLog.getMessage(0, crossSize));
+ } finally {
+ shutdownAndDestroy(followerStore);
+ shutdownAndDestroy(leaderStore);
+ }
+ }
+
+ private PutMessageResult putSingle(DefaultMessageStore messageStore, String topic, long expectedLogicOffset)
+ throws Exception {
+ MessageExtBrokerInner message = buildMessage();
+ message.setTopic(topic);
+ message.setQueueId(QUEUE_ID);
+ PutMessageResult result = messageStore.asyncPutMessage(message).get(5, SECONDS);
+ Assert.assertEquals(PutMessageStatus.PUT_OK, result.getPutMessageStatus());
+ Assert.assertNotNull(result.getAppendMessageResult());
+ Assert.assertEquals(expectedLogicOffset, result.getAppendMessageResult().getLogicsOffset());
+ return result;
+ }
+
+ private PutMessageResult putBatch(DefaultMessageStore messageStore, String topic, int batchSize,
+ long expectedLogicOffset) throws Exception {
+ MessageExtBatch batch = buildBatchMessage(batchSize);
+ batch.setTopic(topic);
+ batch.setQueueId(QUEUE_ID);
+ PutMessageResult result = messageStore.asyncPutMessages(batch).get(5, SECONDS);
+ Assert.assertEquals(PutMessageStatus.PUT_OK, result.getPutMessageStatus());
+ Assert.assertNotNull(result.getAppendMessageResult());
+ Assert.assertEquals(expectedLogicOffset, result.getAppendMessageResult().getLogicsOffset());
+ return result;
+ }
+
+ private DefaultMessageStore awaitLeader(List<DefaultMessageStore> stores) {
+ AtomicReference<DefaultMessageStore> leaderRef = new AtomicReference<>();
+ await().atMost(15, SECONDS).pollInterval(100, MILLISECONDS).until(() -> {
+ DefaultMessageStore leader = null;
+ for (DefaultMessageStore store : stores) {
+ if (commitLog(store).getdLedgerServer().getMemberState().isLeader()) {
+ if (leader != null) {
+ return false;
+ }
+ leader = store;
+ }
+ }
+ leaderRef.set(leader);
+ return leader != null;
+ });
+ return leaderRef.get();
+ }
+
+ private void awaitStoreReady(DefaultMessageStore messageStore, String topic, long expectedMaxOffset) {
+ await().atMost(15, SECONDS).pollInterval(100, MILLISECONDS).untilAsserted(() -> {
+ Assert.assertEquals(expectedMaxOffset, messageStore.getMaxOffsetInQueue(topic, QUEUE_ID));
+ Assert.assertEquals(0, messageStore.dispatchBehindBytes());
+ });
+ }
+
+ private void awaitCommittedPast(DefaultMessageStore messageStore, long previousCommittedIndex) {
+ DLedgerServer server = commitLog(messageStore).getdLedgerServer();
+ await().atMost(15, SECONDS).pollInterval(100, MILLISECONDS).until(() -> {
+ long committedIndex = server.getMemberState().getCommittedIndex();
+ return committedIndex > previousCommittedIndex
+ && committedIndex == server.getDLedgerStore().getLedgerEndIndex();
+ });
+ }
+
+ private void awaitNoopConsumed(DefaultMessageStore messageStore) {
+ await().atMost(15, SECONDS).pollInterval(100, MILLISECONDS).untilAsserted(() -> {
+ Assert.assertEquals(commitLog(messageStore).getCommittedPos(), messageStore.getCommitLog().getMaxOffset());
+ Assert.assertEquals(0, messageStore.dispatchBehindBytes());
+ });
+ }
+
+ private List<byte[]> readMessageBodies(DefaultMessageStore messageStore, String topic, int queueId,
+ int messageCount) {
+ List<byte[]> bodies = new ArrayList<>(messageCount);
+ for (int i = 0; i < messageCount; i++) {
+ GetMessageResult result = messageStore.getMessage("group", topic, queueId, i, 1, null);
+ Assert.assertNotNull(result);
+ try {
+ Assert.assertFalse(result.getMessageBufferList().isEmpty());
+ MessageExt message = MessageDecoder.decode(result.getMessageBufferList().get(0));
+ Assert.assertNotNull(message);
+ Assert.assertEquals(i, message.getQueueOffset());
+ bodies.add(Arrays.copyOf(message.getBody(), message.getBody().length));
+ } finally {
+ result.release();
+ }
+ }
+ return bodies;
+ }
+
+ private void assertMessageBodies(DefaultMessageStore messageStore, String topic, int queueId,
+ List<byte[]> expectedBodies) {
+ List<byte[]> actualBodies = readMessageBodies(messageStore, topic, queueId, expectedBodies.size());
+ for (int i = 0; i < expectedBodies.size(); i++) {
+ Assert.assertArrayEquals(expectedBodies.get(i), actualBodies.get(i));
+ }
+ }
+
+ private void assertMessageMagic(DefaultMessageStore messageStore, String topic, int queueId,
+ int expectedMagic) {
+ GetMessageResult result = messageStore.getMessage("group", topic, queueId, 0, 1, null);
+ Assert.assertNotNull(result);
+ try {
+ Assert.assertFalse(result.getMessageBufferList().isEmpty());
+ ByteBuffer messageBuffer = result.getMessageBufferList().get(0).duplicate();
+ Assert.assertEquals(expectedMagic,
+ messageBuffer.getInt(messageBuffer.position() + MessageDecoder.MESSAGE_MAGIC_CODE_POSITION));
+ } finally {
+ result.release();
+ }
+ }
+
+ private Status appendRaftLogNoop(DefaultMessageStore messageStore) throws Exception {
+ CompletableFuture<Status> result = new CompletableFuture<>();
+ commitLog(messageStore).getdLedgerServer().handleRead(ReadMode.RAFT_LOG_READ, new ReadClosure() {
+ @Override
+ public void done(Status status) {
+ result.complete(status);
+ }
+ });
+ return result.get(5, SECONDS);
+ }
+
+ private ByteBuffer noopHeader(int entrySize) {
+ ByteBuffer buffer = ByteBuffer.allocate(DLedgerEntry.BODY_OFFSET);
+ buffer.putInt(DLedgerEntryType.NOOP.getMagic());
+ buffer.putInt(entrySize);
+ buffer.position(0);
+ buffer.limit(DLedgerEntry.BODY_OFFSET);
+ return buffer;
+ }
+
+ private ByteBuffer normalEntryBuffer(byte[] innerMessage, int bodyPadding, byte[] trailingBytes) {
+ int entrySize = DLedgerEntry.BODY_OFFSET + innerMessage.length + bodyPadding;
+ int trailingSize = trailingBytes == null ? 0 : trailingBytes.length;
+ ByteBuffer buffer = ByteBuffer.allocate(entrySize + trailingSize);
+ buffer.putInt(DLedgerEntryType.NORMAL.getMagic());
+ buffer.putInt(entrySize);
+ buffer.position(DLedgerEntry.BODY_OFFSET);
+ buffer.put(innerMessage);
+ buffer.position(entrySize);
+ if (trailingBytes != null) {
+ buffer.put(trailingBytes);
+ }
+ buffer.flip();
+ return buffer;
+ }
+
+ private void assertNoopEntry(DefaultMessageStore messageStore, long index) {
+ DLedgerServer server = commitLog(messageStore).getdLedgerServer();
+ DLedgerEntry entry = server.getDLedgerStore().get(index);
+ Assert.assertNotNull(entry);
+ Assert.assertEquals(DLedgerEntryType.NOOP.getMagic(), entry.getMagic());
+ }
+
+ private long committedIndex(DefaultMessageStore messageStore) {
+ return commitLog(messageStore).getdLedgerServer().getMemberState().getCommittedIndex();
+ }
+
+ private DLedgerCommitLog commitLog(DefaultMessageStore messageStore) {
+ return (DLedgerCommitLog) messageStore.getCommitLog();
+ }
+
+ private void shutdownAndDestroy(DefaultMessageStore messageStore) {
+ if (messageStore == null) {
+ return;
+ }
+ try {
+ if (!messageStore.isShutdown()) {
+ messageStore.shutdown();
+ }
+ } finally {
+ messageStore.destroy();
+ }
+ }
+}