[ISSUE #275] Complete snapshot(sync mode to install) (#298)

* feat(core): support protocol about install snapshot

1. support protocol about install snapshot

Closes #275

* feat(core): support protocol about install snapshot

1. support protocol about install snapshot

Closes #275

* feat(core): refactor basic dledger overall structure to make it more "raft" like

1. refactor basic dledger overall structure to make it more "raft" like

* feat(core): pass all original test

1. pass all original test

* feat(core): support batch append

1. support batch append

* fix(example): resolve conflicts after rebasing master

1. resolve conflicts after rebasing master

* fix(jepsen): resolve conflicts about jepsen after rebasing master

1. resolve conflicts about jepsen after rebasing master

* fix(jepsen): fix type error

1. fix type error

* feat(core): support installing snapshot

1. support installing snapshot

* feat(core): support installing snapshot

1. support installing snapshot

* feat(jepsen): test snapshot in jepsen

1. test snapshot in jepsen

* test(core): polish flaky test

1. polish flaky test

* rerun

* feat(core): commit entry which is in current term

1. commit entry which is in current term

* rerun

* rerun

* rerun

* fix(core): make the field: position in DLedgerEntry meaningless

1. make the field: position in DLedgerEntry meaningless

* test(core): use different store base path for each ut

1. use different store base path for each ut

* test(core): use different store base path for each ut

1. use different store base path for each ut

* rerun

* test(core): use different store base path for each ut

1. use different store base path for each ut

* test(core): use different store base path for each ut

1. use different store base path for each ut

* test(core): use different store base path for each ut

1. use different store base path for each ut

* fix(core): update peer watermark when compare success

1. update peer watermark when compare success

* fix(core): fix

1. fix

* fix(core): fix

1. fix

* test(proxy): remove proxy test

1. remove proxy test

* feat(example): add batch append cmd

1. add batch append cmd

* fix(core): reuse forks

1. reuse forks

* chore(global): add more git ignore file

1. add more git ignore file

* build(global): set reuseForks to false

1.  set reuseForks to false

* rerun

* feat(core): clear pending map and set writeIndex when role dispatcher role change from append to compare

1. clear pending map and set writeIndex when role dispatcher role change
from append to compare

* rerun

.gitignore

@@ -14,4 +14,6 @@ devenv
 .DS_Store
 nohup.out
 *.gz
-dledger-example.jar
+dledger-example.jar
+.clj-kondo
+*.lsp

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

@@ -16,6 +16,7 @@
 
 package io.openmessaging.storage.dledger;
 
+import io.openmessaging.storage.dledger.snapshot.SnapshotEntryResetStrategy;
 import io.openmessaging.storage.dledger.store.file.DLedgerMmapFileStore;
 
 import io.openmessaging.storage.dledger.utils.DLedgerUtils;
@@ -86,16 +87,31 @@ public class DLedgerConfig {
     private int minTakeLeadershipVoteIntervalMs = 30;
     private int maxTakeLeadershipVoteIntervalMs = 100;
 
-    private boolean isEnableBatchPush = false;
-    private int maxBatchPushSize = 4 * 1024;
+    private boolean isEnableBatchAppend = false;
+
+    // max size in bytes for each append request
+    private int maxBatchAppendSize = 4 * 1024;
 
     private long leadershipTransferWaitTimeout = 1000;
 
     private boolean enableSnapshot = false;
 
+    private SnapshotEntryResetStrategy snapshotEntryResetStrategy = SnapshotEntryResetStrategy.RESET_ALL_SYNC;
+
     private int snapshotThreshold = 1000;
+
+    private int resetSnapshotEntriesDelayTime = 5 * 1000;
+
+    /**
+     * reset snapshot entries but keep last entries num.
+     * .e.g 10, when we load from snapshot which lastIncludedIndex = 100, we will delete the entries in (..., 90]
+     */
+    private int resetSnapshotEntriesButKeepLastEntriesNum = 10;
     private int maxSnapshotReservedNum = 3;
 
+    // max interval in ms for each append request
+    private int maxBatchAppendIntervalMs = 1000;
+
     public String getDefaultPath() {
         return storeBaseDir + File.separator + "dledger-" + selfId;
     }
@@ -394,20 +410,20 @@ public void setMaxTakeLeadershipVoteIntervalMs(int maxTakeLeadershipVoteInterval
         this.maxTakeLeadershipVoteIntervalMs = maxTakeLeadershipVoteIntervalMs;
     }
 
-    public boolean isEnableBatchPush() {
-        return isEnableBatchPush;
+    public boolean isEnableBatchAppend() {
+        return isEnableBatchAppend;
     }
 
-    public void setEnableBatchPush(boolean enableBatchPush) {
-        isEnableBatchPush = enableBatchPush;
+    public void setEnableBatchAppend(boolean enableBatchAppend) {
+        isEnableBatchAppend = enableBatchAppend;
     }
 
-    public int getMaxBatchPushSize() {
-        return maxBatchPushSize;
+    public int getMaxBatchAppendSize() {
+        return maxBatchAppendSize;
     }
 
-    public void setMaxBatchPushSize(int maxBatchPushSize) {
-        this.maxBatchPushSize = maxBatchPushSize;
+    public void setMaxBatchAppendSize(int maxBatchAppendSize) {
+        this.maxBatchAppendSize = maxBatchAppendSize;
     }
 
     public long getLeadershipTransferWaitTimeout() {
@@ -495,4 +511,37 @@ public boolean isEnableSnapshot() {
     public void setEnableSnapshot(boolean enableSnapshot) {
         this.enableSnapshot = enableSnapshot;
     }
+
+    public int getMaxBatchAppendIntervalMs() {
+        return maxBatchAppendIntervalMs;
+    }
+
+    public SnapshotEntryResetStrategy getSnapshotEntryResetStrategy() {
+        return snapshotEntryResetStrategy;
+    }
+
+    public void setSnapshotEntryResetStrategy(
+        SnapshotEntryResetStrategy snapshotEntryResetStrategy) {
+        this.snapshotEntryResetStrategy = snapshotEntryResetStrategy;
+    }
+
+    public void setMaxBatchAppendIntervalMs(int maxBatchAppendIntervalMs) {
+        this.maxBatchAppendIntervalMs = maxBatchAppendIntervalMs;
+    }
+
+    public int getResetSnapshotEntriesDelayTime() {
+        return resetSnapshotEntriesDelayTime;
+    }
+
+    public void setResetSnapshotEntriesDelayTime(int resetSnapshotEntriesDelayTime) {
+        this.resetSnapshotEntriesDelayTime = resetSnapshotEntriesDelayTime;
+    }
+
+    public int getResetSnapshotEntriesButKeepLastEntriesNum() {
+        return resetSnapshotEntriesButKeepLastEntriesNum;
+    }
+
+    public void setResetSnapshotEntriesButKeepLastEntriesNum(int resetSnapshotEntriesButKeepLastEntriesNum) {
+        this.resetSnapshotEntriesButKeepLastEntriesNum = resetSnapshotEntriesButKeepLastEntriesNum;
+    }
 }