Refactor LogRecord

Author: szetszwo

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogRecord.java

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ratis.server.raftlog.segmented;
+
+import org.apache.ratis.proto.RaftProtos.LogEntryProto;
+import org.apache.ratis.server.protocol.TermIndex;
+import org.apache.ratis.server.raftlog.LogEntryHeader;
+import org.apache.ratis.util.Preconditions;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+/**
+ * Log record of a RaftLog segment.
+ */
+public final class LogRecord {
+  public static LogRecord valueOf(long offset, LogEntryProto entry) {
+    return new LogRecord(offset, LogEntryHeader.valueOf(entry));
+  }
+
+  /** starting offset in the file */
+  private final long offset;
+  private final LogEntryHeader header;
+
+  private LogRecord(long offset, LogEntryHeader header) {
+    this.offset = offset;
+    this.header = header;
+  }
+
+  public LogEntryHeader getHeader() {
+    return header;
+  }
+
+  public TermIndex getTermIndex() {
+    return getHeader().getTermIndex();
+  }
+
+  public long getOffset() {
+    return offset;
+  }
+
+  /**
+   * A thread-safed mop supporting random read but NOT random write.
+   * For write operations, it supports append, removeLast and clear.
+   */
+  public static class AppendOnlyMap {
+    private final ConcurrentNavigableMap<Long, LogRecord> map = new ConcurrentSkipListMap<>();
+
+    public int size() {
+      return map.size();
+    }
+
+    public LogRecord getFirst() {
+      final Map.Entry<Long, LogRecord> first = map.firstEntry();
+      return first != null? first.getValue() : null;
+    }
+
+    public LogRecord getLast() {
+      final Map.Entry<Long, LogRecord> last = map.lastEntry();
+      return last != null? last.getValue() : null;
+    }
+
+    public LogRecord get(long i) {
+      return map.get(i);
+    }
+
+    public long append(LogRecord record) {
+      final long index = record.getTermIndex().getIndex();
+      final LogRecord previous = map.put(index, record);
+      Preconditions.assertNull(previous, "previous");
+      return index;
+    }
+
+    public LogRecord removeLast() {
+      final Map.Entry<Long, LogRecord> last = map.pollLastEntry();
+      return Objects.requireNonNull(last, "last == null").getValue();
+    }
+
+    public void clear() {
+      map.clear();
+    }
+  }
+}

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/v2/LogSegment.java

@@ -15,15 +15,16 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.ratis.server.raftlog.segmented.v2;
 
 import org.apache.ratis.proto.RaftProtos.LogEntryProto;
 import org.apache.ratis.server.RaftServerConfigKeys.Log.CorruptionPolicy;
 import org.apache.ratis.server.metrics.SegmentedRaftLogMetrics;
 import org.apache.ratis.server.protocol.TermIndex;
-import org.apache.ratis.server.raftlog.LogEntryHeader;
 import org.apache.ratis.server.raftlog.LogProtoUtils;
 import org.apache.ratis.server.raftlog.RaftLogIOException;
+import org.apache.ratis.server.raftlog.segmented.LogRecord;
 import org.apache.ratis.server.storage.RaftStorage;
 import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.ratis.thirdparty.com.google.common.cache.CacheLoader;
@@ -41,8 +42,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentNavigableMap;
-import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -82,67 +81,6 @@ static long getEntrySize(LogEntryProto entry, Op op) {
     return serialized + CodedOutputStream.computeUInt32SizeNoTag(serialized) + 4L;
   }
 
