@Test public void getIndexTest() throws InterruptedException, IOException { long timeout = 500000L; List<ByteBuffer> msgs = MessageUtils.build(20, 255); long startTime = SystemClock.now(); short partition = partitions[0]; final EventFuture<WriteResult> future = new EventFuture<>(); for (int i = 0; i < 20; i++) { ByteBuffer msg = msgs.get(i); MessageParser.setLong(msg, MessageParser.CLIENT_TIMESTAMP, startTime); MessageParser.setShort(msg, MessageParser.PARTITION, partition); if (i < 10 || i >= 15) { store.asyncWrite(QosLevel.PERSISTENCE, future, new WriteRequest(partition, msgs.get(i))); } else if (i == 10) { store.asyncWrite(QosLevel.PERSISTENCE, future, new WriteRequest(partition, MessageUtils.toBatchMessage(msgs.get(i), (short) 5))); } Thread.sleep(100L); } store.commit(store.rightPosition()); long t0 = SystemClock.now(); while (SystemClock.now() - t0 < timeout && store.getRightIndex(partition) < 20) { Thread.sleep(10L); } long[] storageTimestamps = Arrays.stream(store.read(partition, 0L, 20, 0).getMessages()) .mapToLong(b -> MessageParser.getInt(b, MessageParser.STORAGE_TIMESTAMP)).toArray(); Assert.assertEquals(0L, store.getIndex(partition, startTime - 1L)); Assert.assertEquals(0L, store.getIndex(partition, startTime + storageTimestamps[0])); Assert.assertEquals(4L, store.getIndex(partition, startTime + storageTimestamps[3] + 10L)); Assert.assertEquals(4L, store.getIndex(partition, startTime + storageTimestamps[4])); Assert.assertEquals(10L, store.getIndex(partition, startTime + storageTimestamps[10])); Assert.assertEquals(15L, store.getIndex(partition, startTime + storageTimestamps[10] + 10L)); Assert.assertEquals(19L, store.getIndex(partition, startTime + storageTimestamps[19 - 4])); Assert.assertEquals(-1L, store.getIndex(partition, startTime + storageTimestamps[19 - 4] + 10L)); }
public long getIndex(short partition, long timestamp) { try { if (partitionMap.containsKey(partition)) { PositioningStore<IndexItem> indexStore = partitionMap.get(partition).store; long searchedIndex = binarySearchByTimestamp(timestamp, store, indexStore, indexStore.left() / IndexItem.STORAGE_SIZE, indexStore.right() / IndexItem.STORAGE_SIZE - 1); while (searchedIndex - 1 >= indexStore.left() && timestamp <= getStorageTimestamp(store, indexStore, searchedIndex - 1)) { searchedIndex--; } return searchedIndex; } } catch (PositionOverflowException | PositionUnderflowException | IOException e) { logger.warn("Exception: ", e); } return -1L; }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { public long getIndex(short partition, long timestamp) { try { if (partitionMap.containsKey(partition)) { PositioningStore<IndexItem> indexStore = partitionMap.get(partition).store; long searchedIndex = binarySearchByTimestamp(timestamp, store, indexStore, indexStore.left() / IndexItem.STORAGE_SIZE, indexStore.right() / IndexItem.STORAGE_SIZE - 1); while (searchedIndex - 1 >= indexStore.left() && timestamp <= getStorageTimestamp(store, indexStore, searchedIndex - 1)) { searchedIndex--; } return searchedIndex; } } catch (PositionOverflowException | PositionUnderflowException | IOException e) { logger.warn("Exception: ", e); } return -1L; } }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { public long getIndex(short partition, long timestamp) { try { if (partitionMap.containsKey(partition)) { PositioningStore<IndexItem> indexStore = partitionMap.get(partition).store; long searchedIndex = binarySearchByTimestamp(timestamp, store, indexStore, indexStore.left() / IndexItem.STORAGE_SIZE, indexStore.right() / IndexItem.STORAGE_SIZE - 1); while (searchedIndex - 1 >= indexStore.left() && timestamp <= getStorageTimestamp(store, indexStore, searchedIndex - 1)) { searchedIndex--; } return searchedIndex; } } catch (PositionOverflowException | PositionUnderflowException | IOException e) { logger.warn("Exception: ", e); } return -1L; } PartitionGroupStoreManager(String topic, int partitionGroup, File base, Config config,
                                      PreloadBufferPool bufferPool); }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { public long getIndex(short partition, long timestamp) { try { if (partitionMap.containsKey(partition)) { PositioningStore<IndexItem> indexStore = partitionMap.get(partition).store; long searchedIndex = binarySearchByTimestamp(timestamp, store, indexStore, indexStore.left() / IndexItem.STORAGE_SIZE, indexStore.right() / IndexItem.STORAGE_SIZE - 1); while (searchedIndex - 1 >= indexStore.left() && timestamp <= getStorageTimestamp(store, indexStore, searchedIndex - 1)) { searchedIndex--; } return searchedIndex; } } catch (PositionOverflowException | PositionUnderflowException | IOException e) { logger.warn("Exception: ", e); } return -1L; } PartitionGroupStoreManager(String topic, int partitionGroup, File base, Config config,
                                      PreloadBufferPool bufferPool); void recover(); String getTopic(); int getPartitionGroup(); ReadResult read(short partition, long index, int count, long maxSize); long getLeftIndex(short partition); long getLeftIndexAndCheck(short partition); long getRightIndex(short partition); long getRightIndexAndCheck(short partition); @Override boolean serviceStatus(); @Override void enable(); @Override void disable(); @Override void setRightPosition(long position); @Override void clear(long position); @Override long commitPosition(); @Override int term(); @Override void term(int term); @Override ByteBuffer readEntryBuffer(long position, int length); @Override long appendEntryBuffer(ByteBuffer byteBuffer); @Override long position(long position, int offsetCount); @Override int lastEntryTerm(); @Override void commit(long position); @Override int getEntryTerm(long position); @Override long leftPosition(); long rightPosition(); @Override void close(); long getIndex(short partition, long timestamp); }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { public long getIndex(short partition, long timestamp) { try { if (partitionMap.containsKey(partition)) { PositioningStore<IndexItem> indexStore = partitionMap.get(partition).store; long searchedIndex = binarySearchByTimestamp(timestamp, store, indexStore, indexStore.left() / IndexItem.STORAGE_SIZE, indexStore.right() / IndexItem.STORAGE_SIZE - 1); while (searchedIndex - 1 >= indexStore.left() && timestamp <= getStorageTimestamp(store, indexStore, searchedIndex - 1)) { searchedIndex--; } return searchedIndex; } } catch (PositionOverflowException | PositionUnderflowException | IOException e) { logger.warn("Exception: ", e); } return -1L; } PartitionGroupStoreManager(String topic, int partitionGroup, File base, Config config,
                                      PreloadBufferPool bufferPool); void recover(); String getTopic(); int getPartitionGroup(); ReadResult read(short partition, long index, int count, long maxSize); long getLeftIndex(short partition); long getLeftIndexAndCheck(short partition); long getRightIndex(short partition); long getRightIndexAndCheck(short partition); @Override boolean serviceStatus(); @Override void enable(); @Override void disable(); @Override void setRightPosition(long position); @Override void clear(long position); @Override long commitPosition(); @Override int term(); @Override void term(int term); @Override ByteBuffer readEntryBuffer(long position, int length); @Override long appendEntryBuffer(ByteBuffer byteBuffer); @Override long position(long position, int offsetCount); @Override int lastEntryTerm(); @Override void commit(long position); @Override int getEntryTerm(long position); @Override long leftPosition(); long rightPosition(); @Override void close(); long getIndex(short partition, long timestamp);  }