[ISSUE #10651] Adapt RocketMQ to latest DLedger and remove fastjson1

WORKSPACE

@@ -40,8 +40,7 @@ load("@rules_jvm_external//:defs.bzl", "maven_install")
 maven_install(
     artifacts = [
         "junit:junit:4.13.2",
-        "com.alibaba:fastjson:1.2.83",
-        "com.alibaba.fastjson2:fastjson2:2.0.59",
+        "com.alibaba.fastjson2:fastjson2:2.0.64",
         "org.hamcrest:hamcrest-library:1.3",
         "io.netty:netty-all:4.1.130.Final",
         "org.assertj:assertj-core:3.22.0",
@@ -54,7 +53,7 @@ maven_install(
         "commons-validator:commons-validator:1.10.0",
         "org.apache.commons:commons-lang3:3.20.0",
         "org.hamcrest:hamcrest-core:1.3",
-        "io.openmessaging.storage:dledger:0.3.2",
+        "io.openmessaging.storage:dledger:0.3.3-pr336-f2-64-SNAPSHOT",
         "net.java.dev.jna:jna:4.2.2",
         "ch.qos.logback:logback-classic:1.2.10",
         "ch.qos.logback:logback-core:1.2.10",
@@ -117,6 +116,7 @@ maven_install(
         "org.slf4j:slf4j-api:2.0.3",
         "org.javassist:javassist:3.20.0-GA",
     ],
+    excluded_artifacts = ["org.apache.rocketmq:rocketmq-remoting"],
     fetch_sources = False,
     repositories = [
         "https://repo1.maven.org/maven2",

broker/src/main/java/org/apache/rocketmq/broker/dledger/DLedgerRoleChangeHandler.java

@@ -80,7 +80,7 @@ public void run() {
                                 if (dLegerServer.getDLedgerStore().getLedgerEndIndex() == -1) {
                                     break;
                                 }
-                                if (dLegerServer.getDLedgerStore().getLedgerEndIndex() == dLegerServer.getDLedgerStore().getCommittedIndex()
+                                if (dLegerServer.getDLedgerStore().getLedgerEndIndex() == dLegerServer.getMemberState().getCommittedIndex()
                                     && messageStore.dispatchBehindBytes() == 0) {
                                     break;
                                 }

common/pom.xml

@@ -32,10 +32,6 @@
     </properties>
 
     <dependencies>
-        <dependency>
-            <groupId>com.alibaba</groupId>
-            <artifactId>fastjson</artifactId>
-        </dependency>
         <dependency>
             <groupId>com.alibaba.fastjson2</groupId>
             <artifactId>fastjson2</artifactId>

controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java

@@ -17,11 +17,11 @@
 package org.apache.rocketmq.controller.impl;
 
 import com.google.common.base.Stopwatch;
-import io.openmessaging.storage.dledger.AppendFuture;
 import io.openmessaging.storage.dledger.DLedgerConfig;
 import io.openmessaging.storage.dledger.DLedgerLeaderElector;
 import io.openmessaging.storage.dledger.DLedgerServer;
 import io.openmessaging.storage.dledger.MemberState;
+import io.openmessaging.storage.dledger.common.AppendFuture;
 import io.openmessaging.storage.dledger.protocol.AppendEntryRequest;
 import io.openmessaging.storage.dledger.protocol.AppendEntryResponse;
 import io.openmessaging.storage.dledger.protocol.BatchAppendEntryRequest;

controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerControllerStateMachine.java

@@ -20,7 +20,8 @@
 import io.openmessaging.storage.dledger.exception.DLedgerException;
 import io.openmessaging.storage.dledger.snapshot.SnapshotReader;
 import io.openmessaging.storage.dledger.snapshot.SnapshotWriter;
-import io.openmessaging.storage.dledger.statemachine.CommittedEntryIterator;
+import io.openmessaging.storage.dledger.statemachine.ApplyEntry;
+import io.openmessaging.storage.dledger.statemachine.ApplyEntryIterator;
 import io.openmessaging.storage.dledger.statemachine.StateMachine;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.controller.impl.event.EventMessage;
@@ -51,12 +52,13 @@ public String generateDLedgerId(String dLedgerGroupId, String dLedgerSelfId) {
     }
 
     @Override
-    public void onApply(CommittedEntryIterator iterator) {
+    public void onApply(ApplyEntryIterator iterator) {
         int applyingSize = 0;
         long firstApplyIndex = -1;
         long lastApplyIndex = -1;
         while (iterator.hasNext()) {
-            final DLedgerEntry entry = iterator.next();
+            final ApplyEntry applyEntry = iterator.next();
+            final DLedgerEntry entry = applyEntry.getEntry();
             final byte[] body = entry.getBody();
             if (body != null && body.length > 0) {
                 final EventMessage event = this.eventSerializer.deserialize(body);

controller/src/test/java/org/apache/rocketmq/controller/impl/DLedgerControllerTest.java

@@ -40,6 +40,7 @@
 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.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 DLedgerController waitLeader(final List<DLedgerController> controllers) t
 
     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 void testElectMaster() throws Exception {
         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 void testBrokerLifecycleListener() throws Exception {
             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);

pom.xml

@@ -104,8 +104,7 @@
         <netty.version>4.1.130.Final</netty.version>
         <netty.tcnative.version>2.0.53.Final</netty.tcnative.version>
         <bcpkix-jdk18on.version>1.83</bcpkix-jdk18on.version>
-        <fastjson.version>1.2.83</fastjson.version>
-        <fastjson2.version>2.0.63</fastjson2.version>
+        <fastjson2.version>2.0.64</fastjson2.version>
         <javassist.version>3.20.0-GA</javassist.version>
         <jna.version>4.2.2</jna.version>
         <commons-lang3.version>3.20.0</commons-lang3.version>
@@ -123,7 +122,7 @@
         <lz4-java.version>1.10.3</lz4-java.version>
         <opentracing.version>0.33.0</opentracing.version>
         <jaeger.version>1.8.1</jaeger.version>
-        <dleger.version>0.3.2</dleger.version>
+        <dleger.version>0.3.3-pr336-f2-64-SNAPSHOT</dleger.version>
         <annotations-api.version>6.0.53</annotations-api.version>
         <extra-enforcer-rules.version>1.0-beta-4</extra-enforcer-rules.version>
         <concurrentlinkedhashmap-lru.version>1.4.2</concurrentlinkedhashmap-lru.version>
@@ -688,11 +687,6 @@
                 <type>jar</type>
                 <version>${bcpkix-jdk18on.version}</version>
             </dependency>
-            <dependency>
-                <groupId>com.alibaba</groupId>
-                <artifactId>fastjson</artifactId>
-                <version>${fastjson.version}</version>
-            </dependency>
             <dependency>
                 <groupId>com.alibaba.fastjson2</groupId>
                 <artifactId>fastjson2</artifactId>

remoting/BUILD.bazel

@@ -52,7 +52,6 @@ java_library(
         "//common",
         "//:test_deps",
         "@maven//:org_objenesis_objenesis",
-        "@maven//:com_alibaba_fastjson",
         "@maven//:com_alibaba_fastjson2_fastjson2",
         "@maven//:com_google_code_gson_gson",
         "@maven//:com_google_guava_guava",