diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java index b80ebdbed622e..eab2054939048 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java @@ -23,6 +23,7 @@ import java.net.URISyntaxException; import java.util.ArrayList; import java.util.EnumSet; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.NavigableMap; @@ -64,6 +65,9 @@ private static class Value { private static final Map> STATIC_MAPS = new MapMaker() .weakValues().makeMap(); + // Manage all instances to facilitate registration to the same listener + private static final Map> STATIC_INSTANCE = new MapMaker() + .weakValues().makeMap(); private static final Map STATIC_ID_GEN_MAP = new MapMaker() .weakValues().makeMap(); @@ -84,6 +88,17 @@ public LocalMemoryMetadataStore(String metadataURL, MetadataStoreConfig metadata // Use a reference from a shared data set String name = uri.getHost(); map = STATIC_MAPS.computeIfAbsent(name, __ -> new TreeMap<>()); + STATIC_INSTANCE.compute(name, (key, value) -> { + if (value == null) { + value = new HashSet<>(); + } + value.forEach(v -> { + registerListener(v); + v.registerListener(this); + }); + value.add(this); + return value; + }); sequentialIdGenerator = STATIC_ID_GEN_MAP.computeIfAbsent(name, __ -> new AtomicLong()); log.info("Created LocalMemoryDataStore for '{}'", name); }