[BUG] Fetch Stream Out of Bound

[Chillax-0v0]

Version: 1.1.2-rc1

What's wrong:

What happened:

In the logs, we can see that a segment swap occurred before the issue happened. During this swap, the "log" segment was replaced with the "log.cleaned" segment, the latter of which used the problematic Stream-300979.

automq/core/src/main/scala/kafka/log/streamaspect/ElasticLogSegmentManager.java

Lines 100 to 118 in f2066cb

	private void trimStream0(ElasticLogMeta meta) {
	Map<String, Long> streamMinOffsets = new HashMap<>();
	inflightCleanedSegments.forEach((offset, segment) -> {
	ElasticStreamSegmentMeta segMeta = segment.meta();
	calStreamsMinOffset(streamMinOffsets, segMeta);
	});
	for (ElasticStreamSegmentMeta segMeta : meta.getSegmentMetas()) {
	calStreamsMinOffset(streamMinOffsets, segMeta);
	}
	streamManager.streams().forEach((streamName, stream) -> {
	var minOffset = streamMinOffsets.get(streamName);
	// if minOffset == null, then stream is not used by any segment, should trim it to end.
	minOffset = Optional.ofNullable(minOffset).orElse(stream.nextOffset());
	if (minOffset > stream.startOffset()) {
	stream.trim(minOffset);
	}
	});
	}

In the code, we can see that before trimming a stream, the minimum offset for each stream is calculated. Specifically, if a stream is not being used by any segment, it will be trimmed to the end.
However, when calculating the minimum offset, the inflightCleanedSegments (a member variable) and meta.getSegmentMetas() (a method parameter) do not represent the same view at the same point in time. This discrepancy can lead to the incorrect assumption that a stream is not being used by any segment.

[superhx]