protected void replaceFiles(List<MappedFile> mappedFileList, TopicPartitionLog current,
        TopicPartitionLog newLog) {

        MappedFileQueue dest = current.getLog();
        MappedFileQueue src = newLog.getLog();

        long beginTime = System.nanoTime();
//        List<String> fileNameToReplace = mappedFileList.stream()
//            .map(m -> m.getFile().getName())
//            .collect(Collectors.toList());

        List<String> fileNameToReplace = dest.getMappedFiles().stream()
            .filter(mappedFileList::contains)
            .map(mf -> mf.getFile().getName())
            .collect(Collectors.toList());

        mappedFileList.forEach(MappedFile::renameToDelete);

        src.getMappedFiles().forEach(mappedFile -> {
            try {
                mappedFile.flush(0);
                mappedFile.moveToParent();
            } catch (IOException e) {
                log.error("move file {} to parent directory exception: ", mappedFile.getFileName());
            }
        });

        dest.getMappedFiles().stream()
            .filter(m -> !mappedFileList.contains(m))
            .forEach(m -> src.getMappedFiles().add(m));

        readMessageLock.lock();
        try {
            mappedFileList.forEach(mappedFile -> mappedFile.destroy(1000));

            dest.getMappedFiles().clear();
            dest.getMappedFiles().addAll(src.getMappedFiles());
            src.getMappedFiles().clear();

            replaceCqFiles(getCQ(), newLog.getCQ(), fileNameToReplace);

            log.info("replace file elapsed {} milliseconds",
                TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime));
        } finally {
            readMessageLock.unlock();
        }
    }
@Test
    public void testReplaceFiles() throws IOException, IllegalAccessException {
        Assume.assumeFalse(MixAll.isWindows());
        CompactionLog clog = mock(CompactionLog.class);
        doCallRealMethod().when(clog).replaceFiles(anyList(), any(CompactionLog.TopicPartitionLog.class),
            any(CompactionLog.TopicPartitionLog.class));
        doCallRealMethod().when(clog).replaceCqFiles(any(SparseConsumeQueue.class),
            any(SparseConsumeQueue.class), anyList());

        CompactionLog.TopicPartitionLog dest = mock(CompactionLog.TopicPartitionLog.class);
        MappedFileQueue destMFQ = mock(MappedFileQueue.class);
        when(dest.getLog()).thenReturn(destMFQ);
        List<MappedFile> destFiles = Lists.newArrayList();
        when(destMFQ.getMappedFiles()).thenReturn(destFiles);

        List<MappedFile> srcFiles = Lists.newArrayList();
        String fileName = logPath + File.separator + COMPACTING_SUB_FOLDER + File.separator + String.format("%010d", 0);
        MappedFile mf = new DefaultMappedFile(fileName, 1024);
        srcFiles.add(mf);
        MappedFileQueue srcMFQ = mock(MappedFileQueue.class);
        when(srcMFQ.getMappedFiles()).thenReturn(srcFiles);
        CompactionLog.TopicPartitionLog src = mock(CompactionLog.TopicPartitionLog.class);
        when(src.getLog()).thenReturn(srcMFQ);

        FieldUtils.writeField(clog, "readMessageLock", new PutMessageSpinLock(), true);

        clog.replaceFiles(Lists.newArrayList(), dest, src);
        assertEquals(destFiles.size(), 1);
        destFiles.forEach(f -> {
            assertFalse(f.getFileName().contains(COMPACTING_SUB_FOLDER));
        });
    }