@Test public void testRecover() throws Exception { String topic = "recoverTopic"; MessageBody = StoreMessage.getBytes(); for (int i = 0; i < 100; i++) { MessageExtBrokerInner messageExtBrokerInner = buildMessage(); messageExtBrokerInner.setTopic(topic); messageExtBrokerInner.setQueueId(0); messageStore.putMessage(messageExtBrokerInner); } StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore); long maxPhyOffset = messageStore.getMaxPhyOffset(); long maxCqOffset = messageStore.getMaxOffsetInQueue(topic, 0); messageStore.shutdown(); messageStore = buildMessageStore(); boolean load = messageStore.load(); assertTrue(load); messageStore.start(); assertTrue(maxPhyOffset == messageStore.getMaxPhyOffset()); assertTrue(maxCqOffset == messageStore.getMaxOffsetInQueue(topic, 0)); for (int i = 0; i < 100; i++) { MessageExtBrokerInner messageExtBrokerInner = buildMessage(); messageExtBrokerInner.setTopic(topic); messageExtBrokerInner.setQueueId(0); messageStore.putMessage(messageExtBrokerInner); } StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore); long secondLastPhyOffset = messageStore.getMaxPhyOffset(); long secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0); MessageExtBrokerInner messageExtBrokerInner = buildMessage(); messageExtBrokerInner.setTopic(topic); messageExtBrokerInner.setQueueId(0); messageStore.putMessage(messageExtBrokerInner); messageStore.shutdown(); damageCommitlog(secondLastPhyOffset); messageStore = buildMessageStore(); load = messageStore.load(); assertTrue(load); messageStore.start(); assertTrue(secondLastPhyOffset == messageStore.getMaxPhyOffset()); assertTrue(secondLastCqOffset == messageStore.getMaxOffsetInQueue(topic, 0)); for (int i = 0; i < 100; i++) { messageExtBrokerInner = buildMessage(); messageExtBrokerInner.setTopic(topic); messageExtBrokerInner.setQueueId(0); messageStore.putMessage(messageExtBrokerInner); } StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore); secondLastPhyOffset = messageStore.getMaxPhyOffset(); secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0); messageExtBrokerInner = buildMessage(); messageExtBrokerInner.setTopic(topic); messageExtBrokerInner.setQueueId(0); messageStore.putMessage(messageExtBrokerInner); messageStore.shutdown(); damageCommitlog(secondLastPhyOffset); String fileName = StorePathConfigHelper.getAbortFile(((DefaultMessageStore) messageStore).getMessageStoreConfig().getStorePathRootDir()); File file = new File(fileName); MappedFile.ensureDirOK(file.getParent()); file.createNewFile(); messageStore = buildMessageStore(); load = messageStore.load(); assertTrue(load); messageStore.start(); assertTrue(secondLastPhyOffset == messageStore.getMaxPhyOffset()); assertTrue(secondLastCqOffset == messageStore.getMaxOffsetInQueue(topic, 0)); for (int i = 0; i < 100; i++) { messageExtBrokerInner = buildMessage(); messageExtBrokerInner.setTopic(topic); messageExtBrokerInner.setQueueId(0); messageStore.putMessage(messageExtBrokerInner); } }
private void recover(final boolean lastExitOK) { long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue(); if (lastExitOK) { this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue); } else { this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue); } this.recoverTopicQueueTable(); }
DefaultMessageStore implements MessageStore { private void recover(final boolean lastExitOK) { long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue(); if (lastExitOK) { this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue); } else { this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue); } this.recoverTopicQueueTable(); } }
DefaultMessageStore implements MessageStore { private void recover(final boolean lastExitOK) { long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue(); if (lastExitOK) { this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue); } else { this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue); } this.recoverTopicQueueTable(); } DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig); }
DefaultMessageStore implements MessageStore { private void recover(final boolean lastExitOK) { long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue(); if (lastExitOK) { this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue); } else { this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue); } this.recoverTopicQueueTable(); } 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 { private void recover(final boolean lastExitOK) { long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue(); if (lastExitOK) { this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue); } else { this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue); } this.recoverTopicQueueTable(); } 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);  }