Fix totalSnapshotLength

Author: nodece

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

@@ -342,7 +342,8 @@ private void afterCreateImmutableBucket(Pair<ImmutableBucket, DelayedIndex> immu
                 CompletableFuture<Long> future = createFuture.handle((bucketId, ex) -> {
                     if (ex == null) {
                         immutableBucket.setSnapshotSegments(null);
-                        immutableBucket.asyncUpdateSnapshotLength();
+                        immutableBucket.asyncUpdateSnapshotLength()
+                                .thenRun(() -> immutableBuckets.recomputeCounters());
                         log.info()
                                 .attr("bucketKey", immutableBucket.bucketKey())
                                 .log("Create bucket snapshot finish, bucketKey");

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java

@@ -130,19 +130,20 @@ private CompletableFuture<List<DelayedIndex>> asyncLoadNextBucketSnapshotEntry(b
                                     .log("Failed to get bucket snapshot segment");
                         }
                     }), BucketSnapshotPersistenceException.class, MaxRetryTimes)
-                    .thenApply(bucketSnapshotSegments -> {
+                    .thenCompose(bucketSnapshotSegments -> {
                         if (CollectionUtils.isEmpty(bucketSnapshotSegments)) {
-                            return Collections.emptyList();
+                            return CompletableFuture.completedFuture(Collections.emptyList());
                         }
 
                         SnapshotSegment snapshotSegment =
                                 bucketSnapshotSegments.get(0);
                         List<DelayedIndex> indexList = snapshotSegment.getIndexesList();
                         this.setCurrentSegmentEntryId(nextSegmentEntryId);
                         if (isRecover) {
-                            this.asyncUpdateSnapshotLength();
+                            return this.asyncUpdateSnapshotLength()
+                                    .thenApply(__ -> indexList);
                         }
-                        return indexList;
+                        return CompletableFuture.completedFuture(indexList);
                     });
         });
     }