[ISSUE #690] Support batch msgs in dledger mode (#2406)

* implement issue-690

* add unit test

* fix version

* fix wroteOffset;update version;polish

* polish

* fix wrong wroteOffset of AppendMessageResult

* move serialization out of lock in async method

store/pom.xml

@@ -31,7 +31,7 @@
         <dependency>
             <groupId>io.openmessaging.storage</groupId>
             <artifactId>dledger</artifactId>
-            <version>0.2.0</version>
+            <version>0.2.2</version>
             <exclusions>
                 <exclusion>
                     <groupId>org.apache.rocketmq</groupId>

store/src/test/java/org/apache/rocketmq/store/StoreTestBase.java

@@ -16,17 +16,19 @@
  */
 package org.apache.rocketmq.store;
 
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExtBatch;
+import org.junit.After;
+
 import java.io.File;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.net.UnknownHostException;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.rocketmq.common.UtilAll;
-import org.junit.After;
 
 public class StoreTestBase {
 
@@ -44,6 +46,28 @@ public class StoreTestBase {
         return port.addAndGet(5);
     }
 
+    protected MessageExtBatch buildBatchMessage(int size) {
+        MessageExtBatch messageExtBatch = new MessageExtBatch();
+        messageExtBatch.setTopic("StoreTest");
+        messageExtBatch.setTags("TAG1");
+        messageExtBatch.setKeys("Hello");
+        messageExtBatch.setQueueId(Math.abs(QueueId.getAndIncrement()) % QUEUE_TOTAL);
+        messageExtBatch.setSysFlag(0);
+
+        messageExtBatch.setBornTimestamp(System.currentTimeMillis());
+        messageExtBatch.setBornHost(BornHost);
+        messageExtBatch.setStoreHost(StoreHost);
+
+        List<Message> messageList = new ArrayList<>(size);
+        for (int i = 0; i < size; i++) {
+            messageList.add(buildMessage());
+        }
+
+        messageExtBatch.setBody(MessageDecoder.encodeMessages(messageList));
+
+        return messageExtBatch;
+    }
+
     protected MessageExtBrokerInner buildMessage() {
         MessageExtBrokerInner msg = new MessageExtBrokerInner();
         msg.setTopic("StoreTest");
@@ -59,6 +83,40 @@ public class StoreTestBase {
         return msg;
     }
 
+    protected MessageExtBatch buildIPv6HostBatchMessage(int size) {
+        MessageExtBatch messageExtBatch = new MessageExtBatch();
+        messageExtBatch.setTopic("StoreTest");
+        messageExtBatch.setTags("TAG1");
+        messageExtBatch.setKeys("Hello");
+        messageExtBatch.setBody(MessageBody);
+        messageExtBatch.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        messageExtBatch.setKeys(String.valueOf(System.currentTimeMillis()));
+        messageExtBatch.setQueueId(Math.abs(QueueId.getAndIncrement()) % QUEUE_TOTAL);
+        messageExtBatch.setSysFlag(0);
+        messageExtBatch.setBornHostV6Flag();
+        messageExtBatch.setStoreHostAddressV6Flag();
+        messageExtBatch.setBornTimestamp(System.currentTimeMillis());
+        try {
+            messageExtBatch.setBornHost(new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 8123));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+        }
+
+        try {
+            messageExtBatch.setStoreHost(new InetSocketAddress(InetAddress.getByName("::1"), 8123));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+        }
+
+        List<Message> messageList = new ArrayList<>(size);
+        for (int i = 0; i < size; i++) {
+            messageList.add(buildIPv6HostMessage());
+        }
+
+        messageExtBatch.setBody(MessageDecoder.encodeMessages(messageList));
+        return messageExtBatch;
+    }
+
     protected MessageExtBrokerInner buildIPv6HostMessage() {
         MessageExtBrokerInner msg = new MessageExtBrokerInner();
         msg.setTopic("StoreTest");

store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest.java

@@ -19,14 +19,17 @@ package org.apache.rocketmq.store.dledger;
 import io.openmessaging.storage.dledger.DLedgerServer;
 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.List;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+
 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.store.DefaultMessageStore;
 import org.apache.rocketmq.store.GetMessageResult;
 import org.apache.rocketmq.store.GetMessageStatus;
@@ -94,7 +97,6 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
     }
 
 
-
     @Test
     public void testRecover() throws Exception {
         String base = createBaseDir();
@@ -135,7 +137,6 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
     }
 
 
-
     @Test
     public void testPutAndGetMessage() throws Exception {
         String base = createBaseDir();
@@ -177,6 +178,50 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
         messageStore.shutdown();
     }
 
+    @Test
+    public void testBatchPutAndGetMessage() throws Exception {
+        String base = createBaseDir();
+        String peers = String.format("n0-localhost:%d", nextPort());
+        String group = UUID.randomUUID().toString();
+        DefaultMessageStore messageStore = createDledgerMessageStore(base, group, "n0", peers, null, false, 0);
+        Thread.sleep(1000);
+        String topic = UUID.randomUUID().toString();
+        // should be less than 4
+        int batchMessageSize = 2;
+        int repeat = 10;
+        List<PutMessageResult> results = new ArrayList<>();
+        for (int i = 0; i < repeat; i++) {
+            MessageExtBatch messageExtBatch =
+                    i < repeat / 10 ? buildBatchMessage(batchMessageSize) : buildIPv6HostBatchMessage(batchMessageSize);
+            messageExtBatch.setTopic(topic);
+            messageExtBatch.setQueueId(0);
+            PutMessageResult putMessageResult = messageStore.putMessages(messageExtBatch);
+            results.add(putMessageResult);
+            Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
+            Assert.assertEquals(i * batchMessageSize, putMessageResult.getAppendMessageResult().getLogicsOffset());
+        }
+        Thread.sleep(100);
+        Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, 0));
+        Assert.assertEquals(repeat * batchMessageSize, messageStore.getMaxOffsetInQueue(topic, 0));
+        Assert.assertEquals(0, messageStore.dispatchBehindBytes());
+        GetMessageResult getMessageResult = messageStore.getMessage("group", topic, 0, 0, 100, null);
+        Assert.assertEquals(GetMessageStatus.FOUND, getMessageResult.getStatus());
+
+        Assert.assertEquals(repeat * batchMessageSize > 32 ? 32 : repeat * batchMessageSize, getMessageResult.getMessageBufferList().size());
+        Assert.assertEquals(repeat * batchMessageSize > 32 ? 32 : repeat * batchMessageSize, getMessageResult.getMessageMapedList().size());
+        Assert.assertEquals(repeat * batchMessageSize, getMessageResult.getMaxOffset());
+
+        for (int i = 0; i < results.size(); i++) {
+            ByteBuffer buffer = getMessageResult.getMessageBufferList().get(i * batchMessageSize);
+            MessageExt messageExt = MessageDecoder.decode(buffer);
+            Assert.assertEquals(i * batchMessageSize, messageExt.getQueueOffset());
+            Assert.assertEquals(results.get(i).getAppendMessageResult().getMsgId().split(",").length, batchMessageSize);
+            Assert.assertEquals(results.get(i).getAppendMessageResult().getWroteOffset(), messageExt.getCommitLogOffset());
+        }
+        messageStore.destroy();
+        messageStore.shutdown();
+    }
+
     @Test
     public void testAsyncPutAndGetMessage() throws Exception {
         String base = createBaseDir();
@@ -219,12 +264,57 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
         messageStore.shutdown();
     }
 
