fix(s3stream): avoid StreamMetadataManager add callback when retry (#… (#2661)

fix(s3stream): avoid StreamMetadataManager add callback when retry (#2659)

Co-authored-by: lifepuzzlefun <[email protected]>

Author: Gezi-lzq

core/src/main/scala/kafka/log/stream/s3/metadata/StreamMetadataManager.java

@@ -138,12 +138,12 @@ public Set<Long> getStreamSetObjectIds() {
     public CompletableFuture<InRangeObjects> fetch(long streamId, long startOffset, long endOffset, int limit) {
         // TODO: cache the object list for next search
         CompletableFuture<InRangeObjects> cf = new CompletableFuture<>();
-        exec(() -> fetch0(cf, streamId, startOffset, endOffset, limit), cf, LOGGER, "fetchObjects");
+        exec(() -> fetch0(cf, streamId, startOffset, endOffset, limit, false), cf, LOGGER, "fetchObjects");
         return cf;
     }
 
     private void fetch0(CompletableFuture<InRangeObjects> cf, long streamId,
-        long startOffset, long endOffset, int limit) {
+        long startOffset, long endOffset, int limit, boolean retryFetch) {
         Image image = getImage();
         try {
             final S3StreamsMetadataImage streamsImage = image.streamsMetadata();
@@ -179,9 +179,11 @@ private void fetch0(CompletableFuture<InRangeObjects> cf, long streamId,
                     streamId, startOffset, endOffset, limit, rst.objects().size(), rst.endOffset());
 
                 CompletableFuture<Void> pendingCf = pendingFetch();
-                pendingCf.thenAccept(nil -> fetch0(cf, streamId, startOffset, endOffset, limit));
-                cf.whenComplete((r, ex) ->
-                    LOGGER.info("[FetchObjects],[COMPLETE_PENDING],streamId={} startOffset={} endOffset={} limit={}", streamId, startOffset, endOffset, limit));
+                pendingCf.thenAccept(nil -> fetch0(cf, streamId, startOffset, endOffset, limit, true));
+                if (!retryFetch) {
+                    cf.whenComplete((r, ex) ->
+                        LOGGER.info("[FetchObjects],[COMPLETE_PENDING],streamId={} startOffset={} endOffset={} limit={}", streamId, startOffset, endOffset, limit));
+                }
             }).exceptionally(ex -> {
                 cf.completeExceptionally(ex);
                 return null;