private void submitOrDrainPendingPersistentMetaQueue(long segmentBaseOffset) {
            if (segmentBaseOffset != NO_OP_OFFSET) {
                pendingDeleteSegmentBaseOffset.add(segmentBaseOffset);
            }

            synchronized (this) {
                if (pendingPersistentMetaCf != null && !pendingPersistentMetaCf.isDone()) {
                    return;
                }

                long maxOffset = NO_OP_OFFSET;

                while (!pendingDeleteSegmentBaseOffset.isEmpty()) {
                    long baseOffset = pendingDeleteSegmentBaseOffset.poll();
                    maxOffset = Math.max(maxOffset, baseOffset);
                }

                if (maxOffset != NO_OP_OFFSET) {
                    if (metaStream.isFenced()) {
                        if (LOGGER.isDebugEnabled()) {
                            LOGGER.debug("{} meta stream is closed, skip persisting log meta", logIdent);
                        }

                        return;
                    }

                    long finalMaxOffset = maxOffset;
                    pendingPersistentMetaCf = asyncPersistLogMeta();
                    pendingPersistentMetaCf.whenCompleteAsync((res, e) -> {
                        if (e != null) {
                            LOGGER.error("error when persisLogMeta maxOffset {}", finalMaxOffset, e);
                        }

                        submitOrDrainPendingPersistentMetaQueue(-1);
                    });
                }
            }
        }
Calculate trim offset of each stream.

@return stream trim offset map, key is stream name, value is trim offset.
core/src/main/scala/kafka/log/streamaspect/ElasticLogSegmentManager.java