@Test public void testGetOffsetInQueueByTime() { final int totalCount = 10; int queueId = 0; String topic = "FooBar"; AppendMessageResult[] appendMessageResults = putMessages(totalCount, topic, queueId, true); StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore); ConsumeQueue consumeQueue = getDefaultMessageStore().findConsumeQueue(topic, queueId); for (AppendMessageResult appendMessageResult : appendMessageResults) { long offset = messageStore.getOffsetInQueueByTime(topic, queueId, appendMessageResult.getStoreTimestamp()); SelectMappedBufferResult indexBuffer = consumeQueue.getIndexBuffer(offset); assertThat(indexBuffer.getByteBuffer().getLong()).isEqualTo(appendMessageResult.getWroteOffset()); assertThat(indexBuffer.getByteBuffer().getInt()).isEqualTo(appendMessageResult.getWroteBytes()); indexBuffer.release(); } }
public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) { ConsumeQueue logic = this.findConsumeQueue(topic, queueId); if (logic != null) { return logic.getOffsetInQueueByTime(timestamp); } return 0; }
DefaultMessageStore implements MessageStore { public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) { ConsumeQueue logic = this.findConsumeQueue(topic, queueId); if (logic != null) { return logic.getOffsetInQueueByTime(timestamp); } return 0; } }
DefaultMessageStore implements MessageStore { public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) { ConsumeQueue logic = this.findConsumeQueue(topic, queueId); if (logic != null) { return logic.getOffsetInQueueByTime(timestamp); } return 0; } DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig); }
DefaultMessageStore implements MessageStore { public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) { ConsumeQueue logic = this.findConsumeQueue(topic, queueId); if (logic != null) { return logic.getOffsetInQueueByTime(timestamp); } return 0; } DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig); void truncateDirtyLogicFiles(long phyOffset); boolean load(); void start(); void shutdown(); void destroy(); void destroyLogics(); @Override CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg); CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch); @Override PutMessageResult putMessage(MessageExtBrokerInner msg); @Override PutMessageResult putMessages(MessageExtBatch messageExtBatch); @Override boolean isOSPageCacheBusy(); @Override long lockTimeMills(); SystemClock getSystemClock(); CommitLog getCommitLog(); GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,
        final int maxMsgNums,
        final MessageFilter messageFilter); long getMaxOffsetInQueue(String topic, int queueId); long getMinOffsetInQueue(String topic, int queueId); @Override long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQueueOffset); long getOffsetInQueueByTime(String topic, int queueId, long timestamp); MessageExt lookMessageByOffset(long commitLogOffset); @Override SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset); @Override SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset, int msgSize); String getRunningDataInfo(); @Override HashMap<String, String> getRuntimeInfo(); @Override long getMaxPhyOffset(); @Override long getMinPhyOffset(); @Override long getEarliestMessageTime(String topic, int queueId); @Override long getEarliestMessageTime(); @Override long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset); @Override long getMessageTotalInQueue(String topic, int queueId); @Override SelectMappedBufferResult getCommitLogData(final long offset); @Override boolean appendToCommitLog(long startOffset, byte[] data); @Override void executeDeleteFilesManually(); @Override QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end); @Override void updateHaMasterAddress(String newAddr); @Override long slaveFallBehindMuch(); @Override long now(); @Override int cleanUnusedTopic(Set<String> topics); void cleanExpiredConsumerQueue(); Map<String, Long> getMessageIds(final String topic, final int queueId, long minOffset, long maxOffset,
        SocketAddress storeHost); @Override boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset); @Override long dispatchBehindBytes(); @Override long flush(); @Override boolean resetWriteOffset(long phyOffset); @Override long getConfirmOffset(); @Override void setConfirmOffset(long phyOffset); MessageExt lookMessageByOffset(long commitLogOffset, int size); ConsumeQueue findConsumeQueue(String topic, int queueId); MessageStoreConfig getMessageStoreConfig(); TransientStorePool getTransientStorePool(); void recoverTopicQueueTable(); AllocateMappedFileService getAllocateMappedFileService(); StoreStatsService getStoreStatsService(); RunningFlags getAccessRights(); ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> getConsumeQueueTable(); StoreCheckpoint getStoreCheckpoint(); HAService getHaService(); ScheduleMessageService getScheduleMessageService(); RunningFlags getRunningFlags(); void doDispatch(DispatchRequest req); void putMessagePositionInfo(DispatchRequest dispatchRequest); @Override BrokerStatsManager getBrokerStatsManager(); @Override void handleScheduleMessageService(final BrokerRole brokerRole); int remainTransientStoreBufferNumbs(); @Override boolean isTransientStorePoolDeficient(); @Override LinkedList<CommitLogDispatcher> getDispatcherList(); @Override ConsumeQueue getConsumeQueue(String topic, int queueId); void unlockMappedFile(final MappedFile mappedFile); }