+    @Test
+    public void testAsyncBatchPutAndGetMessage() throws Exception {
+        String base = createBaseDir();
+        String peers = String.format("n0-localhost:%d", nextPort());
+        String group = UUID.randomUUID().toString();
+        DefaultMessageStore messageStore = createDledgerMessageStore(base, group, "n0", peers, null, false, 0);
+        Thread.sleep(1000);
+        String topic = UUID.randomUUID().toString();
+        // should be less than 4
+        int batchMessageSize = 2;
+        int repeat = 10;
+
+        List<PutMessageResult> results = new ArrayList<>();
+        for (int i = 0; i < repeat; i++) {
+            MessageExtBatch messageExtBatch =
+                    i < 5 ? buildBatchMessage(batchMessageSize) : buildIPv6HostBatchMessage(batchMessageSize);
+            messageExtBatch.setTopic(topic);
+            messageExtBatch.setQueueId(0);
+            CompletableFuture<PutMessageResult> futureResult = messageStore.asyncPutMessages(messageExtBatch);
+            PutMessageResult putMessageResult = futureResult.get(3000, TimeUnit.MILLISECONDS);
+            results.add(putMessageResult);
+            Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
+            Assert.assertEquals(i * batchMessageSize, putMessageResult.getAppendMessageResult().getLogicsOffset());
+        }
+        Thread.sleep(100);
+        Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, 0));
+        Assert.assertEquals(repeat * batchMessageSize, messageStore.getMaxOffsetInQueue(topic, 0));
+        Assert.assertEquals(0, messageStore.dispatchBehindBytes());
+        GetMessageResult getMessageResult = messageStore.getMessage("group", topic, 0, 0, 32, null);
+        Assert.assertEquals(GetMessageStatus.FOUND, getMessageResult.getStatus());
+
+        Assert.assertEquals(repeat * batchMessageSize > 32 ? 32 : repeat * batchMessageSize, getMessageResult.getMessageBufferList().size());
+        Assert.assertEquals(repeat * batchMessageSize > 32 ? 32 : repeat * batchMessageSize, getMessageResult.getMessageMapedList().size());
+        Assert.assertEquals(repeat * batchMessageSize, getMessageResult.getMaxOffset());
+
+        for (int i = 0; i < results.size(); i++) {
+            ByteBuffer buffer = getMessageResult.getMessageBufferList().get(i * batchMessageSize);
+            MessageExt messageExt = MessageDecoder.decode(buffer);
+            Assert.assertEquals(i * batchMessageSize, messageExt.getQueueOffset());
+            Assert.assertEquals(results.get(i).getAppendMessageResult().getMsgId().split(",").length, batchMessageSize);
+            Assert.assertEquals(results.get(i).getAppendMessageResult().getWroteOffset(), messageExt.getCommitLogOffset());
+        }
+        messageStore.destroy();
+        messageStore.shutdown();
+    }
 
     @Test
     public void testCommittedPos() throws Exception {
         String peers = String.format("n0-localhost:%d;n1-localhost:%d", nextPort(), nextPort());
         String group = UUID.randomUUID().toString();
-        DefaultMessageStore leaderStore = createDledgerMessageStore(createBaseDir(), group,"n0", peers, "n0", false, 0);
+        DefaultMessageStore leaderStore = createDledgerMessageStore(createBaseDir(), group, "n0", peers, "n0", false, 0);
 
         String topic = UUID.randomUUID().toString();
         MessageExtBrokerInner msgInner = buildMessage();
@@ -239,7 +329,7 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
         Assert.assertEquals(0, leaderStore.getMaxOffsetInQueue(topic, 0));
 
 
-        DefaultMessageStore followerStore = createDledgerMessageStore(createBaseDir(), group,"n1", peers, "n0", false, 0);
+        DefaultMessageStore followerStore = createDledgerMessageStore(createBaseDir(), group, "n1", peers, "n0", false, 0);
         Thread.sleep(2000);
 
         Assert.assertEquals(1, leaderStore.getMaxOffsetInQueue(topic, 0));
@@ -258,7 +348,7 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
     public void testIPv6HostMsgCommittedPos() throws Exception {
         String peers = String.format("n0-localhost:%d;n1-localhost:%d", nextPort(), nextPort());
         String group = UUID.randomUUID().toString();
-        DefaultMessageStore leaderStore = createDledgerMessageStore(createBaseDir(), group,"n0", peers, "n0", false, 0);
+        DefaultMessageStore leaderStore = createDledgerMessageStore(createBaseDir(), group, "n0", peers, "n0", false, 0);
 
         String topic = UUID.randomUUID().toString();
         MessageExtBrokerInner msgInner = buildIPv6HostMessage();
@@ -273,7 +363,7 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
         Assert.assertEquals(0, leaderStore.getMaxOffsetInQueue(topic, 0));
 
 
-        DefaultMessageStore followerStore = createDledgerMessageStore(createBaseDir(), group,"n1", peers, "n0", false, 0);
+        DefaultMessageStore followerStore = createDledgerMessageStore(createBaseDir(), group, "n1", peers, "n0", false, 0);
         Thread.sleep(2000);
 
         Assert.assertEquals(1, leaderStore.getMaxOffsetInQueue(topic, 0));