@Test public void readByPartitionTest() throws IOException, InterruptedException { int maxLength = 1024; int size = 1024; List<byte[]> entries = ByteUtils.createRandomSizeByteList(maxLength, size); List<JournalEntry> storageEntries = entries.stream() .map(entry -> journalEntryParser.createJournalEntry(entry)) .collect(Collectors.toList()); Map<Integer, List<JournalEntry>> partitionEntries = partitions.stream().collect(Collectors.toMap(p -> p, p -> new ArrayList<>())); 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); partitionEntries.get(partition).add(entry); } for (JournalEntry storageEntry : storageEntries) { journal.append(storageEntry); } journal.commit(journal.maxIndex()); long commitIndex = journal.commitIndex(); journal.flush(); journal.close(); journal = createJournal(commitIndex); for (Map.Entry<Integer, List<JournalEntry>> entry : partitionEntries.entrySet()) { Integer partition = entry.getKey(); List<JournalEntry> pEntries = entry.getValue(); Assert.assertEquals(pEntries.size(), journal.maxIndex(partition)); Assert.assertEquals(0, journal.minIndex(partition)); for (int i = 0; i < pEntries.size(); i++) { JournalEntry pEntry = pEntries.get(i); JournalEntry batchEntries = journal.readByPartition(partition, i); Assert.assertEquals(pEntry.getBatchSize(), batchEntries.getBatchSize()); Assert.assertEquals(0, batchEntries.getOffset()); Assert.assertEquals(pEntry.getPayload(), batchEntries.getPayload()); } } }
@Override public JournalEntry readByPartition(int partition, long index) { JournalPersistence pp = getPartitionPersistence(partition); long offset = readOffset(pp, index); long journalOffset; int relIndex; if (offset < 0) { journalOffset = readOffset(pp, index + offset); relIndex = (int) (-1 * offset); } else { journalOffset = offset; relIndex = 0; } JournalEntry journal = readByOffset(journalOffset); journal.setOffset(relIndex); return journal; }
Journal implements RaftJournal, Flushable, Closeable { @Override public JournalEntry readByPartition(int partition, long index) { JournalPersistence pp = getPartitionPersistence(partition); long offset = readOffset(pp, index); long journalOffset; int relIndex; if (offset < 0) { journalOffset = readOffset(pp, index + offset); relIndex = (int) (-1 * offset); } else { journalOffset = offset; relIndex = 0; } JournalEntry journal = readByOffset(journalOffset); journal.setOffset(relIndex); return journal; } }
Journal implements RaftJournal, Flushable, Closeable { @Override public JournalEntry readByPartition(int partition, long index) { JournalPersistence pp = getPartitionPersistence(partition); long offset = readOffset(pp, index); long journalOffset; int relIndex; if (offset < 0) { journalOffset = readOffset(pp, index + offset); relIndex = (int) (-1 * offset); } else { journalOffset = offset; relIndex = 0; } JournalEntry journal = readByOffset(journalOffset); journal.setOffset(relIndex); return journal; } Journal(PersistenceFactory persistenceFactory, BufferPool bufferPool, JournalEntryParser journalEntryParser); }
Journal implements RaftJournal, Flushable, Closeable { @Override public JournalEntry readByPartition(int partition, long index) { JournalPersistence pp = getPartitionPersistence(partition); long offset = readOffset(pp, index); long journalOffset; int relIndex; if (offset < 0) { journalOffset = readOffset(pp, index + offset); relIndex = (int) (-1 * offset); } else { journalOffset = offset; relIndex = 0; } JournalEntry journal = readByOffset(journalOffset); journal.setOffset(relIndex); return journal; } 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 { @Override public JournalEntry readByPartition(int partition, long index) { JournalPersistence pp = getPartitionPersistence(partition); long offset = readOffset(pp, index); long journalOffset; int relIndex; if (offset < 0) { journalOffset = readOffset(pp, index + offset); relIndex = (int) (-1 * offset); } else { journalOffset = offset; relIndex = 0; } JournalEntry journal = readByOffset(journalOffset); journal.setOffset(relIndex); return journal; } 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; }