DefaultMessageStore implements MessageStore { public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) { ConsumeQueue logic = this.findConsumeQueue(topic, queueId); if (logic != null) { return logic.getOffsetInQueueByTime(timestamp); } return 0; } DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig); void truncateDirtyLogicFiles(long phyOffset); boolean load(); void start(); void shutdown(); void destroy(); void destroyLogics(); @Override CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg); CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch); @Override PutMessageResult putMessage(MessageExtBrokerInner msg); @Override PutMessageResult putMessages(MessageExtBatch messageExtBatch); @Override boolean isOSPageCacheBusy(); @Override long lockTimeMills(); SystemClock getSystemClock(); CommitLog getCommitLog(); GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,
        final int maxMsgNums,
        final MessageFilter messageFilter); long getMaxOffsetInQueue(String topic, int queueId); long getMinOffsetInQueue(String topic, int queueId); @Override long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQueueOffset); long getOffsetInQueueByTime(String topic, int queueId, long timestamp); MessageExt lookMessageByOffset(long commitLogOffset); @Override SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset); @Override SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset, int msgSize); String getRunningDataInfo(); @Override HashMap<String, String> getRuntimeInfo(); @Override long getMaxPhyOffset(); @Override long getMinPhyOffset(); @Override long getEarliestMessageTime(String topic, int queueId); @Override long getEarliestMessageTime(); @Override long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset); @Override long getMessageTotalInQueue(String topic, int queueId); @Override SelectMappedBufferResult getCommitLogData(final long offset); @Override boolean appendToCommitLog(long startOffset, byte[] data); @Override void executeDeleteFilesManually(); @Override QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end); @Override void updateHaMasterAddress(String newAddr); @Override long slaveFallBehindMuch(); @Override long now(); @Override int cleanUnusedTopic(Set<String> topics); void cleanExpiredConsumerQueue(); Map<String, Long> getMessageIds(final String topic, final int queueId, long minOffset, long maxOffset,
        SocketAddress storeHost); @Override boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset); @Override long dispatchBehindBytes(); @Override long flush(); @Override boolean resetWriteOffset(long phyOffset); @Override long getConfirmOffset(); @Override void setConfirmOffset(long phyOffset); MessageExt lookMessageByOffset(long commitLogOffset, int size); ConsumeQueue findConsumeQueue(String topic, int queueId); MessageStoreConfig getMessageStoreConfig(); TransientStorePool getTransientStorePool(); void recoverTopicQueueTable(); AllocateMappedFileService getAllocateMappedFileService(); StoreStatsService getStoreStatsService(); RunningFlags getAccessRights(); ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> getConsumeQueueTable(); StoreCheckpoint getStoreCheckpoint(); HAService getHaService(); ScheduleMessageService getScheduleMessageService(); RunningFlags getRunningFlags(); void doDispatch(DispatchRequest req); void putMessagePositionInfo(DispatchRequest dispatchRequest); @Override BrokerStatsManager getBrokerStatsManager(); @Override void handleScheduleMessageService(final BrokerRole brokerRole); int remainTransientStoreBufferNumbs(); @Override boolean isTransientStorePoolDeficient(); @Override LinkedList<CommitLogDispatcher> getDispatcherList(); @Override ConsumeQueue getConsumeQueue(String topic, int queueId); void unlockMappedFile(final MappedFile mappedFile);  }