@Test public void testDeleteExpiredFileByOffset() { MappedFileQueue mappedFileQueue = new MappedFileQueue("target/unit_test_store/e", 5120, null); for (int i = 0; i < 2048; i++) { MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); assertThat(mappedFile).isNotNull(); ByteBuffer byteBuffer = ByteBuffer.allocate(ConsumeQueue.CQ_STORE_UNIT_SIZE); byteBuffer.putLong(i); byte[] padding = new byte[12]; Arrays.fill(padding, (byte) '0'); byteBuffer.put(padding); byteBuffer.flip(); assertThat(mappedFile.appendMessage(byteBuffer.array())).isTrue(); } MappedFile first = mappedFileQueue.getFirstMappedFile(); first.hold(); assertThat(mappedFileQueue.deleteExpiredFileByOffset(20480, ConsumeQueue.CQ_STORE_UNIT_SIZE)).isEqualTo(0); first.release(); assertThat(mappedFileQueue.deleteExpiredFileByOffset(20480, ConsumeQueue.CQ_STORE_UNIT_SIZE)).isGreaterThan(0); first = mappedFileQueue.getFirstMappedFile(); assertThat(first.getFileFromOffset()).isGreaterThan(0); mappedFileQueue.shutdown(1000); mappedFileQueue.destroy(); }
public int deleteExpiredFileByOffset(long offset, int unitSize) { Object[] mfs = this.copyMappedFiles(0); List<MappedFile> files = new ArrayList<MappedFile>(); int deleteCount = 0; if (null != mfs) { int mfsLength = mfs.length - 1; for (int i = 0; i < mfsLength; i++) { boolean destroy; MappedFile mappedFile = (MappedFile) mfs[i]; SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize); if (result != null) { long maxOffsetInLogicQueue = result.getByteBuffer().getLong(); result.release(); destroy = maxOffsetInLogicQueue < offset; if (destroy) { log.info("physic min offset " + offset + ", logics in current mappedFile max offset " + maxOffsetInLogicQueue + ", delete it"); } } else if (!mappedFile.isAvailable()) { log.warn("Found a hanged consume queue file, attempting to delete it."); destroy = true; } else { log.warn("this being not executed forever."); break; } if (destroy && mappedFile.destroy(1000 * 60)) { files.add(mappedFile); deleteCount++; } else { break; } } } deleteExpiredFile(files); return deleteCount; }
MappedFileQueue { public int deleteExpiredFileByOffset(long offset, int unitSize) { Object[] mfs = this.copyMappedFiles(0); List<MappedFile> files = new ArrayList<MappedFile>(); int deleteCount = 0; if (null != mfs) { int mfsLength = mfs.length - 1; for (int i = 0; i < mfsLength; i++) { boolean destroy; MappedFile mappedFile = (MappedFile) mfs[i]; SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize); if (result != null) { long maxOffsetInLogicQueue = result.getByteBuffer().getLong(); result.release(); destroy = maxOffsetInLogicQueue < offset; if (destroy) { log.info("physic min offset " + offset + ", logics in current mappedFile max offset " + maxOffsetInLogicQueue + ", delete it"); } } else if (!mappedFile.isAvailable()) { log.warn("Found a hanged consume queue file, attempting to delete it."); destroy = true; } else { log.warn("this being not executed forever."); break; } if (destroy && mappedFile.destroy(1000 * 60)) { files.add(mappedFile); deleteCount++; } else { break; } } } deleteExpiredFile(files); return deleteCount; } }
MappedFileQueue { public int deleteExpiredFileByOffset(long offset, int unitSize) { Object[] mfs = this.copyMappedFiles(0); List<MappedFile> files = new ArrayList<MappedFile>(); int deleteCount = 0; if (null != mfs) { int mfsLength = mfs.length - 1; for (int i = 0; i < mfsLength; i++) { boolean destroy; MappedFile mappedFile = (MappedFile) mfs[i]; SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize); if (result != null) { long maxOffsetInLogicQueue = result.getByteBuffer().getLong(); result.release(); destroy = maxOffsetInLogicQueue < offset; if (destroy) { log.info("physic min offset " + offset + ", logics in current mappedFile max offset " + maxOffsetInLogicQueue + ", delete it"); } } else if (!mappedFile.isAvailable()) { log.warn("Found a hanged consume queue file, attempting to delete it."); destroy = true; } else { log.warn("this being not executed forever."); break; } if (destroy && mappedFile.destroy(1000 * 60)) { files.add(mappedFile); deleteCount++; } else { break; } } } deleteExpiredFile(files); return deleteCount; } MappedFileQueue(final String storePath, int mappedFileSize,
        AllocateMappedFileService allocateMappedFileService); }
