[ISSUE #2217] Upgrade dledger version (#2218)

* feature(dledger):upgrade dledger version and support preferred leader/batch push

* feature(dledger):pass the failed UT and pass the CI

store/pom.xml

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

store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java

@@ -148,6 +148,10 @@ public class MessageStoreConfig {
     private String dLegerPeers;
     private String dLegerSelfId;
 
+    private String preferredLeaderId;
+
+    private boolean isEnableBatchPush = false;
+
     public boolean isDebugLockEnable() {
         return debugLockEnable;
     }
@@ -702,4 +706,20 @@ public class MessageStoreConfig {
     public void setEnableDLegerCommitLog(boolean enableDLegerCommitLog) {
         this.enableDLegerCommitLog = enableDLegerCommitLog;
     }
+
+    public String getPreferredLeaderId() {
+        return preferredLeaderId;
+    }
+
+    public void setPreferredLeaderId(String preferredLeaderId) {
+        this.preferredLeaderId = preferredLeaderId;
+    }
+
+    public boolean isEnableBatchPush() {
+        return isEnableBatchPush;
+    }
+
+    public void setEnableBatchPush(boolean enableBatchPush) {
+        isEnableBatchPush = enableBatchPush;
+    }
 }

store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java

@@ -84,6 +84,9 @@ public class DLedgerCommitLog extends CommitLog {
         dLedgerConfig.setMappedFileSizeForEntryData(defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog());
         dLedgerConfig.setDeleteWhen(defaultMessageStore.getMessageStoreConfig().getDeleteWhen());
         dLedgerConfig.setFileReservedHours(defaultMessageStore.getMessageStoreConfig().getFileReservedTime() + 1);
+        dLedgerConfig.setPreferredLeaderId(defaultMessageStore.getMessageStoreConfig().getPreferredLeaderId());
+        dLedgerConfig.setEnableBatchPush(defaultMessageStore.getMessageStoreConfig().isEnableBatchPush());
+
         id = Integer.valueOf(dLedgerConfig.getSelfId().substring(1)) + 1;
         dLedgerServer = new DLedgerServer(dLedgerConfig);
         dLedgerFileStore = (DLedgerMmapFileStore) dLedgerServer.getdLedgerStore();

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

@@ -63,9 +63,9 @@ public class MessageStoreTestBase extends StoreTestBase {
         if (leaderId != null) {
             dLegerServer.getdLedgerConfig().setEnableLeaderElector(false);
             if (selfId.equals(leaderId)) {
-                dLegerServer.getMemberState().changeToLeader(-1);
+                dLegerServer.getMemberState().changeToLeader(0);
             } else {
-                dLegerServer.getMemberState().changeToFollower(-1, leaderId);
+                dLegerServer.getMemberState().changeToFollower(0, leaderId);
             }
 
         }