-  static class LogRecord {
-    /** starting offset in the file */
-    private final long offset;
-    private final LogEntryHeader logEntryHeader;
-
-    LogRecord(long offset, LogEntryProto entry) {
-      this.offset = offset;
-      this.logEntryHeader = LogEntryHeader.valueOf(entry);
-    }
-
-    LogEntryHeader getLogEntryHeader() {
-      return logEntryHeader;
-    }
-
-    TermIndex getTermIndex() {
-      return getLogEntryHeader().getTermIndex();
-    }
-
-    long getOffset() {
-      return offset;
-    }
-  }
-
-  private static class Records {
-    private final ConcurrentNavigableMap<Long, LogRecord> map = new ConcurrentSkipListMap<>();
-
-    int size() {
-      return map.size();
-    }
-
-    LogRecord getFirst() {
-      final Map.Entry<Long, LogRecord> first = map.firstEntry();
-      return first != null? first.getValue() : null;
-    }
-
-    LogRecord getLast() {
-      final Map.Entry<Long, LogRecord> last = map.lastEntry();
-      return last != null? last.getValue() : null;
-    }
-
-    LogRecord get(long i) {
-      return map.get(i);
-    }
-
-    long append(LogRecord record) {
-      final long index = record.getTermIndex().getIndex();
-      final LogRecord previous = map.put(index, record);
-      Preconditions.assertNull(previous, "previous");
-      return index;
-    }
-
-    LogRecord removeLast() {
-      final Map.Entry<Long, LogRecord> last = map.pollLastEntry();
-      return Objects.requireNonNull(last, "last == null").getValue();
-    }
-
-    void clear() {
-      map.clear();
-    }
-  }
-
   static LogSegment newOpenSegment(RaftStorage storage, long start, SizeInBytes maxOpSize,
       SegmentedRaftLogMetrics raftLogMetrics) {
     Preconditions.assertTrue(start >= 0);
@@ -309,10 +247,8 @@ File getFile() {
   /** later replace it with a metric */
   private final AtomicInteger loadingTimes = new AtomicInteger();
 
-  /**
-   * the list of records is more like the index of a segment
-   */
-  private final Records records = new Records();
+  /** Log records of a segment */
+  private final LogRecord.AppendOnlyMap records = new LogRecord.AppendOnlyMap();
   /**
    * the entryCache caches the content of log entries.
    */
@@ -365,7 +301,7 @@ private void append(boolean keepEntryInCache, LogEntryProto entry, Op op) {
           "gap between entries %s and %s", entry.getIndex(), currentLast.getTermIndex().getIndex());
     }
 
-    final LogRecord record = new LogRecord(totalFileSize, entry);
+    final LogRecord record = LogRecord.valueOf(totalFileSize, entry);
     if (keepEntryInCache) {
       putEntryCache(record.getTermIndex(), entry, op);
     }
@@ -424,7 +360,7 @@ synchronized void truncate(long fromIndex) {
       final LogRecord removed = records.removeLast();
       Preconditions.assertSame(index, removed.getTermIndex().getIndex(), "removedIndex");
       removeEntryCache(removed.getTermIndex(), Op.REMOVE_CACHE);
-      totalFileSize = removed.offset;
+      totalFileSize = removed.getOffset();
     }
     isOpen = false;
     this.endIndex = fromIndex - 1;

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/v2/SegmentedRaftLog.java

@@ -29,9 +29,9 @@
 import org.apache.ratis.server.raftlog.RaftLog;
 import org.apache.ratis.server.raftlog.RaftLogBase;
 import org.apache.ratis.server.raftlog.RaftLogIOException;
+import org.apache.ratis.server.raftlog.segmented.LogRecord;
 import org.apache.ratis.server.storage.RaftStorageMetadata;
 import org.apache.ratis.server.storage.RaftStorage;
-import org.apache.ratis.server.raftlog.segmented.v2.LogSegment.LogRecord;
 import org.apache.ratis.server.raftlog.segmented.v2.SegmentedRaftLogCache.TruncateIndices;
 import org.apache.ratis.proto.RaftProtos.LogEntryProto;
 import org.apache.ratis.statemachine.StateMachine;

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/v2/SegmentedRaftLogCache.java

@@ -25,9 +25,9 @@
 import org.apache.ratis.server.raftlog.LogEntryHeader;
 import org.apache.ratis.server.raftlog.LogProtoUtils;
 import org.apache.ratis.server.raftlog.RaftLog;
+import org.apache.ratis.server.raftlog.segmented.LogRecord;
 import org.apache.ratis.server.storage.RaftStorage;
 import org.apache.ratis.server.raftlog.segmented.v2.CacheInvalidationPolicy.CacheInvalidationPolicyDefault;
-import org.apache.ratis.server.raftlog.segmented.v2.LogSegment.LogRecord;
 import org.apache.ratis.util.AutoCloseableLock;
 import org.apache.ratis.util.AutoCloseableReadWriteLock;
 import org.apache.ratis.util.JavaUtils;
@@ -582,7 +582,7 @@ private static void getFromSegment(LogSegment segment, long startIndex,
     endIndex = Math.min(endIndex, startIndex + size - 1);
     int index = offset;
     for (long i = startIndex; i <= endIndex; i++) {
-      entries[index++] = Optional.ofNullable(segment.getLogRecord(i)).map(LogRecord::getLogEntryHeader).orElse(null);
+      entries[index++] = Optional.ofNullable(segment.getLogRecord(i)).map(LogRecord::getHeader).orElse(null);
     }
   }
 

ratis-test/src/test/java/org/apache/ratis/server/ServerRestartTests.java

@@ -36,7 +36,7 @@
 import org.apache.ratis.server.raftlog.segmented.v2.SegmentedRaftLogFormat;
 import org.apache.ratis.server.RaftServerConfigKeys.Log;
 import org.apache.ratis.server.raftlog.segmented.v2.SegmentedRaftLogTestUtils;
-import org.apache.ratis.server.raftlog.segmented.TestSegmentedRaftLog;
+import org.apache.ratis.server.raftlog.segmented.v2.TestSegmentedRaftLog;
 import org.apache.ratis.statemachine.impl.SimpleStateMachine4Testing;
 import org.apache.ratis.statemachine.StateMachine;
 import org.apache.ratis.util.FileUtils;

ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/v2/TestLogSegment.java

@@ -28,6 +28,7 @@
 import org.apache.ratis.server.metrics.SegmentedRaftLogMetrics;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.ratis.server.raftlog.LogProtoUtils;
+import org.apache.ratis.server.raftlog.segmented.LogRecord;
 import org.apache.ratis.server.storage.RaftStorage;
 import org.apache.ratis.server.storage.RaftStorageTestUtils;
 import org.apache.ratis.thirdparty.com.google.protobuf.CodedOutputStream;
@@ -136,7 +137,7 @@ static void checkLogSegment(LogSegment segment, long start, long end,
 
     long offset = SegmentedRaftLogFormat.getHeaderLength();
     for (long i = start; i <= end; i++) {
-      LogSegment.LogRecord record = segment.getLogRecord(i);
+      LogRecord record = segment.getLogRecord(i);
       final TermIndex ti = record.getTermIndex();
       Assertions.assertEquals(i, ti.getIndex());
       Assertions.assertEquals(term, ti.getTerm());

ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/v2/TestRaftLogReadWrite.java

@@ -44,7 +44,7 @@
 import java.util.Collections;
 import java.util.List;
 
-import static org.apache.ratis.server.raftlog.segmented.v2.TestLogSegment.ZERO_START_NULL_END;
+import static org.apache.ratis.server.raftlog.segmented.v2.t TestLogSegment.ZERO_START_NULL_END;
 
 /**
  * Test basic functionality of LogReader, SegmentedRaftLogInputStream, and SegmentedRaftLogOutputStream.

ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/v2/TestSegmentedRaftLogCache.java

@@ -26,7 +26,7 @@
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.ratis.server.raftlog.LogEntryHeader;
 import org.apache.ratis.server.raftlog.LogProtoUtils;
-import org.apache.ratis.server.raftlog.segmented.v2.LogSegment.LogRecord;
+import org.apache.ratis.server.raftlog.segmented.LogRecord;
 import org.apache.ratis.server.raftlog.segmented.v2.SegmentedRaftLogCache.TruncationSegments;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;