This is an automated email from the ASF dual-hosted git repository.
chaow pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.12 by this push:
new a203da6 init dummyIndex after restart cluster (#3939) (#4104)
a203da6 is described below
commit a203da61130a3647b8b6dc41c45fd45e1a52fe76
Author: Houliang Qi <[email protected]>
AuthorDate: Mon Oct 11 17:47:49 2021 +0800
init dummyIndex after restart cluster (#3939) (#4104)
Co-authored-by: lisijia <[email protected]>
---
.../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 d8d511b..43b35ce 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 446eefc..3ae298c 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();
