[ISSUE #289] Support Register-Model server based on DLedger

.gitignore

@@ -3,13 +3,17 @@
 .project
 .settings/
 target/
-localbin/
+localbin/
 devenv
 *.log*
 *.iml
 .idea/
 *.versionsBackup
 !NOTICE-BIN
 !LICENSE-BIN
-.DS_Store
-nohup.out
+.DS_Store
+.clj-kondo
+.lsp
+*.jar
+*.tar.gz
+nohup.out

dledger/pom.xml

@@ -24,7 +24,7 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>dledger</artifactId>
+    <artifactId>dledger-core</artifactId>
 
     <dependencies>
         <dependency>

dledger/src/main/java/io/openmessaging/storage/dledger/BatchAppendFuture.java

@@ -22,6 +22,10 @@ public BatchAppendFuture() {
 
     }
 
+    public BatchAppendFuture(long[] positions) {
+        this.positions = positions;
+    }
+
     public BatchAppendFuture(long timeOutMs) {
         super(timeOutMs);
     }

dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerServer.java

@@ -94,12 +94,12 @@ public DLedgerServer(DLedgerConfig dLedgerConfig, NettyServerConfig nettyServerC
     }
 
     public DLedgerServer(DLedgerConfig dLedgerConfig, NettyServerConfig nettyServerConfig,
-                         NettyClientConfig nettyClientConfig) {
+        NettyClientConfig nettyClientConfig) {
         this(dLedgerConfig, nettyServerConfig, nettyClientConfig, null);
     }
 
     public DLedgerServer(DLedgerConfig dLedgerConfig, NettyServerConfig nettyServerConfig,
-                         NettyClientConfig nettyClientConfig, ChannelEventListener channelEventListener) {
+        NettyClientConfig nettyClientConfig, ChannelEventListener channelEventListener) {
         dLedgerConfig.init();
         this.dLedgerConfig = dLedgerConfig;
         this.memberState = new MemberState(dLedgerConfig);
@@ -266,52 +266,19 @@ public CompletableFuture<AppendEntryResponse> handleAppend(AppendEntryRequest re
                 appendEntryResponse.setTerm(currTerm);
                 appendEntryResponse.setLeaderId(memberState.getSelfId());
                 return AppendFuture.newCompletedFuture(-1, appendEntryResponse);
-            } else {
-                AppendFuture<AppendEntryResponse> future = new AppendFuture<>();
-                DLedgerEntry resEntry = null;
-                if (request instanceof BatchAppendEntryRequest) {
-                    BatchAppendEntryRequest batchRequest = (BatchAppendEntryRequest) request;
-                    if (batchRequest.getBatchMsgs() != null && batchRequest.getBatchMsgs().size() != 0) {
-                        // record positions to return;
-                        long[] positions = new long[batchRequest.getBatchMsgs().size()];
-                        // split bodys to append
-                        int index = 0;
-                        for (byte[] bytes : batchRequest.getBatchMsgs()) {
-                            DLedgerEntry dLedgerEntry = new DLedgerEntry();
-                            dLedgerEntry.setBody(bytes);
-                            resEntry = dLedgerStore.appendAsLeader(dLedgerEntry);
-                            positions[index++] = resEntry.getPos();
-                        }
-                        // only wait last entry ack is ok
-                        future = new BatchAppendFuture<>();
-                        ((BatchAppendFuture<?>) future).setPositions(positions);
-                    } else {
-                        throw new DLedgerException(DLedgerResponseCode.REQUEST_WITH_EMPTY_BODYS, "BatchAppendEntryRequest" +
-                                " with empty bodys");
-                    }
-                } else {
-                    DLedgerEntry dLedgerEntry = new DLedgerEntry();
-                    dLedgerEntry.setBody(request.getBody());
-                    resEntry = dLedgerStore.appendAsLeader(dLedgerEntry);
+            }
+            AppendFuture<AppendEntryResponse> future;
+            if (request instanceof BatchAppendEntryRequest) {
+                BatchAppendEntryRequest batchRequest = (BatchAppendEntryRequest) request;
+                if (batchRequest.getBatchMsgs() == null || batchRequest.getBatchMsgs().isEmpty()) {
+                    throw new DLedgerException(DLedgerResponseCode.REQUEST_WITH_EMPTY_BODYS, "BatchAppendEntryRequest" +
+                        " with empty bodys");
                 }
-                DLedgerEntry finalResEntry = resEntry;
-                AppendFuture<AppendEntryResponse> finalFuture = future;
-                Closure closure = new Closure() {
-                    @Override
-                    void done(Status status) {
-                        AppendEntryResponse response = new AppendEntryResponse();
-                        response.setGroup(DLedgerServer.this.memberState.getGroup());
-                        response.setTerm(DLedgerServer.this.memberState.currTerm());
-                        response.setIndex(finalResEntry.getIndex());
-                        response.setLeaderId(DLedgerServer.this.memberState.getLeaderId());
-                        response.setPos(finalResEntry.getPos());
-                        response.setCode(status.code.getCode());
-                        finalFuture.complete(response);
-                    }
-                };
-                dLedgerEntryPusher.appendClosure(closure, resEntry.getTerm(), resEntry.getIndex());
-                return finalFuture;
+                future = appendAsLeader(batchRequest.getBatchMsgs());
+            } else {
+                future = appendAsLeader(request.getBody());
             }
+            return future;
         } catch (DLedgerException e) {
             LOGGER.error("[{}][HandleAppend] failed", memberState.getSelfId(), e);
             AppendEntryResponse response = new AppendEntryResponse();
@@ -322,6 +289,52 @@ void done(Status status) {
         }
     }
 
+    public AppendFuture<AppendEntryResponse> appendAsLeader(byte[] body) throws DLedgerException {
+        return this.appendAsLeader(Arrays.asList(body));
+    }
+
+    public AppendFuture<AppendEntryResponse> appendAsLeader(List<byte[]> bodies) throws DLedgerException {
+        PreConditions.check(memberState.isLeader(), DLedgerResponseCode.NOT_LEADER);
+        if (bodies.size() == 0) {
+            return AppendFuture.newCompletedFuture(-1, null);
+        }
+        AppendFuture<AppendEntryResponse> future;
+        DLedgerEntry entry = new DLedgerEntry();
+        if (bodies.size() > 1) {
+            long[] positions = new long[bodies.size()];
+            for (int i = 0; i < bodies.size(); i++) {
+                DLedgerEntry dLedgerEntry = new DLedgerEntry();
+                dLedgerEntry.setBody(bodies.get(i));
+                entry = dLedgerStore.appendAsLeader(dLedgerEntry);
+                positions[i] = entry.getPos();
+            }
+            // only wait last entry ack is ok
+            future = new BatchAppendFuture<>(positions);
+        } else {
+            DLedgerEntry dLedgerEntry = new DLedgerEntry();
+            dLedgerEntry.setBody(bodies.get(0));
+            entry = dLedgerStore.appendAsLeader(dLedgerEntry);
+            future = new AppendFuture<>();
+        }
+        final DLedgerEntry finalResEntry = entry;
+        final AppendFuture<AppendEntryResponse> finalFuture = future;
+        Closure closure = new Closure() {
+            @Override
+            void done(Status status) {
+                AppendEntryResponse response = new AppendEntryResponse();
+                response.setGroup(DLedgerServer.this.memberState.getGroup());
+                response.setTerm(DLedgerServer.this.memberState.currTerm());
+                response.setIndex(finalResEntry.getIndex());
+                response.setLeaderId(DLedgerServer.this.memberState.getLeaderId());
+                response.setPos(finalResEntry.getPos());
+                response.setCode(status.code.getCode());
+                finalFuture.complete(response);
+            }
+        };
+        dLedgerEntryPusher.appendClosure(closure, finalResEntry.getTerm(), finalResEntry.getIndex());
+        return finalFuture;
+    }
+
     @Override
     public void handleRead(ReadMode mode, ReadClosure closure) {
         try {
@@ -354,7 +367,9 @@ private void dealUnsafeRead(ReadClosure closure) throws DLedgerException {
     private void dealRaftLogRead(ReadClosure closure) throws DLedgerException {
         PreConditions.check(memberState.isLeader(), DLedgerResponseCode.NOT_LEADER);
         // append an empty raft log, call closure when this raft log is applied
-        DLedgerEntry dLedgerEntry = dLedgerStore.appendAsLeader(new DLedgerEntry());
+        DLedgerEntry emptyEntry = new DLedgerEntry();
+        emptyEntry.setBody(new byte[0]);
+        DLedgerEntry dLedgerEntry = dLedgerStore.appendAsLeader(emptyEntry);
         dLedgerEntryPusher.appendClosure(closure, dLedgerEntry.getTerm(), dLedgerEntry.getIndex());
     }
 
@@ -434,7 +449,7 @@ public CompletableFuture<PushEntryResponse> handlePush(PushEntryRequest request)
 
     @Override
     public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(
-            LeadershipTransferRequest request) throws Exception {
+        LeadershipTransferRequest request) throws Exception {
         LOGGER.info("handleLeadershipTransfer: {}", request);
         try {
             PreConditions.check(memberState.getSelfId().equals(request.getRemoteId()), DLedgerResponseCode.UNKNOWN_MEMBER, "%s != %s", request.getRemoteId(), memberState.getSelfId());
@@ -448,7 +463,7 @@ public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(
                 // check fall transferee not fall behind much.
                 long transfereeFallBehind = dLedgerStore.getLedgerEndIndex() - dLedgerEntryPusher.getPeerWaterMark(request.getTerm(), request.getTransfereeId());
                 PreConditions.check(transfereeFallBehind < dLedgerConfig.getMaxLeadershipTransferWaitIndex(),
-                        DLedgerResponseCode.FALL_BEHIND_TOO_MUCH, "transferee fall behind too much, diff=%s", transfereeFallBehind);
+                    DLedgerResponseCode.FALL_BEHIND_TOO_MUCH, "transferee fall behind too much, diff=%s", transfereeFallBehind);
                 return dLedgerLeaderElector.handleLeadershipTransfer(request);
             } else if (memberState.getSelfId().equals(request.getTransfereeId())) {
                 // It's the transferee received the take leadership command.
@@ -462,8 +477,8 @@ public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(
 
                     if (costTime > dLedgerConfig.getLeadershipTransferWaitTimeout()) {
                         throw new DLedgerException(DLedgerResponseCode.TAKE_LEADERSHIP_FAILED,
-                                "transferee fall behind, wait timeout. timeout = {}, diff = {}",
-                                dLedgerConfig.getLeadershipTransferWaitTimeout(), fallBehind);
+                            "transferee fall behind, wait timeout. timeout = {}, diff = {}",
+                            dLedgerConfig.getLeadershipTransferWaitTimeout(), fallBehind);
                     }
 
                     LOGGER.warn("transferee fall behind, diff = {}", fallBehind);

dledger/src/main/java/io/openmessaging/storage/dledger/protocol/userdefine/UserDefineProcessor.java

@@ -16,12 +16,19 @@
 
 package io.openmessaging.storage.dledger.protocol.userdefine;
 
+import io.openmessaging.storage.dledger.DLedgerServer;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.util.concurrent.CompletableFuture;
 
 public abstract class UserDefineProcessor<T extends UserDefineRequest, V extends UserDefineResponse> {
 
+    protected final DLedgerServer dLedgerServer;
+
+    public UserDefineProcessor(DLedgerServer dLedgerServer) {
+        this.dLedgerServer = dLedgerServer;
+    }
+
     public abstract CompletableFuture<V> handleRequest(T t);
 
     public abstract Integer getRequestTypeCode();
@@ -37,5 +44,4 @@ public Type getResponseType() {
         return parameterizedType.getActualTypeArguments()[1];
     }
 
-
 }

dledger/src/test/java/io/openmessaging/storage/dledger/util/BytesUtil.java → dledger/src/main/java/io/openmessaging/storage/dledger/utils/BytesUtil.java

@@ -14,11 +14,11 @@
  * limitations under the License.
  */
 
-package io.openmessaging.storage.dledger.util;
+package io.openmessaging.storage.dledger.utils;
 
 public class BytesUtil {
 
-    public static byte[] intToBytes( int value ) {
+    public static byte[] intToBytes(int value) {
         byte[] src = new byte[4];
         src[3] = (byte) ((value >> 24) & 0xFF);
         src[2] = (byte) ((value >> 16) & 0xFF);
@@ -30,9 +30,9 @@ public static byte[] intToBytes( int value ) {
     public static int bytesToInt(byte[] src, int offset) {
         int value;
         value = (int) ((src[offset] & 0xFF)
-                | ((src[offset+1] & 0xFF)<<8)
-                | ((src[offset+2] & 0xFF)<<16)
-                | ((src[offset+3] & 0xFF)<<24));
+            | ((src[offset + 1] & 0xFF) << 8)
+            | ((src[offset + 2] & 0xFF) << 16)
+            | ((src[offset + 3] & 0xFF) << 24));
         return value;
     }
 }

dledger/src/test/java/io/openmessaging/storage/dledger/AppendAndReadTest.java

@@ -19,12 +19,11 @@
 import io.openmessaging.storage.dledger.client.DLedgerClient;
 import io.openmessaging.storage.dledger.protocol.AppendEntryResponse;
 import io.openmessaging.storage.dledger.protocol.DLedgerResponseCode;
-import io.openmessaging.storage.dledger.protocol.RequestOrResponse;
 import io.openmessaging.storage.dledger.statemachine.register.RegisterReadProcessor;
 import io.openmessaging.storage.dledger.statemachine.register.RegisterReadRequest;
 import io.openmessaging.storage.dledger.statemachine.register.RegisterReadResponse;
 import io.openmessaging.storage.dledger.statemachine.register.RegisterStateMachine;
-import io.openmessaging.storage.dledger.util.BytesUtil;
+import io.openmessaging.storage.dledger.utils.BytesUtil;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 

dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/register/RegisterReadProcessor.java

@@ -29,10 +29,8 @@ public class RegisterReadProcessor extends UserDefineProcessor<RegisterReadReque
 
     private Integer requestTypeCode;
 
-    private final DLedgerServer server;
-
     public RegisterReadProcessor(DLedgerServer server) {
-        this.server = server;
+        super(server);
         RegisterReadRequest registerReadRequest = new RegisterReadRequest(0);
         this.requestTypeCode = registerReadRequest.getRequestTypeCode();
     }
@@ -46,7 +44,7 @@ public CompletableFuture<RegisterReadResponse> handleRequest(RegisterReadRequest
             @Override
             public void done(Status status) {
                 if (status.isOk()) {
-                    RegisterStateMachine registerStateMachine = (RegisterStateMachine) RegisterReadProcessor.this.server.getStateMachine();
+                    RegisterStateMachine registerStateMachine = (RegisterStateMachine) dLedgerServer.getStateMachine();
                     Integer value = registerStateMachine.getValue(key);
                     response.setValue(value);
                     future.complete(response);
@@ -56,7 +54,7 @@ public void done(Status status) {
                 }
             }
         };
-        RegisterReadProcessor.this.server.handleRead(ReadMode.RAFT_LOG_READ, closure);
+        dLedgerServer.handleRead(ReadMode.RAFT_LOG_READ, closure);
         return future;
     }
 

dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/register/RegisterStateMachine.java

@@ -23,7 +23,7 @@
 import io.openmessaging.storage.dledger.statemachine.CommittedEntryIterator;
 import io.openmessaging.storage.dledger.statemachine.RegisterSnapshotFile;
 import io.openmessaging.storage.dledger.statemachine.StateMachine;
-import io.openmessaging.storage.dledger.util.BytesUtil;
+import io.openmessaging.storage.dledger.utils.BytesUtil;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -37,7 +37,7 @@ public class RegisterStateMachine implements StateMachine {
     public void onApply(CommittedEntryIterator iter) {
         while (iter.hasNext()) {
             final DLedgerEntry entry = iter.next();
-            if (entry != null && entry.getBody() != null) {
+            if (entry != null && entry.getBody() != null && entry.getBody().length == 8) {
                 byte[] bytes = entry.getBody();
                 int key = BytesUtil.bytesToInt(bytes, 0);
                 int value = BytesUtil.bytesToInt(bytes, 4);

example/pom.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>io.openmessaging.storage</groupId>
+        <artifactId>dledger-all</artifactId>
+        <version>0.3.3-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>dledger-example</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.openmessaging.storage</groupId>
+            <artifactId>dledger-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.beust</groupId>
+            <artifactId>jcommander</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <mainClass>io.openmessaging.storage.dledger.example.register.BossCommand
+                            </mainClass>
+                        </manifest>
+                    </archive>
+                    <finalName>register-dledger</finalName>
+                    <appendAssemblyId>false</appendAssemblyId>
+                    <attach>false</attach>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>