fix(metadataimage): fix getObjects not complete cf when throw exception (AutoMQ#1764)

* fix(controller): npe when cleanup scale in node

* fix(metadataimage): fix getObjects  not complete cf when throw exception

* fix(metadataimage): fix getObjects not complete cf when throw exception

* fix(metadataimage): fix getObjects not complete cf when throw exception

* fix(metadataimage): fix getObjects not complete cf when throw exception

fix comment

Author: lifepuzzlefun

metadata/src/main/java/org/apache/kafka/image/S3StreamsMetadataImage.java

@@ -37,6 +37,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.metadata.AssignedStreamIdRecord;
@@ -258,6 +259,7 @@ void fillObjects0(
                 } else {
                     streamSetObjects = Collections.emptyList();
                 }
+
                 CompletableFuture<Integer> startSearchIndexCf = getStartSearchIndex(node, nextStartOffset, ctx);
                 final int finalLastRangeIndex = lastRangeIndex;
                 final long finalNextStartOffset = nextStartOffset;
@@ -643,7 +645,7 @@ static class GetObjectsContext {
         LocalStreamRangeIndexCache indexCache;
 
         CompletableFuture<InRangeObjects> cf = new CompletableFuture<>();
-        Map<Long, Optional<StreamOffsetRange>> object2range = new HashMap<>();
+        Map<Long, Optional<StreamOffsetRange>> object2range = new ConcurrentHashMap<>();
         List<String> debugContext = new ArrayList<>();
 
         GetObjectsContext(long streamId, long startOffset, long endOffset, int limit,