@Override
        public void onEvent(long segmentBaseOffset, ElasticLogSegmentEvent event) {
            switch (event) {
                case SEGMENT_DELETE: {
                    boolean deleted = remove(segmentBaseOffset) != null;
                    if (deleted) {
                        // This may happen since kafka.log.LocalLog.deleteSegmentFiles schedules the delayed deletion task.
                        if (metaStream.isFenced()) {
                            if (LOGGER.isDebugEnabled()) {
                                LOGGER.debug("{} meta stream is closed, skip persisting log meta", logIdent);
                            }
                        } else {
                            submitOrDrainPendingPersistentMetaQueue(segmentBaseOffset);
                        }
                    }
                    break;
                }
                case SEGMENT_UPDATE: {
                    persistLogMeta();
                    break;
                }
                default: {
                    throw new IllegalStateException("Unsupported event " + event);
                }
            }
        }
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