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 f644676f64d3d9a5620f63d7e6ee7e7155f7eba3 Author: 通融 <[email protected]> AuthorDate: Sun Aug 16 09:21:11 2026 +0800 fix: handle single-message DLedger batches --- .../2026-08-14-dledger-latest-pr336-fastjson2.md | 40 ++++++++++++++++------ .../rocketmq/store/dledger/DLedgerCommitLog.java | 22 ++++++++++-- .../store/dledger/DLedgerLatestCommitLogTest.java | 15 +++++--- 3 files changed, 59 insertions(+), 18 deletions(-) diff --git a/docs/superpowers/plans/2026-08-14-dledger-latest-pr336-fastjson2.md b/docs/superpowers/plans/2026-08-14-dledger-latest-pr336-fastjson2.md index 79a4878af3..6863e7de45 100644 --- a/docs/superpowers/plans/2026-08-14-dledger-latest-pr336-fastjson2.md +++ b/docs/superpowers/plans/2026-08-14-dledger-latest-pr336-fastjson2.md @@ -992,16 +992,25 @@ Also call the fixed-size `getData` overload and assert `false`. This test is the - [ ] **Step 3: Test single and batch appends on the latest API** -Start one electing node, wait for leadership, write one `MessageExtBrokerInner`, then a three-message `MessageExtBatch`. Assert both `PutMessageStatus.PUT_OK`, batch `AppendMessageResult.getMsgNum() == 3`, four logical messages in queue 0, strictly increasing physical offsets, non-null message IDs, and successful reads of queue offsets 0 through 3. +Start one electing node, wait for leadership, write one `MessageExtBrokerInner`, then a one-message `MessageExtBatch`, followed by a three-message `MessageExtBatch`. Assert all three `PutMessageStatus.PUT_OK`, batch `AppendMessageResult.getMsgNum()` values of 1 and 3, five logical messages in queue 0, strictly increasing physical offsets, non-null message IDs, and successful reads of queue offsets 0 through 4. The exact batch setup is: ```java +MessageExtBatch oneMessageBatch = buildBatchMessage(1); +oneMessageBatch.setTopic(topic); +oneMessageBatch.setQueueId(0); +PutMessageResult oneMessageBatchResult = store.putMessages(oneMessageBatch); +Assert.assertEquals(PutMessageStatus.PUT_OK, oneMessageBatchResult.getPutMessageStatus()); +Assert.assertEquals(1, oneMessageBatchResult.getAppendMessageResult().getLogicsOffset()); +Assert.assertEquals(1, oneMessageBatchResult.getAppendMessageResult().getMsgNum()); + MessageExtBatch batch = buildBatchMessage(3); batch.setTopic(topic); batch.setQueueId(0); PutMessageResult batchResult = store.putMessages(batch); Assert.assertEquals(PutMessageStatus.PUT_OK, batchResult.getPutMessageStatus()); +Assert.assertEquals(2, batchResult.getAppendMessageResult().getLogicsOffset()); Assert.assertEquals(3, batchResult.getAppendMessageResult().getMsgNum()); ``` @@ -1018,7 +1027,7 @@ Allocate three DLedger ports and construct one peers string. Start `n0`, `n1`, a Use `try/finally`; shut down every started store and let the inherited cleanup destroy only the test-created base directories. -The complete final test file for Steps 1–4 and Task 7 is below. Its independently compiled Corretto 8 prototype passed all seven methods after the production fix; this version additionally asserts that malformed NOOP parsing leaves the buffer position unchanged. Use the file exactly, then follow the RED/GREEN ordering in the surrounding steps rather than running all seven methods prematurely: +The complete final test file for Steps 1–4 and Task 7 is below. Its independently compiled Corretto 8 prototype passed all eight methods after the production fix; this version additionally asserts that malformed NOOP parsing leaves the buffer position unchanged. Use the file exactly, then follow the RED/GREEN ordering in the surrounding steps rather than running all eight methods prematurely: ```java /* @@ -1112,18 +1121,27 @@ public class DLedgerLatestCommitLogTest extends MessageStoreTestBase { String topic = UUID.randomUUID().toString(); PutMessageResult singleResult = putSingle(messageStore, topic, 0); - PutMessageResult batchResult = putBatch(messageStore, topic, 3, 1); + PutMessageResult singleMessageBatchResult = putBatch(messageStore, topic, 1, 1); + PutMessageResult batchResult = putBatch(messageStore, topic, 3, 2); Assert.assertTrue(singleResult.getAppendMessageResult().getWroteOffset() > 0); - Assert.assertTrue(batchResult.getAppendMessageResult().getWroteOffset() + Assert.assertTrue(singleMessageBatchResult.getAppendMessageResult().getWroteOffset() > singleResult.getAppendMessageResult().getWroteOffset()); + Assert.assertTrue(batchResult.getAppendMessageResult().getWroteOffset() + > singleMessageBatchResult.getAppendMessageResult().getWroteOffset()); + Assert.assertEquals(1, singleMessageBatchResult.getAppendMessageResult().getMsgNum()); Assert.assertEquals(3, batchResult.getAppendMessageResult().getMsgNum()); + Assert.assertNotNull(singleResult.getAppendMessageResult().getMsgId()); + Assert.assertNotNull(singleMessageBatchResult.getAppendMessageResult().getMsgId()); + Assert.assertNotNull(batchResult.getAppendMessageResult().getMsgId()); + Assert.assertEquals(1, + singleMessageBatchResult.getAppendMessageResult().getMsgId().split(",").length); Assert.assertEquals(3, batchResult.getAppendMessageResult().getMsgId().split(",").length); - awaitStoreReady(messageStore, topic, 4); + awaitStoreReady(messageStore, topic, 5); Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, QUEUE_ID)); Assert.assertTrue(commitLog(messageStore).getCommittedPos() > batchResult.getAppendMessageResult().getWroteOffset()); - doGetMessages(messageStore, topic, QUEUE_ID, 4, 0); + doGetMessages(messageStore, topic, QUEUE_ID, 5, 0); } finally { shutdownAndDestroy(messageStore); } @@ -1465,7 +1483,7 @@ cd '/Users/jinrongtong/.codex/worktrees/9e8c/RocketMQ 开源' '-Dtest=DLedgerLatestCommitLogTest#testSingleAndBatchAppendPositions' test ``` -Expected: the master artifact fails with an immediate-position symptom such as `OS_PAGE_CACHE_BUSY` or an unknown append result. Then run only the three pre-restart methods with the PR artifact; the four NOOP/restart methods are intentionally left for their RED gates in Task 7: +Expected: the master artifact fails with an immediate-position symptom such as `OS_PAGE_CACHE_BUSY` or an unknown append result. Then run only the three pre-restart methods with the PR artifact; the five remaining restart, NOOP, recovery, and committed-boundary methods—for eight methods total—are intentionally left for their RED gates in Task 7: ```bash set -euo pipefail @@ -2001,7 +2019,7 @@ MARKER="$LOGS/rocketmq-store-full-green.start" touch "$MARKER" $RUN "$LOGS/rocketmq-store-full-green.log" $JDK8 mvn "${MVN_ARGS[@]}" \ -pl store -am -Dtest=DLedgerLatestCommitLogTest test -$ASSERT "$STORE_REPORT" 7 "$MARKER" +$ASSERT "$STORE_REPORT" 8 "$MARKER" MARKER="$LOGS/rocketmq-controller-restart-after-store-green.start" touch "$MARKER" $RUN "$LOGS/rocketmq-controller-restart-after-store-green.log" $JDK8 mvn "${MVN_ARGS[@]}" \ @@ -2015,7 +2033,7 @@ $RUN "$LOGS/rocketmq-controller-full-green.log" $JDK8 mvn "${MVN_ARGS[@]}" \ $ASSERT "$CONTROLLER_REPORT" 6 "$MARKER" ``` -Expected: store leadership converges, committed index equals ledger end, NOOP advances the physical/reput boundary without changing CQ offsets, old messages are readable before a user write, all seven store methods pass including abnormal recovery and the second restart, controller metadata remains available before a new mutation, and all six controller methods pass after controller replay/leader change. +Expected: store leadership converges, committed index equals ledger end, NOOP advances the physical/reput boundary without changing CQ offsets, old messages are readable before a user write, all eight store methods pass including abnormal recovery and the second restart, controller metadata remains available before a new mutation, and all six controller methods pass after controller replay/leader change. - [ ] **Step 8: Commit the restart/failover behavior** @@ -2965,7 +2983,7 @@ $ASSERT remoting/target/surefire-reports/TEST-org.apache.rocketmq.remoting.proto MARKER="$LOGS/rocketmq-maven-store.start" touch "$MARKER" /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/run-logged /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/logs/rocketmq-maven-store.log /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/run-jdk8 mvn -B -ntp -nsu -Dmaven.repo.local=/private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/m2 -Djava.io.tmpdir=/private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/tmp -pl store -am -Dsurefire.failIfNoSpecifiedTests=false -Dtest=D [...] -$ASSERT store/target/surefire-reports/TEST-org.apache.rocketmq.store.dledger.DLedgerLatestCommitLogTest.xml 7 "$MARKER" +$ASSERT store/target/surefire-reports/TEST-org.apache.rocketmq.store.dledger.DLedgerLatestCommitLogTest.xml 8 "$MARKER" MARKER="$LOGS/rocketmq-maven-controller.start" touch "$MARKER" /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/run-logged /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/logs/rocketmq-maven-controller.log /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/run-jdk8 mvn -B -ntp -nsu -Dmaven.repo.local=/private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/m2 -Djava.io.tmpdir=/private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/tmp -pl controller -am -Dsurefire.failIfNoSpecifiedTests=fals [...] @@ -3177,7 +3195,7 @@ rg -F 'OK (5 tests)' /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-valida $RUN_LOGGED /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/logs/rocketmq-bazel-store.log $BAZEL --output_user_root="$ROOT" test --repository_cache="$CACHE" \ --java_runtime_version=8 --nocache_test_results --test_output=all --local_test_jobs=1 \ '//store:src/test/java/org/apache/rocketmq/store/dledger/DLedgerLatestCommitLogTest' -rg -F 'OK (7 tests)' /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/logs/rocketmq-bazel-store.log +rg -F 'OK (8 tests)' /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/logs/rocketmq-bazel-store.log $RUN_LOGGED /private/tmp/rocketmq-dledger-fastjson2/pr336-latest-validation/logs/rocketmq-bazel-controller.log $BAZEL --output_user_root="$ROOT" test --repository_cache="$CACHE" \ --java_runtime_version=8 --nocache_test_results --test_output=all --local_test_jobs=1 \ 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 58f65579e3..142aab7af8 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 @@ -808,7 +808,7 @@ public class DLedgerCommitLog extends CommitLog { // Back to Results AppendMessageResult appendResult; - BatchAppendFuture<AppendEntryResponse> dledgerFuture; + AppendFuture<AppendEntryResponse> dledgerFuture; EncodeResult encodeResult; encodeResult = this.messageSerializer.serialize(messageExtBatch); @@ -840,7 +840,23 @@ public class DLedgerCommitLog extends CommitLog { log.warn("HandleAppend return false due to error code {}", appendFuture.get().getCode()); return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR))); } - dledgerFuture = (BatchAppendFuture<AppendEntryResponse>) appendFuture; + dledgerFuture = appendFuture; + + long[] positions; + if (batchNum == 1) { + positions = new long[] {appendFuture.getPos()}; + } else { + if (!(appendFuture instanceof BatchAppendFuture)) { + throw new IllegalStateException("Unexpected append future type for " + batchNum + + "-message batch: " + appendFuture.getClass().getName()); + } + positions = ((BatchAppendFuture<AppendEntryResponse>) appendFuture).getPositions(); + if (positions == null || positions.length != batchNum + || appendFuture.getPos() != positions[batchNum - 1]) { + throw new IllegalStateException("Inconsistent DLedger batch positions: expected " + batchNum + + " entries ending at " + appendFuture.getPos()); + } + } long wroteOffset = 0; @@ -849,7 +865,7 @@ public class DLedgerCommitLog extends CommitLog { boolean isFirstOffset = true; long firstWroteOffset = 0; - for (long pos : dledgerFuture.getPositions()) { + for (long pos : positions) { wroteOffset = pos + DLedgerEntry.BODY_OFFSET; if (isFirstOffset) { firstWroteOffset = wroteOffset; 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 index cf16b393a1..6f6cbe3303 100644 --- a/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerLatestCommitLogTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerLatestCommitLogTest.java @@ -93,20 +93,27 @@ public class DLedgerLatestCommitLogTest extends MessageStoreTestBase { String topic = UUID.randomUUID().toString(); PutMessageResult singleResult = putSingle(messageStore, topic, 0); - PutMessageResult batchResult = putBatch(messageStore, topic, 3, 1); + PutMessageResult singleMessageBatchResult = putBatch(messageStore, topic, 1, 1); + PutMessageResult batchResult = putBatch(messageStore, topic, 3, 2); Assert.assertTrue(singleResult.getAppendMessageResult().getWroteOffset() > 0); - Assert.assertTrue(batchResult.getAppendMessageResult().getWroteOffset() + Assert.assertTrue(singleMessageBatchResult.getAppendMessageResult().getWroteOffset() > singleResult.getAppendMessageResult().getWroteOffset()); + Assert.assertTrue(batchResult.getAppendMessageResult().getWroteOffset() + > singleMessageBatchResult.getAppendMessageResult().getWroteOffset()); + Assert.assertEquals(1, singleMessageBatchResult.getAppendMessageResult().getMsgNum()); Assert.assertEquals(3, batchResult.getAppendMessageResult().getMsgNum()); Assert.assertNotNull(singleResult.getAppendMessageResult().getMsgId()); + Assert.assertNotNull(singleMessageBatchResult.getAppendMessageResult().getMsgId()); Assert.assertNotNull(batchResult.getAppendMessageResult().getMsgId()); + Assert.assertEquals(1, + singleMessageBatchResult.getAppendMessageResult().getMsgId().split(",").length); Assert.assertEquals(3, batchResult.getAppendMessageResult().getMsgId().split(",").length); - awaitStoreReady(messageStore, topic, 4); + awaitStoreReady(messageStore, topic, 5); Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, QUEUE_ID)); Assert.assertTrue(commitLog(messageStore).getCommittedPos() > batchResult.getAppendMessageResult().getWroteOffset()); - doGetMessages(messageStore, topic, QUEUE_ID, 4, 0); + doGetMessages(messageStore, topic, QUEUE_ID, 5, 0); } finally { shutdownAndDestroy(messageStore); }
