Fix code

apache · Mar 26, 2023 · bff1402 · bff1402
1 parent ce358bf
commit bff1402
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 8 deletions.
diff --git a/...r/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/...r/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
@@ -436,12 +436,13 @@ private synchronized CompletableFuture<Void> asyncMergeBucketSnapshot(ImmutableB
                     .thenAccept(combinedDelayedIndexQueue -> {
                         synchronized (BucketDelayedDeliveryTracker.this) {
                             long createStartTime = System.currentTimeMillis();
-                    stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.create);
-                    Pair<ImmutableBucket, DelayedIndex> immutableBucketDelayedIndexPair =
-                            lastMutableBucket.createImmutableBucketAndAsyncPersistent(
-                                    timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment,
-                                    sharedBucketPriorityQueue, combinedDelayedIndexQueue, bucketA.startLedgerId,
-                                    bucketB.endLedgerId);
+                            stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.create);
+                            Pair<ImmutableBucket, DelayedIndex> immutableBucketDelayedIndexPair =
+                                    lastMutableBucket.createImmutableBucketAndAsyncPersistent(
+                                            timeStepPerBucketSnapshotSegmentInMillis,
+                                            maxIndexesPerBucketSnapshotSegment,
+                                            sharedBucketPriorityQueue, combinedDelayedIndexQueue, bucketA.startLedgerId,
+                                            bucketB.endLedgerId);
 
                             // Merge bit map to new bucket
                             Map<Long, RoaringBitmap> delayedIndexBitMapA = bucketA.getDelayedIndexBitMap();

diff --git a/...src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java b/...src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java
@@ -75,7 +75,7 @@ public Map<String, TopicMetricBean> genTopicMetricMap() {
             String[] labels = splitKey(k);
             String[] labelsAndValues = new String[] {"state", labels[0], "type", labels[1]};
             String key = OP_COUNT_NAME + joinKey(labelsAndValues);
-            metrics.put(key, new TopicMetricBean(OP_COUNT_NAME, count.longValue(), labelsAndValues));
+            metrics.put(key, new TopicMetricBean(OP_COUNT_NAME, count.sumThenReset(), labelsAndValues));
         });
 
         delayedMessageIndexBucketOpLatencyMs.forEach((typeName, statsBuckets) -> {

diff --git a/...c/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/...c/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
@@ -158,7 +158,7 @@ public Object[][] provider(Method method) throws Exception {
     }
 
     @Test(dataProvider = "delayedTracker")
-    public void testContainsMessage(DelayedDeliveryTracker tracker) {
+    public void testContainsMessage(BucketDelayedDeliveryTracker tracker) {
         tracker.addMessage(1, 1, 10);
         tracker.addMessage(2, 2, 20);