public CompletableFuture<ElasticLogMeta> asyncPersistLogMeta() {
        ElasticLogMeta meta;
        Map<String, Long> trimOffsets;

        segmentLock.lock();
        try {
            Map<String, Stream> streams = streamManager.streams();
            List<ElasticStreamSegmentMeta> segmentList = segments.values().stream()
                .sorted()
                .map(ElasticLogSegment::meta)
                .collect(Collectors.toList());

            meta = logMeta(streams, segmentList);
            this.logMeta = meta;
            // We calculate trimOffsets in the lock to ensure that no more new stream with data is created during the calculation.
            trimOffsets = calTrimOffset(
                streams,
                segmentList.iterator(),
                inflightCleanedSegments.values().stream().map(ElasticLogSegment::meta).iterator()
            );
        } finally {
            segmentLock.unlock();
        }

        MetaKeyValue kv = MetaKeyValue.of(MetaStream.LOG_META_KEY, ElasticLogMeta.encode(meta));
        return metaStream.append(kv).thenApply(nil -> {
            LOGGER.info("{} save log meta {}", logIdent, meta);
            trimStream(trimOffsets);
            return meta;
        }).whenComplete((nil, ex) -> {
            if (ex != null) {
                LOGGER.error("{} persist log meta {} fail", logIdent, meta, ex);
            }
        });
    }
The lock of {@link #segments} and {@link #inflightCleanedSegments}
core/src/main/scala/kafka/log/streamaspect/ElasticLogSegmentManager.java