<NME> SegmentedLog.java
<BEF> package com.github.wenweihu86.raft.storage;

import com.github.wenweihu86.raft.RaftOption;
import com.github.wenweihu86.raft.util.RaftFileUtils;
import com.github.wenweihu86.raft.proto.Raft;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.RandomAccessFile;
import java.io.IOException;
import java.util.List;
import java.util.ArrayList;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by wenweihu86 on 2017/5/3.
 */
public class SegmentedLog {

    private static Logger LOG = LoggerFactory.getLogger(SegmentedLog.class);

    private String logDir = RaftOption.dataDir + File.pathSeparator + "log";
    private Raft.LogMetaData metaData;
    private List<Segment> segments = new ArrayList<>();
    private TreeMap<Long, Segment> startLogIndexSegmentMap = new TreeMap<>();
    private AtomicLong openedSegmentIndex = new AtomicLong(0);
    // segment log占用的内存大小，用于判断是否需要做snapshot
    private volatile long totalSize;

    private volatile long totalSize;

        if (!file.exists()) {
            file.mkdirs();
        }
        segments = this.readSegments();
        for (Segment segment : segments) {
            this.loadSegmentData(segment);
        }

        metaData = this.readMetaData();
        if (metaData == null) {
            if (segments.size() > 0) {
                LOG.error("No readable metadata file but found segments in {}", logDir);
                throw new RuntimeException("No readable metadata file but found segments");
            }
        if (metaData == null) {
            if (startLogIndexSegmentMap.size() > 0) {
                LOG.error("No readable metadata file but found segments in {}", logDir);
                throw new RuntimeException("No readable metadata file but found segments");
            }
            metaData = RaftProto.LogMetaData.newBuilder().setFirstLogIndex(1).build();
        }
    }

    public RaftProto.LogEntry getEntry(long index) {
        long firstLogIndex = getFirstLogIndex();
        long lastLogIndex = getLastLogIndex();
        if (index == 0 || index < firstLogIndex || index > lastLogIndex) {
            LOG.debug("index out of range, index={}, firstLogIndex={}, lastLogIndex={}",
                    index, firstLogIndex, lastLogIndex);
            return null;
        }
        if (startLogIndexSegmentMap.size() == 0) {
            return null;
        }
        Segment segment = startLogIndexSegmentMap.floorEntry(index).getValue();
        return segment.getEntry(index);
    }

    public long getEntryTerm(long index) {
        RaftProto.LogEntry entry = getEntry(index);
        if (entry == null) {
            return 0;
        }
        return entry.getTerm();
    }

    public long getFirstLogIndex() {
        return metaData.getFirstLogIndex();
    }

