public Map<String, Object> replay() throws IOException {
        replayDone = false;
        metaCache.clear();
        boolean summaryEnabled = LOGGER.isDebugEnabled();
        StringBuilder sb = new StringBuilder();
        if (summaryEnabled) {
            sb.append(logIdent)
                    .append("metaStream replay summary:")
                    .append(" id: ")
                    .append(streamId())
                    .append(", ");
        }
        long totalValueSize = 0L;

        long startOffset = startOffset();
        long endOffset = nextOffset();
        long pos = startOffset;
        FetchContext fetchContext = new FetchContext();
        ReadOptions readOptions = ReadOptions.builder().prioritizedRead(true).build();
        fetchContext.setReadOptions(readOptions);

        try {
            while (pos < endOffset) {
                FetchResult fetchRst = fetch(fetchContext, pos, endOffset, 64 * 1024).get();
                for (RecordBatchWithContext context : fetchRst.recordBatchList()) {
                    try {
                        MetaKeyValue kv = MetaKeyValue.decode(Unpooled.copiedBuffer(context.rawPayload()).nioBuffer());
                        metaCache.put(kv.getKey(), new MetadataValue(context.baseOffset(), kv.getValue()));
                        totalValueSize += kv.getValue().remaining();
                        if (summaryEnabled) {
                            sb.append("(key: ").append(kv.getKey()).append(", offset: ").append(context.baseOffset()).append(", value size: ").append(kv.getValue().remaining()).append("); ");
                        }
                    } catch (Exception e) {
                        LOGGER.error("{} streamId {}: decode meta failed, offset: {}, error: {}", logIdent, streamId(), context.baseOffset(), e.getMessage());
                    }
                    pos = context.lastOffset();
                }
                fetchRst.free();
            }
            replayDone = true;
        } catch (ExecutionException e) {
            if (e.getCause() instanceof IOException) {
                fenced = true;
                throw (IOException) (e.getCause());
            } else {
                throw new RuntimeException(e.getCause());
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        if (totalValueSize > 0 && summaryEnabled) {
            LOGGER.debug(sb.append("total value size: ").append(totalValueSize).toString());
        }
        return getValidMetaMap();
    }
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