@Test public void recoverTest() throws IOException, InterruptedException { int entrySize = 128; int size = 15; int entriesPerFile = 5; int term = 8; journal.close(); Properties properties = new Properties(); properties.setProperty("persistence.journal.file_data_size", String.valueOf((entrySize + journalEntryParser.headerLength()) * entriesPerFile)); properties.setProperty("persistence.index.file_data_size", String.valueOf(Long.BYTES * entriesPerFile)); journal = createJournal(properties); List<byte[]> entries = ByteUtils.createFixedSizeByteList(entrySize, size); List<byte[]> storageEntries = entries.stream() .map(entry -> journalEntryParser.createJournalEntry(entry)) .peek(entry -> entry.setTerm(term)) .peek(entry -> entry.setPartition(0)) .map(this::serialize) .collect(Collectors.toList()); long maxIndex = journal.appendBatchRaw(storageEntries); Assert.assertEquals(size, maxIndex); Assert.assertEquals(size, journal.maxIndex()); Assert.assertEquals(0, journal.minIndex()); journal.flush(); journal.close(); File lastFile = findLastFile(path); journal = createJournal(properties); Assert.assertEquals(size, journal.maxIndex()); Assert.assertEquals(0, journal.minIndex()); long index = journal.minIndex(); while (index < journal.maxIndex()) { Assert.assertArrayEquals(entries.get((int) index), journal.read(index).getPayload().getBytes()); index++; } journal.flush(); journal.close(); lastFile = findLastFile(path); try (RandomAccessFile raf = new RandomAccessFile(lastFile, "rw"); FileChannel fileChannel = raf.getChannel()) { fileChannel.truncate(fileChannel.size() - entrySize / 3); } size--; journal = createJournal(properties); Assert.assertEquals(size, journal.maxIndex()); Assert.assertEquals(0, journal.minIndex()); index = journal.minIndex(); while (index < journal.maxIndex()) { Assert.assertArrayEquals(entries.get((int) index), journal.read(index).getPayload().getBytes()); index++; } journal.flush(); journal.close(); lastFile = findLastFile(path.resolve("index").resolve("all")); try (RandomAccessFile raf = new RandomAccessFile(lastFile, "rw"); FileChannel fileChannel = raf.getChannel()) { fileChannel.truncate(fileChannel.size() - Long.BYTES * 2 - 1); } journal = createJournal(properties); Assert.assertEquals(size, journal.maxIndex()); Assert.assertEquals(0, journal.minIndex()); index = journal.minIndex(); while (index < journal.maxIndex()) { Assert.assertArrayEquals(entries.get((int) index), journal.read(index).getPayload().getBytes()); index++; } }
public void recover(Path path, long commitIndex, JournalSnapshot journalSnapshot, Properties properties) throws IOException { this.basePath = path; Path indexPath = path.resolve(INDEX_PATH); Path partitionPath = path.resolve(PARTITION_PATH); journalProperties = replacePropertiesNames(properties, JOURNAL_PROPERTIES_PATTERN, DEFAULT_JOURNAL_PROPERTIES); journalPersistence.recover(path, journalSnapshot.minOffset(), journalProperties); truncateJournalTailPartialEntry(); indexProperties = replacePropertiesNames(properties, INDEX_PROPERTIES_PATTERN, DEFAULT_INDEX_PROPERTIES); indexPersistence.recover(indexPath, journalSnapshot.minIndex() * INDEX_STORAGE_SIZE, indexProperties); indexPersistence.truncate(indexPersistence.max() - indexPersistence.max() % INDEX_STORAGE_SIZE); truncateExtraIndices(); buildMissingIndices(); checkAndSetCommitIndex(commitIndex); recoverPartitions(partitionPath, journalSnapshot.partitionMinIndices(), indexProperties); flush(); logger.debug("Journal recovered, minIndex: {}, maxIndex: {}, partitions: {}, path: {}.", minIndex(), maxIndex(), partitionMap.keySet(), path.toAbsolutePath().toString()); }
Journal implements RaftJournal, Flushable, Closeable { public void recover(Path path, long commitIndex, JournalSnapshot journalSnapshot, Properties properties) throws IOException { this.basePath = path; Path indexPath = path.resolve(INDEX_PATH); Path partitionPath = path.resolve(PARTITION_PATH); journalProperties = replacePropertiesNames(properties, JOURNAL_PROPERTIES_PATTERN, DEFAULT_JOURNAL_PROPERTIES); journalPersistence.recover(path, journalSnapshot.minOffset(), journalProperties); truncateJournalTailPartialEntry(); indexProperties = replacePropertiesNames(properties, INDEX_PROPERTIES_PATTERN, DEFAULT_INDEX_PROPERTIES); indexPersistence.recover(indexPath, journalSnapshot.minIndex() * INDEX_STORAGE_SIZE, indexProperties); indexPersistence.truncate(indexPersistence.max() - indexPersistence.max() % INDEX_STORAGE_SIZE); truncateExtraIndices(); buildMissingIndices(); checkAndSetCommitIndex(commitIndex); recoverPartitions(partitionPath, journalSnapshot.partitionMinIndices(), indexProperties); flush(); logger.debug("Journal recovered, minIndex: {}, maxIndex: {}, partitions: {}, path: {}.", minIndex(), maxIndex(), partitionMap.keySet(), path.toAbsolutePath().toString()); } }
Journal implements RaftJournal, Flushable, Closeable { public void recover(Path path, long commitIndex, JournalSnapshot journalSnapshot, Properties properties) throws IOException { this.basePath = path; Path indexPath = path.resolve(INDEX_PATH); Path partitionPath = path.resolve(PARTITION_PATH); journalProperties = replacePropertiesNames(properties, JOURNAL_PROPERTIES_PATTERN, DEFAULT_JOURNAL_PROPERTIES); journalPersistence.recover(path, journalSnapshot.minOffset(), journalProperties); truncateJournalTailPartialEntry(); indexProperties = replacePropertiesNames(properties, INDEX_PROPERTIES_PATTERN, DEFAULT_INDEX_PROPERTIES); indexPersistence.recover(indexPath, journalSnapshot.minIndex() * INDEX_STORAGE_SIZE, indexProperties); indexPersistence.truncate(indexPersistence.max() - indexPersistence.max() % INDEX_STORAGE_SIZE); truncateExtraIndices(); buildMissingIndices(); checkAndSetCommitIndex(commitIndex); recoverPartitions(partitionPath, journalSnapshot.partitionMinIndices(), indexProperties); flush(); logger.debug("Journal recovered, minIndex: {}, maxIndex: {}, partitions: {}, path: {}.", minIndex(), maxIndex(), partitionMap.keySet(), path.toAbsolutePath().toString()); } Journal(PersistenceFactory persistenceFactory, BufferPool bufferPool, JournalEntryParser journalEntryParser); }
Journal implements RaftJournal, Flushable, Closeable { public void recover(Path path, long commitIndex, JournalSnapshot journalSnapshot, Properties properties) throws IOException { this.basePath = path; Path indexPath = path.resolve(INDEX_PATH); Path partitionPath = path.resolve(PARTITION_PATH); journalProperties = replacePropertiesNames(properties, JOURNAL_PROPERTIES_PATTERN, DEFAULT_JOURNAL_PROPERTIES); journalPersistence.recover(path, journalSnapshot.minOffset(), journalProperties); truncateJournalTailPartialEntry(); indexProperties = replacePropertiesNames(properties, INDEX_PROPERTIES_PATTERN, DEFAULT_INDEX_PROPERTIES); indexPersistence.recover(indexPath, journalSnapshot.minIndex() * INDEX_STORAGE_SIZE, indexProperties); indexPersistence.truncate(indexPersistence.max() - indexPersistence.max() % INDEX_STORAGE_SIZE); truncateExtraIndices(); buildMissingIndices(); checkAndSetCommitIndex(commitIndex); recoverPartitions(partitionPath, journalSnapshot.partitionMinIndices(), indexProperties); flush(); logger.debug("Journal recovered, minIndex: {}, maxIndex: {}, partitions: {}, path: {}.", minIndex(), maxIndex(), partitionMap.keySet(), path.toAbsolutePath().toString()); } Journal(PersistenceFactory persistenceFactory, BufferPool bufferPool, JournalEntryParser journalEntryParser); @Override long minIndex(); @Override long maxIndex(); @Override long minIndex(int partition); @Override long maxIndex(int partition); void compact(JournalSnapshot journalSnapshot); long append(JournalEntry entry); void clear(JournalSnapshot snapshot); List<Long> append(List<JournalEntry> entries); void commit(long index); JournalEntry readEntryHeaderByOffset(long offset); long commitIndex(); long appendBatchRaw(List<byte[]> storageEntries); @Override JournalEntry readByPartition(int partition, long index); @Override List<JournalEntry> batchReadByPartition(int partition, long startPartitionIndex, int maxSize); JournalEntry read(long index); JournalEntry readByOffset(long offset); long readOffset(long index); List<JournalEntry> batchRead(long index, int size); List<byte[]> readRaw(long index, int size); int getTerm(long index); void compareOrAppendRaw(List<byte[]> rawEntries, long startIndex); void recover(Path path, long commitIndex, JournalSnapshot journalSnapshot, Properties properties); @Override void flush(); void rePartition(Set<Integer> partitions); @Override long queryIndexByTimestamp(int partition, long timestamp); @Override Set<Integer> getPartitions(); void addPartition(int partition); void removePartition(int partition); @Override void close(); long flushedIndex(); JournalPersistence getJournalPersistence(); JournalPersistence getIndexPersistence(); Map<Integer, JournalPersistence> getPartitionMap(); long maxOffset(); Map<Integer, Long> calcPartitionIndices(long journalOffset); @Override String toString(); }
Journal implements RaftJournal, Flushable, Closeable { public void recover(Path path, long commitIndex, JournalSnapshot journalSnapshot, Properties properties) throws IOException { this.basePath = path; Path indexPath = path.resolve(INDEX_PATH); Path partitionPath = path.resolve(PARTITION_PATH); journalProperties = replacePropertiesNames(properties, JOURNAL_PROPERTIES_PATTERN, DEFAULT_JOURNAL_PROPERTIES); journalPersistence.recover(path, journalSnapshot.minOffset(), journalProperties); truncateJournalTailPartialEntry(); indexProperties = replacePropertiesNames(properties, INDEX_PROPERTIES_PATTERN, DEFAULT_INDEX_PROPERTIES); indexPersistence.recover(indexPath, journalSnapshot.minIndex() * INDEX_STORAGE_SIZE, indexProperties); indexPersistence.truncate(indexPersistence.max() - indexPersistence.max() % INDEX_STORAGE_SIZE); truncateExtraIndices(); buildMissingIndices(); checkAndSetCommitIndex(commitIndex); recoverPartitions(partitionPath, journalSnapshot.partitionMinIndices(), indexProperties); flush(); logger.debug("Journal recovered, minIndex: {}, maxIndex: {}, partitions: {}, path: {}.", minIndex(), maxIndex(), partitionMap.keySet(), path.toAbsolutePath().toString()); } Journal(PersistenceFactory persistenceFactory, BufferPool bufferPool, JournalEntryParser journalEntryParser); @Override long minIndex(); @Override long maxIndex(); @Override long minIndex(int partition); @Override long maxIndex(int partition); void compact(JournalSnapshot journalSnapshot); long append(JournalEntry entry); void clear(JournalSnapshot snapshot); List<Long> append(List<JournalEntry> entries); void commit(long index); JournalEntry readEntryHeaderByOffset(long offset); long commitIndex(); long appendBatchRaw(List<byte[]> storageEntries); @Override JournalEntry readByPartition(int partition, long index); @Override List<JournalEntry> batchReadByPartition(int partition, long startPartitionIndex, int maxSize); JournalEntry read(long index); JournalEntry readByOffset(long offset); long readOffset(long index); List<JournalEntry> batchRead(long index, int size); List<byte[]> readRaw(long index, int size); int getTerm(long index); void compareOrAppendRaw(List<byte[]> rawEntries, long startIndex); void recover(Path path, long commitIndex, JournalSnapshot journalSnapshot, Properties properties); @Override void flush(); void rePartition(Set<Integer> partitions); @Override long queryIndexByTimestamp(int partition, long timestamp); @Override Set<Integer> getPartitions(); void addPartition(int partition); void removePartition(int partition); @Override void close(); long flushedIndex(); JournalPersistence getJournalPersistence(); JournalPersistence getIndexPersistence(); Map<Integer, JournalPersistence> getPartitionMap(); long maxOffset(); Map<Integer, Long> calcPartitionIndices(long journalOffset); @Override String toString(); static final int INDEX_STORAGE_SIZE; }