protected void compaction(List<MappedFile> mappedFileList, OffsetMap offsetMap) throws DigestException {
        compacting = new TopicPartitionLog(this, COMPACTING_SUB_FOLDER);

        for (MappedFile mappedFile : mappedFileList) {
            Iterator<SelectMappedBufferResult> iterator = mappedFile.iterator(0);
            while (iterator.hasNext()) {
                SelectMappedBufferResult smb = null;
                try {
                    smb = iterator.next();
                    MessageExt msgExt = MessageDecoder.decode(smb.getByteBuffer(), true, true);
                    if (msgExt == null) {
                        // file end
                        break;
                    } else {
                        checkAndPutMessage(smb, msgExt, offsetMap, compacting);
                    }
                } finally {
                    if (smb != null) {
                        smb.release();
                    }
                }
            }
        }
        putEndMessage(compacting.getLog());
    }
@Test
    public void testCompaction() throws DigestException, NoSuchAlgorithmException, IllegalAccessException {
        Iterator<SelectMappedBufferResult> iterator = mock(Iterator.class);
        SelectMappedBufferResult smb = mock(SelectMappedBufferResult.class);
        when(iterator.hasNext()).thenAnswer((Answer<Boolean>)invocationOnMock -> queueOffset < 1024);
        when(iterator.next()).thenAnswer((Answer<SelectMappedBufferResult>)invocation ->
            new SelectMappedBufferResult(0, buildMessage(), 0, null));

        MappedFile mf = mock(MappedFile.class);
        List<MappedFile> mappedFileList = Lists.newArrayList(mf);
        doReturn(iterator).when(mf).iterator(0);

        MessageStore messageStore = mock(DefaultMessageStore.class);
        CommitLog commitLog = mock(CommitLog.class);
        when(messageStore.getCommitLog()).thenReturn(commitLog);
        when(commitLog.getCommitLogSize()).thenReturn(1024 * 1024);
        CompactionLog clog = mock(CompactionLog.class);
        FieldUtils.writeField(clog, "defaultMessageStore", messageStore, true);
        doCallRealMethod().when(clog).getOffsetMap(any());
        FieldUtils.writeField(clog, "positionMgr", positionMgr, true);

        queueOffset = 0;
        CompactionLog.OffsetMap offsetMap = clog.getOffsetMap(mappedFileList);
        assertEquals(1023, offsetMap.getLastOffset());

        doCallRealMethod().when(clog).compaction(any(List.class), any(CompactionLog.OffsetMap.class));
        doNothing().when(clog).putEndMessage(any(MappedFileQueue.class));
        doCallRealMethod().when(clog).checkAndPutMessage(any(SelectMappedBufferResult.class),
            any(MessageExt.class), any(CompactionLog.OffsetMap.class), any(CompactionLog.TopicPartitionLog.class));
        doCallRealMethod().when(clog).shouldRetainMsg(any(MessageExt.class), any(CompactionLog.OffsetMap.class));
        List<MessageExt> compactResult = Lists.newArrayList();
        when(clog.asyncPutMessage(any(ByteBuffer.class), any(MessageExt.class),
            any(CompactionLog.TopicPartitionLog.class)))
            .thenAnswer((Answer<CompletableFuture<PutMessageResult>>)invocation -> {
                compactResult.add(invocation.getArgument(1));
                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.PUT_OK,
                    new AppendMessageResult(AppendMessageStatus.PUT_OK)));
            });
        queueOffset = 0;
        clog.compaction(mappedFileList, offsetMap);
        assertEquals(keyCount, compactResult.size());
        assertEquals(1014, compactResult.stream().mapToLong(MessageExt::getQueueOffset).min().orElse(1024));
        assertEquals(1023, compactResult.stream().mapToLong(MessageExt::getQueueOffset).max().orElse(0));
    }