@Test public void compactTest() throws Exception { 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<JournalEntry> storageEntries = entries.stream() .map(entry -> journalEntryParser.createJournalEntry(entry)) .peek(entry -> entry.setTerm(term)) .peek(entry -> entry.setPartition(0)) .collect(Collectors.toList()); List<Integer> partitionList = new ArrayList<>(partitions); for (int i = 0; i < storageEntries.size(); i++) { int partition = partitionList.get(i % partitions.size()); JournalEntry entry = storageEntries.get(i); entry.setPartition(partition); } long maxIndex = 0L; List<JournalSnapshotImpl> snapshotList = new ArrayList<>(storageEntries.size()); for (JournalEntry storageEntry : storageEntries) { snapshotList.add(new JournalSnapshotImpl(journal)); maxIndex = journal.append(storageEntry); journal.commit(journal.maxIndex()); } Assert.assertEquals(size, maxIndex); Assert.assertEquals(size, journal.maxIndex()); Assert.assertEquals(0, journal.minIndex()); journal.commit(journal.maxIndex()); journal.flush(); journal.compact(snapshotList.get(4)); Assert.assertEquals(4, journal.minIndex()); journal.compact(snapshotList.get(5)); Assert.assertEquals(5, journal.minIndex()); for (int partition : partitions) { journal.readByPartition(partition, journal.minIndex(partition)); } journal.compact(snapshotList.get(12)); Assert.assertEquals(12, journal.minIndex()); }
public void compact(JournalSnapshot journalSnapshot) throws IOException { if (journalSnapshot.minIndex() <= minIndex()) { return; } if (journalSnapshot.minIndex() > commitIndex()) { throw new IllegalArgumentException( String.format("Given snapshot min index %s should less than commit index %s!", ThreadSafeFormat.formatWithComma(journalSnapshot.minIndex()), ThreadSafeFormat.formatWithComma(commitIndex()) ) ); } Map<Integer , Long > partitionMinIndices = journalSnapshot.partitionMinIndices(); synchronized (partitionMap) { Iterator<Map.Entry<Integer, JournalPersistence>> iterator = partitionMap.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, JournalPersistence> entry = iterator.next(); int partition = entry.getKey(); JournalPersistence partitionPersistence = entry.getValue(); if (partitionMinIndices.containsKey(partition)) { long minPartitionIndices = partitionMinIndices.get(entry.getKey()); partitionPersistence.compact(minPartitionIndices * INDEX_STORAGE_SIZE); } else { partitionPersistence.close(); partitionPersistence.delete(); iterator.remove(); } } for (Map.Entry<Integer, Long> entry : partitionMinIndices.entrySet()) { int partition = entry.getKey(); if (!partitionMap.containsKey(partition)) { JournalPersistence partitionPersistence = persistenceFactory.createJournalPersistenceInstance(); partitionPersistence.recover(basePath.resolve(PARTITION_PATH).resolve(String.valueOf(partition)), partitionMinIndices.get(partition) * INDEX_STORAGE_SIZE, indexProperties); partitionMap.put(partition, partitionPersistence); } } } indexPersistence.compact(journalSnapshot.minIndex() * INDEX_STORAGE_SIZE); journalPersistence.compact(journalSnapshot.minOffset()); }
Journal implements RaftJournal, Flushable, Closeable { public void compact(JournalSnapshot journalSnapshot) throws IOException { if (journalSnapshot.minIndex() <= minIndex()) { return; } if (journalSnapshot.minIndex() > commitIndex()) { throw new IllegalArgumentException( String.format("Given snapshot min index %s should less than commit index %s!", ThreadSafeFormat.formatWithComma(journalSnapshot.minIndex()), ThreadSafeFormat.formatWithComma(commitIndex()) ) ); } Map<Integer , Long > partitionMinIndices = journalSnapshot.partitionMinIndices(); synchronized (partitionMap) { Iterator<Map.Entry<Integer, JournalPersistence>> iterator = partitionMap.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, JournalPersistence> entry = iterator.next(); int partition = entry.getKey(); JournalPersistence partitionPersistence = entry.getValue(); if (partitionMinIndices.containsKey(partition)) { long minPartitionIndices = partitionMinIndices.get(entry.getKey()); partitionPersistence.compact(minPartitionIndices * INDEX_STORAGE_SIZE); } else { partitionPersistence.close(); partitionPersistence.delete(); iterator.remove(); } } for (Map.Entry<Integer, Long> entry : partitionMinIndices.entrySet()) { int partition = entry.getKey(); if (!partitionMap.containsKey(partition)) { JournalPersistence partitionPersistence = persistenceFactory.createJournalPersistenceInstance(); partitionPersistence.recover(basePath.resolve(PARTITION_PATH).resolve(String.valueOf(partition)), partitionMinIndices.get(partition) * INDEX_STORAGE_SIZE, indexProperties); partitionMap.put(partition, partitionPersistence); } } } indexPersistence.compact(journalSnapshot.minIndex() * INDEX_STORAGE_SIZE); journalPersistence.compact(journalSnapshot.minOffset()); } }
Journal implements RaftJournal, Flushable, Closeable { public void compact(JournalSnapshot journalSnapshot) throws IOException { if (journalSnapshot.minIndex() <= minIndex()) { return; } if (journalSnapshot.minIndex() > commitIndex()) { throw new IllegalArgumentException( String.format("Given snapshot min index %s should less than commit index %s!", ThreadSafeFormat.formatWithComma(journalSnapshot.minIndex()), ThreadSafeFormat.formatWithComma(commitIndex()) ) ); } Map<Integer , Long > partitionMinIndices = journalSnapshot.partitionMinIndices(); synchronized (partitionMap) { Iterator<Map.Entry<Integer, JournalPersistence>> iterator = partitionMap.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, JournalPersistence> entry = iterator.next(); int partition = entry.getKey(); JournalPersistence partitionPersistence = entry.getValue(); if (partitionMinIndices.containsKey(partition)) { long minPartitionIndices = partitionMinIndices.get(entry.getKey()); partitionPersistence.compact(minPartitionIndices * INDEX_STORAGE_SIZE); } else { partitionPersistence.close(); partitionPersistence.delete(); iterator.remove(); } } for (Map.Entry<Integer, Long> entry : partitionMinIndices.entrySet()) { int partition = entry.getKey(); if (!partitionMap.containsKey(partition)) { JournalPersistence partitionPersistence = persistenceFactory.createJournalPersistenceInstance(); partitionPersistence.recover(basePath.resolve(PARTITION_PATH).resolve(String.valueOf(partition)), partitionMinIndices.get(partition) * INDEX_STORAGE_SIZE, indexProperties); partitionMap.put(partition, partitionPersistence); } } } indexPersistence.compact(journalSnapshot.minIndex() * INDEX_STORAGE_SIZE); journalPersistence.compact(journalSnapshot.minOffset()); } Journal(PersistenceFactory persistenceFactory, BufferPool bufferPool, JournalEntryParser journalEntryParser); }
Journal implements RaftJournal, Flushable, Closeable { public void compact(JournalSnapshot journalSnapshot) throws IOException { if (journalSnapshot.minIndex() <= minIndex()) { return; } if (journalSnapshot.minIndex() > commitIndex()) { throw new IllegalArgumentException( String.format("Given snapshot min index %s should less than commit index %s!", ThreadSafeFormat.formatWithComma(journalSnapshot.minIndex()), ThreadSafeFormat.formatWithComma(commitIndex()) ) ); } Map<Integer , Long > partitionMinIndices = journalSnapshot.partitionMinIndices(); synchronized (partitionMap) { Iterator<Map.Entry<Integer, JournalPersistence>> iterator = partitionMap.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, JournalPersistence> entry = iterator.next(); int partition = entry.getKey(); JournalPersistence partitionPersistence = entry.getValue(); if (partitionMinIndices.containsKey(partition)) { long minPartitionIndices = partitionMinIndices.get(entry.getKey()); partitionPersistence.compact(minPartitionIndices * INDEX_STORAGE_SIZE); } else { partitionPersistence.close(); partitionPersistence.delete(); iterator.remove(); } } for (Map.Entry<Integer, Long> entry : partitionMinIndices.entrySet()) { int partition = entry.getKey(); if (!partitionMap.containsKey(partition)) { JournalPersistence partitionPersistence = persistenceFactory.createJournalPersistenceInstance(); partitionPersistence.recover(basePath.resolve(PARTITION_PATH).resolve(String.valueOf(partition)), partitionMinIndices.get(partition) * INDEX_STORAGE_SIZE, indexProperties); partitionMap.put(partition, partitionPersistence); } } } indexPersistence.compact(journalSnapshot.minIndex() * INDEX_STORAGE_SIZE); journalPersistence.compact(journalSnapshot.minOffset()); } 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 compact(JournalSnapshot journalSnapshot) throws IOException { if (journalSnapshot.minIndex() <= minIndex()) { return; } if (journalSnapshot.minIndex() > commitIndex()) { throw new IllegalArgumentException( String.format("Given snapshot min index %s should less than commit index %s!", ThreadSafeFormat.formatWithComma(journalSnapshot.minIndex()), ThreadSafeFormat.formatWithComma(commitIndex()) ) ); } Map<Integer , Long > partitionMinIndices = journalSnapshot.partitionMinIndices(); synchronized (partitionMap) { Iterator<Map.Entry<Integer, JournalPersistence>> iterator = partitionMap.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, JournalPersistence> entry = iterator.next(); int partition = entry.getKey(); JournalPersistence partitionPersistence = entry.getValue(); if (partitionMinIndices.containsKey(partition)) { long minPartitionIndices = partitionMinIndices.get(entry.getKey()); partitionPersistence.compact(minPartitionIndices * INDEX_STORAGE_SIZE); } else { partitionPersistence.close(); partitionPersistence.delete(); iterator.remove(); } } for (Map.Entry<Integer, Long> entry : partitionMinIndices.entrySet()) { int partition = entry.getKey(); if (!partitionMap.containsKey(partition)) { JournalPersistence partitionPersistence = persistenceFactory.createJournalPersistenceInstance(); partitionPersistence.recover(basePath.resolve(PARTITION_PATH).resolve(String.valueOf(partition)), partitionMinIndices.get(partition) * INDEX_STORAGE_SIZE, indexProperties); partitionMap.put(partition, partitionPersistence); } } } indexPersistence.compact(journalSnapshot.minIndex() * INDEX_STORAGE_SIZE); journalPersistence.compact(journalSnapshot.minOffset()); } 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; }