MappedFileQueue { public int deleteExpiredFileByOffset(long offset, int unitSize) { Object[] mfs = this.copyMappedFiles(0); List<MappedFile> files = new ArrayList<MappedFile>(); int deleteCount = 0; if (null != mfs) { int mfsLength = mfs.length - 1; for (int i = 0; i < mfsLength; i++) { boolean destroy; MappedFile mappedFile = (MappedFile) mfs[i]; SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize); if (result != null) { long maxOffsetInLogicQueue = result.getByteBuffer().getLong(); result.release(); destroy = maxOffsetInLogicQueue < offset; if (destroy) { log.info("physic min offset " + offset + ", logics in current mappedFile max offset " + maxOffsetInLogicQueue + ", delete it"); } } else if (!mappedFile.isAvailable()) { log.warn("Found a hanged consume queue file, attempting to delete it."); destroy = true; } else { log.warn("this being not executed forever."); break; } if (destroy && mappedFile.destroy(1000 * 60)) { files.add(mappedFile); deleteCount++; } else { break; } } } deleteExpiredFile(files); return deleteCount; } MappedFileQueue(final String storePath, int mappedFileSize,
        AllocateMappedFileService allocateMappedFileService); void checkSelf(); MappedFile getMappedFileByTime(final long timestamp); void truncateDirtyFiles(long offset); boolean load(); long howMuchFallBehind(); MappedFile getLastMappedFile(final long startOffset, boolean needCreate); MappedFile getLastMappedFile(final long startOffset); MappedFile getLastMappedFile(); boolean resetOffset(long offset); long getMinOffset(); long getMaxOffset(); long getMaxWrotePosition(); long remainHowManyDataToCommit(); long remainHowManyDataToFlush(); void deleteLastMappedFile(); int deleteExpiredFileByTime(final long expiredTime,
        final int deleteFilesInterval,
        final long intervalForcibly,
        final boolean cleanImmediately); int deleteExpiredFileByOffset(long offset, int unitSize); boolean flush(final int flushLeastPages); boolean commit(final int commitLeastPages); MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound); MappedFile getFirstMappedFile(); MappedFile findMappedFileByOffset(final long offset); long getMappedMemorySize(); boolean retryDeleteFirstFile(final long intervalForcibly); void shutdown(final long intervalForcibly); void destroy(); long getFlushedWhere(); void setFlushedWhere(long flushedWhere); long getStoreTimestamp(); List<MappedFile> getMappedFiles(); int getMappedFileSize(); long getCommittedWhere(); void setCommittedWhere(final long committedWhere); }
MappedFileQueue { public int deleteExpiredFileByOffset(long offset, int unitSize) { Object[] mfs = this.copyMappedFiles(0); List<MappedFile> files = new ArrayList<MappedFile>(); int deleteCount = 0; if (null != mfs) { int mfsLength = mfs.length - 1; for (int i = 0; i < mfsLength; i++) { boolean destroy; MappedFile mappedFile = (MappedFile) mfs[i]; SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize); if (result != null) { long maxOffsetInLogicQueue = result.getByteBuffer().getLong(); result.release(); destroy = maxOffsetInLogicQueue < offset; if (destroy) { log.info("physic min offset " + offset + ", logics in current mappedFile max offset " + maxOffsetInLogicQueue + ", delete it"); } } else if (!mappedFile.isAvailable()) { log.warn("Found a hanged consume queue file, attempting to delete it."); destroy = true; } else { log.warn("this being not executed forever."); break; } if (destroy && mappedFile.destroy(1000 * 60)) { files.add(mappedFile); deleteCount++; } else { break; } } } deleteExpiredFile(files); return deleteCount; } MappedFileQueue(final String storePath, int mappedFileSize,
        AllocateMappedFileService allocateMappedFileService); void checkSelf(); MappedFile getMappedFileByTime(final long timestamp); void truncateDirtyFiles(long offset); boolean load(); long howMuchFallBehind(); MappedFile getLastMappedFile(final long startOffset, boolean needCreate); MappedFile getLastMappedFile(final long startOffset); MappedFile getLastMappedFile(); boolean resetOffset(long offset); long getMinOffset(); long getMaxOffset(); long getMaxWrotePosition(); long remainHowManyDataToCommit(); long remainHowManyDataToFlush(); void deleteLastMappedFile(); int deleteExpiredFileByTime(final long expiredTime,
        final int deleteFilesInterval,
        final long intervalForcibly,
        final boolean cleanImmediately); int deleteExpiredFileByOffset(long offset, int unitSize); boolean flush(final int flushLeastPages); boolean commit(final int commitLeastPages); MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound); MappedFile getFirstMappedFile(); MappedFile findMappedFileByOffset(final long offset); long getMappedMemorySize(); boolean retryDeleteFirstFile(final long intervalForcibly); void shutdown(final long intervalForcibly); void destroy(); long getFlushedWhere(); void setFlushedWhere(long flushedWhere); long getStoreTimestamp(); List<MappedFile> getMappedFiles(); int getMappedFileSize(); long getCommittedWhere(); void setCommittedWhere(final long committedWhere);  }