@Test public void testRecover() throws Exception { String base = createBaseDir(); String peers = String.format("n0-localhost:%d", nextPort()); String group = UUID.randomUUID().toString(); String topic = UUID.randomUUID().toString(); { DefaultMessageStore messageStore = createDledgerMessageStore(base, group, "n0", peers, null, false, 0); Thread.sleep(1000); doPutMessages(messageStore, topic, 0, 1000, 0); Thread.sleep(100); Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, 0)); Assert.assertEquals(1000, messageStore.getMaxOffsetInQueue(topic, 0)); Assert.assertEquals(0, messageStore.dispatchBehindBytes()); doGetMessages(messageStore, topic, 0, 1000, 0); messageStore.shutdown(); } { DefaultMessageStore messageStore = createDledgerMessageStore(base, group, "n0", peers, null, false, 0); Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, 0)); Assert.assertEquals(1000, messageStore.getMaxOffsetInQueue(topic, 0)); Assert.assertEquals(0, messageStore.dispatchBehindBytes()); doGetMessages(messageStore, topic, 0, 1000, 0); messageStore.shutdown(); } { DefaultMessageStore messageStore = createDledgerMessageStore(base, group, "n0", peers, null, true, 0); Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, 0)); Assert.assertEquals(1000, messageStore.getMaxOffsetInQueue(topic, 0)); Assert.assertEquals(0, messageStore.dispatchBehindBytes()); doGetMessages(messageStore, topic, 0, 1000, 0); messageStore.shutdown(); } }
private void recover(long maxPhyOffsetOfConsumeQueue) { dLedgerFileStore.load(); if (dLedgerFileList.getMappedFiles().size() > 0) { dLedgerFileStore.recover(); dividedCommitlogOffset = dLedgerFileList.getFirstMappedFile().getFileFromOffset(); MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile != null) { disableDeleteDledger(); } long maxPhyOffset = dLedgerFileList.getMaxWrotePosition(); if (maxPhyOffsetOfConsumeQueue >= maxPhyOffset) { log.warn("[TruncateCQ]maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, maxPhyOffset); this.defaultMessageStore.truncateDirtyLogicFiles(maxPhyOffset); } return; } isInrecoveringOldCommitlog = true; super.recoverNormally(maxPhyOffsetOfConsumeQueue); isInrecoveringOldCommitlog = false; MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null) { return; } ByteBuffer byteBuffer = mappedFile.sliceByteBuffer(); byteBuffer.position(mappedFile.getWrotePosition()); boolean needWriteMagicCode = true; byteBuffer.getInt(); int magicCode = byteBuffer.getInt(); if (magicCode == CommitLog.BLANK_MAGIC_CODE) { needWriteMagicCode = false; } else { log.info("Recover old commitlog found a illegal magic code={}", magicCode); } dLedgerConfig.setEnableDiskForceClean(false); dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize(); log.info("Recover old commitlog needWriteMagicCode={} pos={} file={} dividedCommitlogOffset={}", needWriteMagicCode, mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(), mappedFile.getFileName(), dividedCommitlogOffset); if (needWriteMagicCode) { byteBuffer.position(mappedFile.getWrotePosition()); byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition()); byteBuffer.putInt(BLANK_MAGIC_CODE); mappedFile.flush(0); } mappedFile.setWrotePosition(mappedFile.getFileSize()); mappedFile.setCommittedPosition(mappedFile.getFileSize()); mappedFile.setFlushedPosition(mappedFile.getFileSize()); dLedgerFileList.getLastMappedFile(dividedCommitlogOffset); log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset); }
DLedgerCommitLog extends CommitLog { private void recover(long maxPhyOffsetOfConsumeQueue) { dLedgerFileStore.load(); if (dLedgerFileList.getMappedFiles().size() > 0) { dLedgerFileStore.recover(); dividedCommitlogOffset = dLedgerFileList.getFirstMappedFile().getFileFromOffset(); MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile != null) { disableDeleteDledger(); } long maxPhyOffset = dLedgerFileList.getMaxWrotePosition(); if (maxPhyOffsetOfConsumeQueue >= maxPhyOffset) { log.warn("[TruncateCQ]maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, maxPhyOffset); this.defaultMessageStore.truncateDirtyLogicFiles(maxPhyOffset); } return; } isInrecoveringOldCommitlog = true; super.recoverNormally(maxPhyOffsetOfConsumeQueue); isInrecoveringOldCommitlog = false; MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null) { return; } ByteBuffer byteBuffer = mappedFile.sliceByteBuffer(); byteBuffer.position(mappedFile.getWrotePosition()); boolean needWriteMagicCode = true; byteBuffer.getInt(); int magicCode = byteBuffer.getInt(); if (magicCode == CommitLog.BLANK_MAGIC_CODE) { needWriteMagicCode = false; } else { log.info("Recover old commitlog found a illegal magic code={}", magicCode); } dLedgerConfig.setEnableDiskForceClean(false); dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize(); log.info("Recover old commitlog needWriteMagicCode={} pos={} file={} dividedCommitlogOffset={}", needWriteMagicCode, mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(), mappedFile.getFileName(), dividedCommitlogOffset); if (needWriteMagicCode) { byteBuffer.position(mappedFile.getWrotePosition()); byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition()); byteBuffer.putInt(BLANK_MAGIC_CODE); mappedFile.flush(0); } mappedFile.setWrotePosition(mappedFile.getFileSize()); mappedFile.setCommittedPosition(mappedFile.getFileSize()); mappedFile.setFlushedPosition(mappedFile.getFileSize()); dLedgerFileList.getLastMappedFile(dividedCommitlogOffset); log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset); } }
DLedgerCommitLog extends CommitLog { private void recover(long maxPhyOffsetOfConsumeQueue) { dLedgerFileStore.load(); if (dLedgerFileList.getMappedFiles().size() > 0) { dLedgerFileStore.recover(); dividedCommitlogOffset = dLedgerFileList.getFirstMappedFile().getFileFromOffset(); MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile != null) { disableDeleteDledger(); } long maxPhyOffset = dLedgerFileList.getMaxWrotePosition(); if (maxPhyOffsetOfConsumeQueue >= maxPhyOffset) { log.warn("[TruncateCQ]maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, maxPhyOffset); this.defaultMessageStore.truncateDirtyLogicFiles(maxPhyOffset); } return; } isInrecoveringOldCommitlog = true; super.recoverNormally(maxPhyOffsetOfConsumeQueue); isInrecoveringOldCommitlog = false; MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null) { return; } ByteBuffer byteBuffer = mappedFile.sliceByteBuffer(); byteBuffer.position(mappedFile.getWrotePosition()); boolean needWriteMagicCode = true; byteBuffer.getInt(); int magicCode = byteBuffer.getInt(); if (magicCode == CommitLog.BLANK_MAGIC_CODE) { needWriteMagicCode = false; } else { log.info("Recover old commitlog found a illegal magic code={}", magicCode); } dLedgerConfig.setEnableDiskForceClean(false); dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize(); log.info("Recover old commitlog needWriteMagicCode={} pos={} file={} dividedCommitlogOffset={}", needWriteMagicCode, mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(), mappedFile.getFileName(), dividedCommitlogOffset); if (needWriteMagicCode) { byteBuffer.position(mappedFile.getWrotePosition()); byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition()); byteBuffer.putInt(BLANK_MAGIC_CODE); mappedFile.flush(0); } mappedFile.setWrotePosition(mappedFile.getFileSize()); mappedFile.setCommittedPosition(mappedFile.getFileSize()); mappedFile.setFlushedPosition(mappedFile.getFileSize()); dLedgerFileList.getLastMappedFile(dividedCommitlogOffset); log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset); } DLedgerCommitLog(final DefaultMessageStore defaultMessageStore); }
DLedgerCommitLog extends CommitLog { private void recover(long maxPhyOffsetOfConsumeQueue) { dLedgerFileStore.load(); if (dLedgerFileList.getMappedFiles().size() > 0) { dLedgerFileStore.recover(); dividedCommitlogOffset = dLedgerFileList.getFirstMappedFile().getFileFromOffset(); MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile != null) { disableDeleteDledger(); } long maxPhyOffset = dLedgerFileList.getMaxWrotePosition(); if (maxPhyOffsetOfConsumeQueue >= maxPhyOffset) { log.warn("[TruncateCQ]maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, maxPhyOffset); this.defaultMessageStore.truncateDirtyLogicFiles(maxPhyOffset); } return; } isInrecoveringOldCommitlog = true; super.recoverNormally(maxPhyOffsetOfConsumeQueue); isInrecoveringOldCommitlog = false; MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null) { return; } ByteBuffer byteBuffer = mappedFile.sliceByteBuffer(); byteBuffer.position(mappedFile.getWrotePosition()); boolean needWriteMagicCode = true; byteBuffer.getInt(); int magicCode = byteBuffer.getInt(); if (magicCode == CommitLog.BLANK_MAGIC_CODE) { needWriteMagicCode = false; } else { log.info("Recover old commitlog found a illegal magic code={}", magicCode); } dLedgerConfig.setEnableDiskForceClean(false); dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize(); log.info("Recover old commitlog needWriteMagicCode={} pos={} file={} dividedCommitlogOffset={}", needWriteMagicCode, mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(), mappedFile.getFileName(), dividedCommitlogOffset); if (needWriteMagicCode) { byteBuffer.position(mappedFile.getWrotePosition()); byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition()); byteBuffer.putInt(BLANK_MAGIC_CODE); mappedFile.flush(0); } mappedFile.setWrotePosition(mappedFile.getFileSize()); mappedFile.setCommittedPosition(mappedFile.getFileSize()); mappedFile.setFlushedPosition(mappedFile.getFileSize()); dLedgerFileList.getLastMappedFile(dividedCommitlogOffset); log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset); } DLedgerCommitLog(final DefaultMessageStore defaultMessageStore); @Override boolean load(); @Override void start(); @Override void shutdown(); @Override long flush(); @Override long getMaxOffset(); @Override long getMinOffset(); @Override long getConfirmOffset(); @Override void setConfirmOffset(long phyOffset); @Override long remainHowManyDataToCommit(); @Override long remainHowManyDataToFlush(); @Override int deleteExpiredFile(
        final long expiredTime,
        final int deleteFilesInterval,
        final long intervalForcibly,
        final boolean cleanImmediately
    ); SelectMappedBufferResult convertSbr(SelectMmapBufferResult sbr); SelectMmapBufferResult truncate(SelectMmapBufferResult sbr); @Override SelectMappedBufferResult getData(final long offset); @Override SelectMappedBufferResult getData(final long offset, final boolean returnFirstOnNotFound); @Override void recoverNormally(long maxPhyOffsetOfConsumeQueue); @Override void recoverAbnormally(long maxPhyOffsetOfConsumeQueue); @Override DispatchRequest checkMessageAndReturnSize(ByteBuffer byteBuffer, final boolean checkCRC); @Override DispatchRequest checkMessageAndReturnSize(ByteBuffer byteBuffer, final boolean checkCRC,
        final boolean readBody); @Override boolean resetOffset(long offset); @Override long getBeginTimeInLock(); @Override PutMessageResult putMessage(final MessageExtBrokerInner msg); @Override PutMessageResult putMessages(final MessageExtBatch messageExtBatch); @Override CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg); @Override CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch); @Override SelectMappedBufferResult getMessage(final long offset, final int size); @Override long rollNextFile(final long offset); @Override HashMap<String, Long> getTopicQueueTable(); @Override void setTopicQueueTable(HashMap<String, Long> topicQueueTable); @Override void destroy(); @Override boolean appendData(long startOffset, byte[] data); @Override void checkSelf(); @Override long lockTimeMills(); DLedgerServer getdLedgerServer(); int getId(); long getDividedCommitlogOffset(); }
DLedgerCommitLog extends CommitLog { private void recover(long maxPhyOffsetOfConsumeQueue) { dLedgerFileStore.load(); if (dLedgerFileList.getMappedFiles().size() > 0) { dLedgerFileStore.recover(); dividedCommitlogOffset = dLedgerFileList.getFirstMappedFile().getFileFromOffset(); MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile != null) { disableDeleteDledger(); } long maxPhyOffset = dLedgerFileList.getMaxWrotePosition(); if (maxPhyOffsetOfConsumeQueue >= maxPhyOffset) { log.warn("[TruncateCQ]maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, maxPhyOffset); this.defaultMessageStore.truncateDirtyLogicFiles(maxPhyOffset); } return; } isInrecoveringOldCommitlog = true; super.recoverNormally(maxPhyOffsetOfConsumeQueue); isInrecoveringOldCommitlog = false; MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile == null) { return; } ByteBuffer byteBuffer = mappedFile.sliceByteBuffer(); byteBuffer.position(mappedFile.getWrotePosition()); boolean needWriteMagicCode = true; byteBuffer.getInt(); int magicCode = byteBuffer.getInt(); if (magicCode == CommitLog.BLANK_MAGIC_CODE) { needWriteMagicCode = false; } else { log.info("Recover old commitlog found a illegal magic code={}", magicCode); } dLedgerConfig.setEnableDiskForceClean(false); dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize(); log.info("Recover old commitlog needWriteMagicCode={} pos={} file={} dividedCommitlogOffset={}", needWriteMagicCode, mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(), mappedFile.getFileName(), dividedCommitlogOffset); if (needWriteMagicCode) { byteBuffer.position(mappedFile.getWrotePosition()); byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition()); byteBuffer.putInt(BLANK_MAGIC_CODE); mappedFile.flush(0); } mappedFile.setWrotePosition(mappedFile.getFileSize()); mappedFile.setCommittedPosition(mappedFile.getFileSize()); mappedFile.setFlushedPosition(mappedFile.getFileSize()); dLedgerFileList.getLastMappedFile(dividedCommitlogOffset); log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset); } DLedgerCommitLog(final DefaultMessageStore defaultMessageStore); @Override boolean load(); @Override void start(); @Override void shutdown(); @Override long flush(); @Override long getMaxOffset(); @Override long getMinOffset(); @Override long getConfirmOffset(); @Override void setConfirmOffset(long phyOffset); @Override long remainHowManyDataToCommit(); @Override long remainHowManyDataToFlush(); @Override int deleteExpiredFile(
        final long expiredTime,
        final int deleteFilesInterval,
        final long intervalForcibly,
        final boolean cleanImmediately
    ); SelectMappedBufferResult convertSbr(SelectMmapBufferResult sbr); SelectMmapBufferResult truncate(SelectMmapBufferResult sbr); @Override SelectMappedBufferResult getData(final long offset); @Override SelectMappedBufferResult getData(final long offset, final boolean returnFirstOnNotFound); @Override void recoverNormally(long maxPhyOffsetOfConsumeQueue); @Override void recoverAbnormally(long maxPhyOffsetOfConsumeQueue); @Override DispatchRequest checkMessageAndReturnSize(ByteBuffer byteBuffer, final boolean checkCRC); @Override DispatchRequest checkMessageAndReturnSize(ByteBuffer byteBuffer, final boolean checkCRC,
        final boolean readBody); @Override boolean resetOffset(long offset); @Override long getBeginTimeInLock(); @Override PutMessageResult putMessage(final MessageExtBrokerInner msg); @Override PutMessageResult putMessages(final MessageExtBatch messageExtBatch); @Override CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg); @Override CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch); @Override SelectMappedBufferResult getMessage(final long offset, final int size); @Override long rollNextFile(final long offset); @Override HashMap<String, Long> getTopicQueueTable(); @Override void setTopicQueueTable(HashMap<String, Long> topicQueueTable); @Override void destroy(); @Override boolean appendData(long startOffset, byte[] data); @Override void checkSelf(); @Override long lockTimeMills(); DLedgerServer getdLedgerServer(); int getId(); long getDividedCommitlogOffset();  }