[iotdb] branch master updated: init dummyIndex after restart cluster (#3939)

[ne...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

neuyilan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 40397fd  init dummyIndex after restart cluster (#3939)
40397fd is described below

commit 40397fd7aed05906121c1bd735014d0e2e2fe981
Author: lisijia <44...@users.noreply.github.com>
AuthorDate: Mon Oct 11 16:58:12 2021 +0800

    init dummyIndex after restart cluster (#3939)
---
 .../iotdb/cluster/log/StableEntryManager.java      |  6 ++++++
 .../cluster/log/manage/CommittedEntryManager.java  | 12 +++++++++++
 .../iotdb/cluster/log/manage/RaftLogManager.java   |  6 +++++-
 .../serializable/SyncLogDequeSerializer.java       | 25 ++++++++++++++++++++++
 4 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/StableEntryManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/StableEntryManager.java
index 62d34f3..c82597e 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/StableEntryManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/StableEntryManager.java
@@ -19,6 +19,8 @@
 
 package org.apache.iotdb.cluster.log;
 
+import org.apache.iotdb.cluster.log.manage.serializable.LogManagerMeta;
+
 import java.io.IOException;
 import java.util.List;
 
@@ -26,6 +28,8 @@ public interface StableEntryManager {
 
   List<Log> getAllEntriesAfterAppliedIndex();
 
+  List<Log> getAllEntriesAfterCommittedIndex();
+
   void append(List<Log> entries, long maxHaveAppliedCommitIndex) throws IOException;
 
   void flushLogBuffer();
@@ -38,6 +42,8 @@ public interface StableEntryManager {
 
   HardState getHardState();
 
+  LogManagerMeta getMeta();
+
   /**
    * @param startIndex (inclusive) the log start index
    * @param endIndex (inclusive) the log end index
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/CommittedEntryManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/CommittedEntryManager.java
index e86df85..4ecd167 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/CommittedEntryManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/CommittedEntryManager.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.cluster.exception.TruncateCommittedEntryException;
 import org.apache.iotdb.cluster.log.Log;
 import org.apache.iotdb.cluster.log.Snapshot;
 import org.apache.iotdb.cluster.log.logtypes.EmptyContentLog;
+import org.apache.iotdb.cluster.log.manage.serializable.LogManagerMeta;
 import org.apache.iotdb.db.utils.TestOnly;
 
 import org.slf4j.Logger;
@@ -53,6 +54,17 @@ public class CommittedEntryManager {
     entryTotalMemSize = 0;
   }
 
+  CommittedEntryManager(int maxNumOfLogInMem, LogManagerMeta meta) {
+    entries = Collections.synchronizedList(new ArrayList<>(maxNumOfLogInMem));
+    entries.add(
+        new EmptyContentLog(
+            meta.getMaxHaveAppliedCommitIndex() == -1
+                ? -1
+                : meta.getMaxHaveAppliedCommitIndex() - 1,
+            meta.getLastLogTerm()));
+    entryTotalMemSize = 0;
+  }
+
   /**
    * Overwrite the contents of this object with those of the given snapshot. Note that this function
    * is only used if you want to override all the contents, otherwise please use
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
index 8709203..e5bfbf6 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.cluster.log.Log;
 import org.apache.iotdb.cluster.log.LogApplier;
 import org.apache.iotdb.cluster.log.Snapshot;
 import org.apache.iotdb.cluster.log.StableEntryManager;
+import org.apache.iotdb.cluster.log.manage.serializable.LogManagerMeta;
 import org.apache.iotdb.cluster.server.monitor.Timer.Statistic;
 import org.apache.iotdb.db.utils.TestOnly;
 import org.apache.iotdb.tsfile.utils.RamUsageEstimator;
@@ -115,7 +116,8 @@ public abstract class RaftLogManager {
   protected RaftLogManager(StableEntryManager stableEntryManager, LogApplier applier, String name) {
     this.logApplier = applier;
     this.name = name;
-    this.setCommittedEntryManager(new CommittedEntryManager(maxNumOfLogsInMem));
+    LogManagerMeta meta = stableEntryManager.getMeta();
+    this.setCommittedEntryManager(new CommittedEntryManager(maxNumOfLogsInMem, meta));
     this.setStableEntryManager(stableEntryManager);
     try {
       this.getCommittedEntryManager().append(stableEntryManager.getAllEntriesAfterAppliedIndex());
@@ -125,6 +127,8 @@ public abstract class RaftLogManager {
     long first = getCommittedEntryManager().getDummyIndex();
     long last = getCommittedEntryManager().getLastIndex();
     this.setUnCommittedEntryManager(new UnCommittedEntryManager(last + 1));
+    this.getUnCommittedEntryManager()
+        .truncateAndAppend(stableEntryManager.getAllEntriesAfterCommittedIndex());
 
     /** must have applied entry [compactIndex,last] to state machine */
     this.commitIndex = last;
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/serializable/SyncLogDequeSerializer.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/serializable/SyncLogDequeSerializer.java
index 87f2c8b..7579a51 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/serializable/SyncLogDequeSerializer.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/serializable/SyncLogDequeSerializer.java
@@ -219,6 +219,7 @@ public class SyncLogDequeSerializer implements StableEntryManager {
   }
 
   /** for log tools */
+  @Override
   public LogManagerMeta getMeta() {
     return meta;
   }
@@ -236,6 +237,30 @@ public class SyncLogDequeSerializer implements StableEntryManager {
     return getLogs(meta.getMaxHaveAppliedCommitIndex(), meta.getCommitLogIndex());
   }
 
+  /**
+   * When raft log files flushed,meta would not be flushed synchronously.So data has flushed to disk
+   * is uncommitted for persistent LogManagerMeta(meta's info is stale).We need to recover these
+   * already persistent logs.
+   *
+   * <p>For example,commitIndex is 5 in persistent LogManagerMeta,But the log file has actually been
+   * flushed to 7,when we restart cluster,we need to recover 6 and 7.
+   *
+   * <p>Maybe,we can extract getAllEntriesAfterAppliedIndex and getAllEntriesAfterCommittedIndex
+   * into getAllEntriesByIndex,but now there are too many test cases using it.
+   */
+  @Override
+  public List<Log> getAllEntriesAfterCommittedIndex() {
+    long lastIndex = firstLogIndex + logIndexOffsetList.size() - 1;
+    logger.debug(
+        "getAllEntriesAfterCommittedIndex, firstUnCommitIndex={}, lastIndexBeforeStart={}",
+        meta.getCommitLogIndex() + 1,
+        lastIndex);
+    if (meta.getCommitLogIndex() >= lastIndex) {
+      return Collections.emptyList();
+    }
+    return getLogs(meta.getCommitLogIndex() + 1, lastIndex);
+  }
+
   @Override
   public void append(List<Log> entries, long maxHaveAppliedCommitIndex) throws IOException {
     lock.lock();