private synchronized CompletableFuture<Void> doCompaction(boolean force) {
        if (!replayDone) {
            return CompletableFuture.completedFuture(null);
        }
        long startOffset = startOffset();
        long endOffset = nextOffset();
        int size = (int) (endOffset - startOffset);
        if (size == 0) {
            return CompletableFuture.completedFuture(null);
        }
        double hollowRate = 1 - (double) metaCache.size() / size;
        if (!force && hollowRate < COMPACTION_HOLLOW_RATE) {
            return CompletableFuture.completedFuture(null);
        }
        MetadataValue last = null;
        for (MetadataValue value : metaCache.values()) {
            if (last == null || value.offset > last.offset) {
                last = value;
            }
        }
        List<MetaKeyValue> overwrite = new LinkedList<>();
        for (Map.Entry<String, MetadataValue> entry : metaCache.entrySet()) {
            String key = entry.getKey();
            MetadataValue value = entry.getValue();
            if (value == last || (!force && last.timestamp - value.timestamp < COMPACTION_THRESHOLD_MS)) {
                continue;
            }
            overwrite.add(MetaKeyValue.of(key, value.value()));
        }
        if (overwrite.isEmpty()) {
            return CompletableFuture.completedFuture(null);
        }
        CompletableFuture<Void> overwriteCf = CompletableFuture.allOf(overwrite.stream().map(this::append).toArray(CompletableFuture[]::new));
        return overwriteCf.thenAccept(nil -> {
            OptionalLong minOffset = metaCache.values().stream().mapToLong(v -> v.offset).min();
            minOffset.ifPresent(this::trim);
        });
    }
Replay meta stream and return a map of meta keyValues. KeyValues will be cached in metaCache.

@return meta keyValues map
core/src/main/scala/kafka/log/streamaspect/MetaStream.java