    public long getLastLogIndex() {
    public long append(List<Raft.LogEntry> entries) {
        long newLastLogIndex = this.getLastLogIndex();
        for (Raft.LogEntry entry : entries) {
            int entrySize = entry.getSerializedSize();
            int segmentSize = segments.size();
            boolean isNeedNewSegmentFile = false;
            try {
                if (segmentSize == 0) {
                    isNeedNewSegmentFile = true;
                } else {
                    Segment segment = segments.get(segmentSize - 1);
                    if (!segment.isCanWrite()) {
                        isNeedNewSegmentFile = true;
                    } else if (segment.getFileSize() + entrySize >= RaftOption.maxSegmentFileSize) {
            int segmentSize = startLogIndexSegmentMap.size();
            boolean isNeedNewSegmentFile = false;
            try {
                if (segmentSize == 0) {
                    isNeedNewSegmentFile = true;
                } else {
                    Segment segment = startLogIndexSegmentMap.lastEntry().getValue();
                    if (!segment.isCanWrite()) {
                        isNeedNewSegmentFile = true;
                    } else if (segment.getFileSize() + entrySize >= maxSegmentFileSize) {
                        isNeedNewSegmentFile = true;
                        segment.setRandomAccessFile(RaftFileUtils.openFile(logDir, newFileName, "r"));
                    }
                }
                // 新建segment文件
                if (isNeedNewSegmentFile) {
                    // open new segment file
                    String newSegmentFileName = String.format("open-%d", openedSegmentIndex.getAndIncrement());
                    File newSegmentFile = new File(newSegmentFileName);
                    newSegmentFile.createNewFile();
                    Segment segment = new Segment();
                    segment.setCanWrite(true);
                    segment.setStartIndex(0);
                    segment.setEndIndex(0);
                    segment.setFileName(newSegmentFileName);
                    segment.setRandomAccessFile(RaftFileUtils.openFile(logDir, newSegmentFileName, "rw"));
                    segments.add(segment);
                }
                // 写proto到segment中
                if (entry.getIndex() == 0) {
                    newLastLogIndex++;
                    entry = Raft.LogEntry.newBuilder(entry)
                            .setIndex(newLastLogIndex).build();
                } else {
                    newLastLogIndex = entry.getIndex();
                }
                segmentSize = segments.size();
                Segment segment = segments.get(segmentSize - 1);
                if (segment.getStartIndex() == 0) {
                    segment.setStartIndex(entry.getIndex());
                    startLogIndexSegmentMap.put(segment.getStartIndex(), segment);
                }
                segment.setEndIndex(entry.getIndex());
                segment.getEntries().add(new Segment.Record(
                        segment.getRandomAccessFile().getFilePointer(), entry));
                RaftFileUtils.writeProtoToFile(segment.getRandomAccessFile(), entry);
                segment.setFileSize(segment.getRandomAccessFile().length());
                totalSize += entrySize;
            }  catch (IOException ex) {
                throw new RuntimeException("meet exception, msg=" + ex.getMessage());
                    entry = RaftProto.LogEntry.newBuilder(entry)
                            .setIndex(newLastLogIndex).build();
        return newLastLogIndex;
    }

    public void loadSegmentData(Segment segment) {
        try {
            RandomAccessFile randomAccessFile = segment.getRandomAccessFile();
        if (newEndIndex >= getLastLogIndex()) {
            return;
        }
        LOG.info("Truncating log from old end index {} to new end index {}",
                getLastLogIndex(), newEndIndex);
        while (!startLogIndexSegmentMap.isEmpty()) {
            Segment segment = startLogIndexSegmentMap.lastEntry().getValue();
            try {
                if (newEndIndex == segment.getEndIndex()) {
                    break;
                } else if (newEndIndex < segment.getStartIndex()) {
                    totalSize -= segment.getFileSize();
                    // delete file
                    segment.getRandomAccessFile().close();
                    String fullFileName = logDataDir + File.separator + segment.getFileName();
                    FileUtils.forceDelete(new File(fullFileName));
                    startLogIndexSegmentMap.remove(segment.getStartIndex());
                } else if (newEndIndex < segment.getEndIndex()) {
        }
    }

    public List<Segment> readSegments() {
        List<String> fileNames = RaftFileUtils.getSortedFilesInDirectory(logDir);
        for (String fileName: fileNames) {
            if (fileName.equals("metadata")) {
                    FileChannel fileChannel = segment.getRandomAccessFile().getChannel();
                    fileChannel.truncate(segment.getFileSize());
                    fileChannel.close();
                    segment.getRandomAccessFile().close();
                    String oldFullFileName = logDataDir + File.separator + segment.getFileName();
                    String newFileName = String.format("%020d-%020d",
                            segment.getStartIndex(), segment.getEndIndex());
                    segment.setFileName(newFileName);
                    String newFullFileName = logDataDir + File.separator + segment.getFileName();
            try {
                if (splitArray[0].equals("open")) {
                    segment.setCanWrite(true);
                    segment.setStartIndex(-1);
                    segment.setEndIndex(-1);
                } else {
                    try {
                        segment.setCanWrite(false);

    public void loadSegmentData(Segment segment) {
        try {
            RandomAccessFile randomAccessFile = segment.getRandomAccessFile();
            long totalLength = segment.getFileSize();
            long offset = 0;
                }
                segment.setRandomAccessFile(RaftFileUtils.openFile(logDir, fileName, "r"));
                segment.setFileSize(segment.getRandomAccessFile().length());
                segments.add(segment);
            } catch (IOException ioException) {
                LOG.warn("open segment file error, file={}, msg={}",
                        fileName, ioException.getMessage());
                throw new RuntimeException("open segment file error");
            }
        }
        return segments;
    }


    public Raft.LogMetaData getMetaData() {
        return metaData;
    }

        int entrySize = segment.getEntries().size();
        if (entrySize > 0) {
            segment.setStartIndex(segment.getEntries().get(0).entry.getIndex());
            segment.setEndIndex(segment.getEntries().get(entrySize - 1).entry.getIndex());
        }
    }

    public void readSegments() {
        try {
            List<String> fileNames = RaftFileUtils.getSortedFilesInDirectory(logDataDir, logDataDir);
            for (String fileName : fileNames) {
                String[] splitArray = fileName.split("-");
                if (splitArray.length != 2) {
                    LOG.warn("segment filename[{}] is not valid", fileName);
                    continue;
                }
                Segment segment = new Segment();
                segment.setFileName(fileName);
                if (splitArray[0].equals("open")) {
                    segment.setCanWrite(true);
                    segment.setStartIndex(Long.valueOf(splitArray[1]));
                    segment.setEndIndex(0);
                } else {
                    try {
                        segment.setCanWrite(false);
                        segment.setStartIndex(Long.parseLong(splitArray[0]));
                        segment.setEndIndex(Long.parseLong(splitArray[1]));
                    } catch (NumberFormatException ex) {
                        LOG.warn("segment filename[{}] is not valid", fileName);
                        continue;
                    }
                }
                segment.setRandomAccessFile(RaftFileUtils.openFile(logDataDir, fileName, "rw"));
                segment.setFileSize(segment.getRandomAccessFile().length());
                startLogIndexSegmentMap.put(segment.getStartIndex(), segment);
            }
        } catch(IOException ioException){
            LOG.warn("readSegments exception:", ioException);
            throw new RuntimeException("open segment file error");
        }
    }

    public RaftProto.LogMetaData readMetaData() {
        String fileName = logDir + File.separator + "metadata";
        File file = new File(fileName);
        try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
            RaftProto.LogMetaData metadata = RaftFileUtils.readProtoFromFile(
                    randomAccessFile, RaftProto.LogMetaData.class);
            return metadata;
        } catch (IOException ex) {
            LOG.warn("meta file not exist, name={}", fileName);
            return null;
        }
    }

    /**
     * 更新raft log meta data，
     * 包括commitIndex， fix bug: https://github.com/wenweihu86/raft-java/issues/19
     * @param currentTerm
     * @param votedFor
     * @param firstLogIndex
     * @param commitIndex
     */
    public void updateMetaData(Long currentTerm, Integer votedFor, Long firstLogIndex, Long commitIndex) {
        RaftProto.LogMetaData.Builder builder = RaftProto.LogMetaData.newBuilder(this.metaData);
        if (currentTerm != null) {
            builder.setCurrentTerm(currentTerm);
        }
        if (votedFor != null) {
            builder.setVotedFor(votedFor);
        }
        if (firstLogIndex != null) {
            builder.setFirstLogIndex(firstLogIndex);
        }
        if (commitIndex != null) {
            builder.setCommitIndex(commitIndex);
        }
        this.metaData = builder.build();

        String fileName = logDir + File.separator + "metadata";
        File file = new File(fileName);
        try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) {
            RaftFileUtils.writeProtoToFile(randomAccessFile, metaData);
            LOG.info("new segment meta info, currentTerm={}, votedFor={}, firstLogIndex={}",
                    metaData.getCurrentTerm(), metaData.getVotedFor(), metaData.getFirstLogIndex());
        } catch (IOException ex) {
            LOG.warn("meta file not exist, name={}", fileName);
        }
    }

    public RaftProto.LogMetaData getMetaData() {
        return metaData;
    }

    public long getTotalSize() {
        return totalSize;
    }

}

<MSG> add truncate log suffix

<DFF> @@ -3,16 +3,16 @@ package com.github.wenweihu86.raft.storage;
 import com.github.wenweihu86.raft.RaftOption;
 import com.github.wenweihu86.raft.util.RaftFileUtils;
 import com.github.wenweihu86.raft.proto.Raft;
+import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.RandomAccessFile;
 import java.io.IOException;
+import java.nio.channels.FileChannel;
 import java.util.List;
-import java.util.ArrayList;
 import java.util.TreeMap;
-import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Created by wenweihu86 on 2017/5/3.
@@ -23,9 +23,7 @@ public class SegmentedLog {
 
     private String logDir = RaftOption.dataDir + File.pathSeparator + "log";
     private Raft.LogMetaData metaData;
-    private List<Segment> segments = new ArrayList<>();
     private TreeMap<Long, Segment> startLogIndexSegmentMap = new TreeMap<>();
-    private AtomicLong openedSegmentIndex = new AtomicLong(0);
     // segment log占用的内存大小，用于判断是否需要做snapshot
     private volatile long totalSize;
 
@@ -34,14 +32,14 @@ public class SegmentedLog {
         if (!file.exists()) {
             file.mkdirs();
         }
-        segments = this.readSegments();
-        for (Segment segment : segments) {
+        readSegments();
+        for (Segment segment : startLogIndexSegmentMap.values()) {
             this.loadSegmentData(segment);
         }
 
         metaData = this.readMetaData();
         if (metaData == null) {
-            if (segments.size() > 0) {
+            if (startLogIndexSegmentMap.size() > 0) {
                 LOG.error("No readable metadata file but found segments in {}", logDir);
                 throw new RuntimeException("No readable metadata file but found segments");
             }
@@ -85,14 +83,15 @@ public class SegmentedLog {
     public long append(List<Raft.LogEntry> entries) {
         long newLastLogIndex = this.getLastLogIndex();
         for (Raft.LogEntry entry : entries) {
+            newLastLogIndex++;
             int entrySize = entry.getSerializedSize();
-            int segmentSize = segments.size();
+            int segmentSize = startLogIndexSegmentMap.size();
             boolean isNeedNewSegmentFile = false;
             try {
                 if (segmentSize == 0) {
                     isNeedNewSegmentFile = true;
                 } else {
-                    Segment segment = segments.get(segmentSize - 1);
+                    Segment segment = startLogIndexSegmentMap.lastEntry().getValue();
                     if (!segment.isCanWrite()) {
                         isNeedNewSegmentFile = true;
                     } else if (segment.getFileSize() + entrySize >= RaftOption.maxSegmentFileSize) {
@@ -110,39 +109,36 @@ public class SegmentedLog {
                         segment.setRandomAccessFile(RaftFileUtils.openFile(logDir, newFileName, "r"));
                     }
                 }
+                Segment newSegment = startLogIndexSegmentMap.lastEntry().getValue();
                 // 新建segment文件
                 if (isNeedNewSegmentFile) {
                     // open new segment file
-                    String newSegmentFileName = String.format("open-%d", openedSegmentIndex.getAndIncrement());
+                    String newSegmentFileName = String.format("open-%d", newLastLogIndex);
                     File newSegmentFile = new File(newSegmentFileName);
-                    newSegmentFile.createNewFile();
+                    if (!newSegmentFile.exists()) {
+                        newSegmentFile.createNewFile();
+                    }
                     Segment segment = new Segment();
                     segment.setCanWrite(true);
-                    segment.setStartIndex(0);
+                    segment.setStartIndex(newLastLogIndex);
                     segment.setEndIndex(0);
                     segment.setFileName(newSegmentFileName);
                     segment.setRandomAccessFile(RaftFileUtils.openFile(logDir, newSegmentFileName, "rw"));
-                    segments.add(segment);
+                    newSegment = segment;
                 }
                 // 写proto到segment中
                 if (entry.getIndex() == 0) {
-                    newLastLogIndex++;
                     entry = Raft.LogEntry.newBuilder(entry)
                             .setIndex(newLastLogIndex).build();
-                } else {
-                    newLastLogIndex = entry.getIndex();
                 }
-                segmentSize = segments.size();
-                Segment segment = segments.get(segmentSize - 1);
-                if (segment.getStartIndex() == 0) {
-                    segment.setStartIndex(entry.getIndex());
-                    startLogIndexSegmentMap.put(segment.getStartIndex(), segment);
+                newSegment.setEndIndex(entry.getIndex());
+                newSegment.getEntries().add(new Segment.Record(
+                        newSegment.getRandomAccessFile().getFilePointer(), entry));
+                RaftFileUtils.writeProtoToFile(newSegment.getRandomAccessFile(), entry);
+                newSegment.setFileSize(newSegment.getRandomAccessFile().length());
+                if (startLogIndexSegmentMap.containsKey(newSegment.getStartIndex())) {
+                    startLogIndexSegmentMap.put(newSegment.getStartIndex(), newSegment);
                 }
-                segment.setEndIndex(entry.getIndex());
-                segment.getEntries().add(new Segment.Record(
-                        segment.getRandomAccessFile().getFilePointer(), entry));
-                RaftFileUtils.writeProtoToFile(segment.getRandomAccessFile(), entry);
-                segment.setFileSize(segment.getRandomAccessFile().length());
                 totalSize += entrySize;
             }  catch (IOException ex) {
                 throw new RuntimeException("meet exception, msg=" + ex.getMessage());
@@ -151,6 +147,53 @@ public class SegmentedLog {
         return newLastLogIndex;
     }
 
+    public void truncatePrefix(long newStartIndex) {
+    }
+
+    public void truncateSuffix(long newEndIndex) {
+        if (newEndIndex >= getLastLogIndex()) {
+            return;
+        }
+        LOG.info("Truncating log from old end index {} to new end index {}",
+                getLastLogIndex(), newEndIndex);
+        while (!startLogIndexSegmentMap.isEmpty()) {
+            Segment segment = startLogIndexSegmentMap.lastEntry().getValue();
+            try {
+                if (newEndIndex == segment.getEndIndex()) {
+                    break;
+                } else if (newEndIndex < segment.getStartIndex()) {
+                    totalSize -= segment.getFileSize();
+                    // delete file
+                    segment.getRandomAccessFile().close();
+                    String fullFileName = logDir + File.pathSeparator + segment.getFileName();
+                    FileUtils.forceDelete(new File(fullFileName));
+                    startLogIndexSegmentMap.remove(segment.getFileName());
+                } else if (newEndIndex < segment.getEndIndex()) {
+                    int i = (int) (newEndIndex + 1 - segment.getStartIndex());
+                    segment.setEndIndex(newEndIndex);
+                    long newFileSize = segment.getEntries().get(i).offset;
+                    totalSize -= (segment.getFileSize() - newFileSize);
+                    segment.setFileSize(newFileSize);
+                    segment.getEntries().removeAll(
+                            segment.getEntries().subList(i, segment.getEntries().size()));
+                    FileChannel fileChannel = segment.getRandomAccessFile().getChannel();
+                    fileChannel.truncate(segment.getFileSize());
+                    fileChannel.close();
+                    segment.getRandomAccessFile().close();
+                    String oldFullFileName = logDir + File.pathSeparator + segment.getFileName();
+                    String newFileName = String.format("%020d-%020d",
+                            segment.getStartIndex(), segment.getEndIndex());
+                    segment.setFileName(newFileName);
+                    String newFullFileName = logDir + File.pathSeparator + segment.getFileName();
+                    new File(oldFullFileName).renameTo(new File(newFullFileName));
+                    segment.setRandomAccessFile(RaftFileUtils.openFile(logDir, segment.getFileName(), "rw"));
+                }
+            } catch (IOException ex) {
+                LOG.warn("io exception, msg={}", ex.getMessage());
+            }
+        }
+    }
+
     public void loadSegmentData(Segment segment) {
         try {
             RandomAccessFile randomAccessFile = segment.getRandomAccessFile();
@@ -175,7 +218,7 @@ public class SegmentedLog {
         }
     }
 
-    public List<Segment> readSegments() {
+    public void readSegments() {
         List<String> fileNames = RaftFileUtils.getSortedFilesInDirectory(logDir);
         for (String fileName: fileNames) {
             if (fileName.equals("metadata")) {
@@ -191,8 +234,8 @@ public class SegmentedLog {
             try {
                 if (splitArray[0].equals("open")) {
                     segment.setCanWrite(true);
-                    segment.setStartIndex(-1);
-                    segment.setEndIndex(-1);
+                    segment.setStartIndex(Long.valueOf(splitArray[1]));
+                    segment.setEndIndex(0);
                 } else {
                     try {
                         segment.setCanWrite(false);
@@ -205,17 +248,15 @@ public class SegmentedLog {
                 }
                 segment.setRandomAccessFile(RaftFileUtils.openFile(logDir, fileName, "r"));
                 segment.setFileSize(segment.getRandomAccessFile().length());
-                segments.add(segment);
+                startLogIndexSegmentMap.put(segment.getStartIndex(), segment);
             } catch (IOException ioException) {
                 LOG.warn("open segment file error, file={}, msg={}",
                         fileName, ioException.getMessage());
                 throw new RuntimeException("open segment file error");
             }
         }
-        return segments;
     }
 
-
     public Raft.LogMetaData getMetaData() {
         return metaData;
     }

add truncate log suffix