crossoverJie commented on code in PR #22735:
URL: https://github.com/apache/pulsar/pull/22735#discussion_r1687642455
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -133,10 +135,17 @@ public
BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispat
new MutableBucket(dispatcher.getName(),
dispatcher.getCursor(), FutureUtil.Sequencer.create(),
bucketSnapshotStorage);
this.stats = new BucketDelayedMessageIndexStats();
- this.numberDelayedMessages = recoverBucketSnapshot();
+
+ // Close the tracker if failed to recover.
+ try {
+ this.numberDelayedMessages = recoverBucketSnapshot();
+ } catch (RecoverDelayedDeliveryTrackerException e) {
+ close();
Review Comment:
It's better to move the `close()` function to the `finally` block.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -284,10 +285,11 @@ public class BrokerService implements Closeable {
private final AtomicBoolean blockedDispatcherOnHighUnackedMsgs = new
AtomicBoolean(false);
private final ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers>
blockedDispatchers;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
-
- @Getter
@VisibleForTesting
private final DelayedDeliveryTrackerFactory delayedDeliveryTrackerFactory;
+ // InMemoryDelayedDeliveryTrackerFactory is for the purpose of
+ // fallback if recover BucketDelayedDeliveryTracker failed.
+ private volatile DelayedDeliveryTrackerFactory
fallbackDelayedDeliveryTrackerFactory;
Review Comment:
Why not reuse the original `delayedDeliveryTrackerFactory`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
