You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2024/09/21 07:02:30 UTC
(pulsar) branch master updated: [improve][broker] Replace ConcurrentOpenHashMap with ConcurrentHashMap in BrokerService (#23320)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new f5c1ad24d77 [improve][broker] Replace ConcurrentOpenHashMap with ConcurrentHashMap in BrokerService (#23320)
f5c1ad24d77 is described below
commit f5c1ad24d7753eb203da3c7f3e4912b005b76044
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sat Sep 21 15:02:24 2024 +0800
[improve][broker] Replace ConcurrentOpenHashMap with ConcurrentHashMap in BrokerService (#23320)
---
.../broker/admin/v2/NonPersistentTopics.java | 10 +-
.../pulsar/broker/namespace/NamespaceService.java | 6 +-
.../org/apache/pulsar/broker/rest/TopicsBase.java | 12 +-
.../pulsar/broker/service/BrokerService.java | 151 ++++-----------------
.../apache/pulsar/broker/service/PulsarStats.java | 5 +-
.../stats/prometheus/NamespaceStatsAggregator.java | 2 +-
.../stats/prometheus/TransactionAggregator.java | 2 +-
.../apache/pulsar/broker/admin/AdminApi2Test.java | 5 +-
.../delayed/DelayedDeliveryTrackerFactoryTest.java | 8 +-
.../broker/namespace/NamespaceServiceTest.java | 6 +-
.../broker/service/AbstractReplicatorTest.java | 4 +-
.../BrokerServiceAutoTopicCreationTest.java | 12 +-
.../pulsar/broker/service/BrokerServiceTest.java | 2 +-
.../broker/service/OneWayReplicatorTest.java | 8 +-
.../broker/service/ReplicatorAdminTlsTest.java | 4 +-
.../ReplicatorAdminTlsWithKeyStoreTest.java | 4 +-
.../broker/service/ReplicatorGlobalNSTest.java | 7 +-
.../pulsar/broker/service/ReplicatorTest.java | 6 +-
.../service/persistent/TopicDuplicationTest.java | 9 +-
.../pulsar/broker/stats/PrometheusMetricsTest.java | 74 +++++-----
.../prometheus/NamespaceStatsAggregatorTest.java | 15 +-
.../TopicTransactionBufferRecoverTest.java | 19 +--
.../broker/transaction/TransactionProduceTest.java | 2 +-
.../pulsar/broker/transaction/TransactionTest.java | 4 +-
.../buffer/TransactionLowWaterMarkTest.java | 10 +-
.../pendingack/PendingAckInMemoryDeleteTest.java | 19 +--
.../apache/pulsar/client/api/TopicReaderTest.java | 2 +-
.../client/impl/BrokerClientIntegrationTest.java | 5 +-
.../client/impl/TransactionEndToEndTest.java | 15 +-
29 files changed, 130 insertions(+), 298 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
index 9f58aa4ca9d..edf4303e1ad 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
@@ -62,7 +62,6 @@ import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.policies.data.stats.NonPersistentPartitionedTopicStatsImpl;
import org.apache.pulsar.common.policies.data.stats.NonPersistentTopicStatsImpl;
import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -478,18 +477,17 @@ public class NonPersistentTopics extends PersistentTopics {
} else {
validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange, true, true)
.thenAccept(nsBundle -> {
- ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>> bundleTopics =
- pulsar().getBrokerService()
- .getMultiLayerTopicsMap().get(namespaceName.toString());
+ final var bundleTopics = pulsar().getBrokerService().getMultiLayerTopicsMap()
+ .get(namespaceName.toString());
if (bundleTopics == null || bundleTopics.isEmpty()) {
asyncResponse.resume(Collections.emptyList());
return;
}
final List<String> topicList = new ArrayList<>();
String bundleKey = namespaceName.toString() + "/" + nsBundle.getBundleRange();
- ConcurrentOpenHashMap<String, Topic> topicMap = bundleTopics.get(bundleKey);
+ final var topicMap = bundleTopics.get(bundleKey);
if (topicMap != null) {
- topicList.addAll(topicMap.keys().stream()
+ topicList.addAll(topicMap.keySet().stream()
.filter(name -> !TopicName.get(name).isPersistent())
.collect(Collectors.toList()));
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 92188f5e6ee..0b1661fb954 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1618,10 +1618,10 @@ public class NamespaceService implements AutoCloseable {
// Non-persistent topics don't have managed ledgers. So we have to retrieve them from local
// cache.
List<String> topics = new ArrayList<>();
- synchronized (pulsar.getBrokerService().getMultiLayerTopicMap()) {
- if (pulsar.getBrokerService().getMultiLayerTopicMap()
+ synchronized (pulsar.getBrokerService().getMultiLayerTopicsMap()) {
+ if (pulsar.getBrokerService().getMultiLayerTopicsMap()
.containsKey(namespaceName.toString())) {
- pulsar.getBrokerService().getMultiLayerTopicMap().get(namespaceName.toString())
+ pulsar.getBrokerService().getMultiLayerTopicsMap().get(namespaceName.toString())
.forEach((__, bundle) -> bundle.forEach((topicName, topic) -> {
if (topic instanceof NonPersistentTopic
&& ((NonPersistentTopic) topic).isActive()) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
index 8f55df1107d..bf6e7350186 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
@@ -38,7 +38,9 @@ import java.util.Base64;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import javax.ws.rs.container.AsyncResponse;
@@ -95,7 +97,6 @@ import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.ObjectMapperFactory;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
import org.apache.pulsar.websocket.data.ProducerAck;
import org.apache.pulsar.websocket.data.ProducerAcks;
import org.apache.pulsar.websocket.data.ProducerMessage;
@@ -122,8 +123,9 @@ public class TopicsBase extends PersistentTopicsBase {
.thenAccept(schemaMeta -> {
// Both schema version and schema data are necessary.
if (schemaMeta.getLeft() != null && schemaMeta.getRight() != null) {
- internalPublishMessages(topicName, request, pulsar().getBrokerService()
- .getOwningTopics().get(topic).values(), asyncResponse,
+ final var partitionIndexes = pulsar().getBrokerService().getOwningTopics()
+ .getOrDefault(topic, Set.of()).stream().toList();
+ internalPublishMessages(topicName, request, partitionIndexes, asyncResponse,
AutoConsumeSchema.getSchema(schemaMeta.getLeft().toSchemaInfo()),
schemaMeta.getRight());
} else {
@@ -446,7 +448,7 @@ public class TopicsBase extends PersistentTopicsBase {
}
pulsar().getBrokerService().getOwningTopics().computeIfAbsent(partitionedTopicName
.getPartitionedTopicName(),
- (key) -> ConcurrentOpenHashSet.<Integer>newBuilder().build())
+ __ -> ConcurrentHashMap.newKeySet())
.add(partitionedTopicName.getPartitionIndex());
completeLookup(Pair.of(Collections.emptyList(), false), redirectAddresses, future);
} else {
@@ -517,7 +519,7 @@ public class TopicsBase extends PersistentTopicsBase {
// Only need to add to first partition the broker owns since the schema id in schema registry are
// same for all partitions which is the partitionedTopicName
List<Integer> partitions = pulsar().getBrokerService().getOwningTopics()
- .get(topicName.getPartitionedTopicName()).values();
+ .get(topicName.getPartitionedTopicName()).stream().toList();
CompletableFuture<SchemaVersion> result = new CompletableFuture<>();
for (int index = 0; index < partitions.size(); index++) {
CompletableFuture<SchemaVersion> future = new CompletableFuture<>();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index c7a210bc543..6b0be07c8f7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -63,6 +63,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.RejectedExecutionException;
@@ -177,8 +178,6 @@ import org.apache.pulsar.common.stats.Metrics;
import org.apache.pulsar.common.util.FieldParser;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.GracefulExecutorServicesShutdown;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
import org.apache.pulsar.common.util.netty.ChannelFutures;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
import org.apache.pulsar.common.util.netty.NettyFutureUtil;
@@ -217,27 +216,26 @@ public class BrokerService implements Closeable {
private final PulsarService pulsar;
private final ManagedLedgerFactory managedLedgerFactory;
- private final ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics;
+ private final Map<String, CompletableFuture<Optional<Topic>>> topics = new ConcurrentHashMap<>();
- private final ConcurrentOpenHashMap<String, PulsarClient> replicationClients;
- private final ConcurrentOpenHashMap<String, PulsarAdmin> clusterAdmins;
+ private final Map<String, PulsarClient> replicationClients = new ConcurrentHashMap<>();
+ private final Map<String, PulsarAdmin> clusterAdmins = new ConcurrentHashMap<>();
// Multi-layer topics map:
// Namespace --> Bundle --> topicName --> topic
- private final ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>>>
- multiLayerTopicsMap;
+ private final Map<String, Map<String, Map<String, Topic>>> multiLayerTopicsMap = new ConcurrentHashMap<>();
// Keep track of topics and partitions served by this broker for fast lookup.
@Getter
- private final ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<Integer>> owningTopics;
+ private final Map<String, Set<Integer>> owningTopics = new ConcurrentHashMap<>();
private long numberOfNamespaceBundles = 0;
private final EventLoopGroup acceptorGroup;
private final EventLoopGroup workerGroup;
private final OrderedExecutor topicOrderedExecutor;
// offline topic backlog cache
- private final ConcurrentOpenHashMap<TopicName, PersistentOfflineTopicStats> offlineTopicStatCache;
- private final ConcurrentOpenHashMap<String, ConfigField> dynamicConfigurationMap;
- private final ConcurrentOpenHashMap<String, Consumer<?>> configRegisteredListeners;
+ private final Map<TopicName, PersistentOfflineTopicStats> offlineTopicStatCache = new ConcurrentHashMap<>();
+ private final Map<String, ConfigField> dynamicConfigurationMap;
+ private final Map<String, Consumer<?>> configRegisteredListeners = new ConcurrentHashMap<>();
private final ConcurrentLinkedQueue<TopicLoadingContext> pendingTopicLoadingQueue;
@@ -297,7 +295,7 @@ public class BrokerService implements Closeable {
private final int maxUnackedMessages;
public final int maxUnackedMsgsPerDispatcher;
private final AtomicBoolean blockedDispatcherOnHighUnackedMsgs = new AtomicBoolean(false);
- private final ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers> blockedDispatchers;
+ private final Set<PersistentDispatcherMultipleConsumers> blockedDispatchers = ConcurrentHashMap.newKeySet();
private final ReadWriteLock lock = new ReentrantReadWriteLock();
@VisibleForTesting
private final DelayedDeliveryTrackerFactory delayedDeliveryTrackerFactory;
@@ -335,28 +333,9 @@ public class BrokerService implements Closeable {
this.preciseTopicPublishRateLimitingEnable =
pulsar.getConfiguration().isPreciseTopicPublishRateLimiterEnable();
this.managedLedgerFactory = pulsar.getManagedLedgerFactory();
- this.topics =
- ConcurrentOpenHashMap.<String, CompletableFuture<Optional<Topic>>>newBuilder()
- .build();
- this.replicationClients =
- ConcurrentOpenHashMap.<String, PulsarClient>newBuilder().build();
- this.clusterAdmins =
- ConcurrentOpenHashMap.<String, PulsarAdmin>newBuilder().build();
this.keepAliveIntervalSeconds = pulsar.getConfiguration().getKeepAliveIntervalSeconds();
- this.configRegisteredListeners =
- ConcurrentOpenHashMap.<String, Consumer<?>>newBuilder().build();
this.pendingTopicLoadingQueue = Queues.newConcurrentLinkedQueue();
-
- this.multiLayerTopicsMap = ConcurrentOpenHashMap.<String,
- ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>>>newBuilder()
- .build();
- this.owningTopics = ConcurrentOpenHashMap.<String,
- ConcurrentOpenHashSet<Integer>>newBuilder()
- .build();
this.pulsarStats = new PulsarStats(pulsar);
- this.offlineTopicStatCache =
- ConcurrentOpenHashMap.<TopicName,
- PersistentOfflineTopicStats>newBuilder().build();
this.topicOrderedExecutor = OrderedExecutor.newBuilder()
.numThreads(pulsar.getConfiguration().getTopicOrderedExecutorThreadNum())
@@ -403,8 +382,6 @@ public class BrokerService implements Closeable {
.build();
this.authenticationService = new AuthenticationService(pulsar.getConfiguration(),
pulsar.getOpenTelemetry().getOpenTelemetry());
- this.blockedDispatchers =
- ConcurrentOpenHashSet.<PersistentDispatcherMultipleConsumers>newBuilder().build();
this.topicFactory = createPersistentTopicFactory();
// update dynamic configuration and register-listener
updateConfigurationAndRegisterListeners();
@@ -500,7 +477,7 @@ public class BrokerService implements Closeable {
public void addTopicEventListener(TopicEventsListener... listeners) {
topicEventsDispatcher.addTopicEventListener(listeners);
- getTopics().keys().forEach(topic ->
+ topics.keySet().forEach(topic ->
TopicEventsDispatcher.notify(listeners, topic, TopicEvent.LOAD, EventStage.SUCCESS, null));
}
@@ -572,13 +549,8 @@ public class BrokerService implements Closeable {
}
public Map<String, TopicStatsImpl> getTopicStats(NamespaceBundle bundle) {
- ConcurrentOpenHashMap<String, Topic> topicMap = getMultiLayerTopicMap()
- .computeIfAbsent(bundle.getNamespaceObject().toString(), k -> {
- return ConcurrentOpenHashMap
- .<String, ConcurrentOpenHashMap<String, Topic>>newBuilder().build();
- }).computeIfAbsent(bundle.toString(), k -> {
- return ConcurrentOpenHashMap.<String, Topic>newBuilder().build();
- });
+ final var topicMap = multiLayerTopicsMap.computeIfAbsent(bundle.getNamespaceObject().toString(),
+ __ -> new ConcurrentHashMap<>()).computeIfAbsent(bundle.toString(), __ -> new ConcurrentHashMap<>());
Map<String, TopicStatsImpl> topicStatsMap = new HashMap<>();
topicMap.forEach((name, topic) -> {
@@ -2047,14 +2019,10 @@ public class BrokerService implements Closeable {
if (namespaceBundle != null) {
synchronized (multiLayerTopicsMap) {
String serviceUnit = namespaceBundle.toString();
- multiLayerTopicsMap //
- .computeIfAbsent(topicName.getNamespace(),
- k -> ConcurrentOpenHashMap.<String,
- ConcurrentOpenHashMap<String, Topic>>newBuilder()
- .build()) //
- .computeIfAbsent(serviceUnit,
- k -> ConcurrentOpenHashMap.<String, Topic>newBuilder().build()) //
- .put(topicName.toString(), topic);
+ multiLayerTopicsMap.computeIfAbsent(topicName.getNamespace(),
+ __ -> new ConcurrentHashMap<>()
+ ).computeIfAbsent(serviceUnit, __ -> new ConcurrentHashMap<>()
+ ).put(topicName.toString(), topic);
}
}
invalidateOfflineTopicStatCache(topicName);
@@ -2353,7 +2321,7 @@ public class BrokerService implements Closeable {
}
public void cleanUnloadedTopicFromCache(NamespaceBundle serviceUnit) {
- for (String topic : topics.keys()) {
+ for (String topic : topics.keySet()) {
TopicName topicName = TopicName.get(topic);
if (serviceUnit.includes(topicName) && getTopicReference(topic).isPresent()) {
log.info("[{}][{}] Clean unloaded topic from cache.", serviceUnit.toString(), topic);
@@ -2418,10 +2386,9 @@ public class BrokerService implements Closeable {
topicEventsDispatcher.notify(topic, TopicEvent.UNLOAD, EventStage.BEFORE);
synchronized (multiLayerTopicsMap) {
- ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>> namespaceMap = multiLayerTopicsMap
- .get(namespaceName);
+ final var namespaceMap = multiLayerTopicsMap.get(namespaceName);
if (namespaceMap != null) {
- ConcurrentOpenHashMap<String, Topic> bundleMap = namespaceMap.get(bundleName);
+ final var bundleMap = namespaceMap.get(bundleName);
if (bundleMap != null) {
bundleMap.remove(topic);
if (bundleMap.isEmpty()) {
@@ -2462,10 +2429,6 @@ public class BrokerService implements Closeable {
return this.numberOfNamespaceBundles;
}
- public ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> getTopics() {
- return topics;
- }
-
private void handleMetadataChanges(Notification n) {
if (n.getType() == NotificationType.Modified && NamespaceResources.pathIsFromNamespace(n.getPath())) {
@@ -2660,10 +2623,6 @@ public class BrokerService implements Closeable {
return workerGroup;
}
- public ConcurrentOpenHashMap<String, PulsarClient> getReplicationClients() {
- return replicationClients;
- }
-
public boolean isAuthenticationEnabled() {
return pulsar.getConfiguration().isAuthenticationEnabled();
}
@@ -2693,17 +2652,17 @@ public class BrokerService implements Closeable {
}
public List<Topic> getAllTopicsFromNamespaceBundle(String namespace, String bundle) {
- ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>> map1 = multiLayerTopicsMap.get(namespace);
+ final var map1 = multiLayerTopicsMap.get(namespace);
if (map1 == null) {
return Collections.emptyList();
}
- ConcurrentOpenHashMap<String, Topic> map2 = map1.get(bundle);
+ final var map2 = map1.get(bundle);
if (map2 == null) {
return Collections.emptyList();
}
- return map2.values();
+ return map2.values().stream().toList();
}
/**
@@ -3080,60 +3039,17 @@ public class BrokerService implements Closeable {
}
}
- /**
- * Updates pulsar.ServiceConfiguration's dynamic field with value persistent into zk-dynamic path. It also validates
- * dynamic-value before updating it and throws {@code IllegalArgumentException} if validation fails
- */
- private void updateDynamicServiceConfiguration() {
- Optional<Map<String, String>> configCache = Optional.empty();
-
- try {
- configCache =
- pulsar().getPulsarResources().getDynamicConfigResources().getDynamicConfiguration();
-
- // create dynamic-config if not exist.
- if (!configCache.isPresent()) {
- pulsar().getPulsarResources().getDynamicConfigResources()
- .setDynamicConfigurationWithCreate(n -> new HashMap<>());
- }
- } catch (Exception e) {
- log.warn("Failed to read dynamic broker configuration", e);
- }
-
- configCache.ifPresent(stringStringMap -> stringStringMap.forEach((key, value) -> {
- // validate field
- if (dynamicConfigurationMap.containsKey(key) && dynamicConfigurationMap.get(key).validator != null) {
- if (!dynamicConfigurationMap.get(key).validator.test(value)) {
- log.error("Failed to validate dynamic config {} with value {}", key, value);
- throw new IllegalArgumentException(
- String.format("Failed to validate dynamic-config %s/%s", key, value));
- }
- }
- // update field value
- try {
- Field field = ServiceConfiguration.class.getDeclaredField(key);
- if (field != null && field.isAnnotationPresent(FieldContext.class)) {
- field.setAccessible(true);
- field.set(pulsar().getConfiguration(), FieldParser.value(value, field));
- log.info("Successfully updated {}/{}", key, value);
- }
- } catch (Exception e) {
- log.warn("Failed to update service configuration {}/{}, {}", key, value, e.getMessage());
- }
- }));
- }
-
public DelayedDeliveryTrackerFactory getDelayedDeliveryTrackerFactory() {
return delayedDeliveryTrackerFactory;
}
public List<String> getDynamicConfiguration() {
- return dynamicConfigurationMap.keys();
+ return dynamicConfigurationMap.keySet().stream().toList();
}
public Map<String, String> getRuntimeConfiguration() {
Map<String, String> configMap = new HashMap<>();
- ConcurrentOpenHashMap<String, Object> runtimeConfigurationMap = getRuntimeConfigurationMap();
+ ConcurrentHashMap<String, Object> runtimeConfigurationMap = getRuntimeConfigurationMap();
runtimeConfigurationMap.forEach((key, value) -> {
configMap.put(key, String.valueOf(value));
});
@@ -3151,9 +3067,8 @@ public class BrokerService implements Closeable {
return true;
}
- private ConcurrentOpenHashMap<String, ConfigField> prepareDynamicConfigurationMap() {
- ConcurrentOpenHashMap<String, ConfigField> dynamicConfigurationMap =
- ConcurrentOpenHashMap.<String, ConfigField>newBuilder().build();
+ private Map<String, ConfigField> prepareDynamicConfigurationMap() {
+ final var dynamicConfigurationMap = new ConcurrentHashMap<String, ConfigField>();
try {
for (Field field : ServiceConfiguration.class.getDeclaredFields()) {
if (field != null && field.isAnnotationPresent(FieldContext.class)) {
@@ -3172,9 +3087,8 @@ public class BrokerService implements Closeable {
return dynamicConfigurationMap;
}
- private ConcurrentOpenHashMap<String, Object> getRuntimeConfigurationMap() {
- ConcurrentOpenHashMap<String, Object> runtimeConfigurationMap =
- ConcurrentOpenHashMap.<String, Object>newBuilder().build();
+ private ConcurrentHashMap<String, Object> getRuntimeConfigurationMap() {
+ final var runtimeConfigurationMap = new ConcurrentHashMap<String, Object>();
for (Field field : ServiceConfiguration.class.getDeclaredFields()) {
if (field != null && field.isAnnotationPresent(FieldContext.class)) {
field.setAccessible(true);
@@ -3337,11 +3251,6 @@ public class BrokerService implements Closeable {
return topicOrderedExecutor;
}
- public ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>>>
- getMultiLayerTopicMap() {
- return multiLayerTopicsMap;
- }
-
/**
* If per-broker unacked message reached to limit then it blocks dispatcher if its unacked message limit has been
* reached to {@link #maxUnackedMsgsPerDispatcher}.
@@ -3393,7 +3302,7 @@ public class BrokerService implements Closeable {
} else if (blockedDispatcherOnHighUnackedMsgs.get() && unAckedMessages < maxUnackedMessages / 2) {
// unblock broker-dispatching if received enough acked messages back
if (blockedDispatcherOnHighUnackedMsgs.compareAndSet(true, false)) {
- unblockDispatchersOnUnAckMessages(blockedDispatchers.values());
+ unblockDispatchersOnUnAckMessages(blockedDispatchers.stream().toList());
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java
index 7ffc7818d4c..b96e00a8909 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java
@@ -38,7 +38,6 @@ import org.apache.pulsar.broker.stats.ClusterReplicationMetrics;
import org.apache.pulsar.broker.stats.NamespaceStats;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.stats.Metrics;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.utils.StatsOutputStream;
import org.slf4j.Logger;
@@ -101,9 +100,7 @@ public class PulsarStats implements Closeable {
return clusterReplicationMetrics;
}
- public synchronized void updateStats(
- ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>>>
- topicsMap) {
+ public synchronized void updateStats(Map<String, Map<String, Map<String, Topic>>> topicsMap) {
StatsOutputStream topicStatsStream = new StatsOutputStream(tempTopicStatsBuf);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
index 25c875778c0..110a8aa82f1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
@@ -83,7 +83,7 @@ public class NamespaceStatsAggregator {
Optional<CompactorMXBean> compactorMXBean = getCompactorMXBean(pulsar);
LongAdder topicsCount = new LongAdder();
Map<String, Long> localNamespaceTopicCount = new HashMap<>();
- pulsar.getBrokerService().getMultiLayerTopicMap().forEach((namespace, bundlesMap) -> {
+ pulsar.getBrokerService().getMultiLayerTopicsMap().forEach((namespace, bundlesMap) -> {
namespaceStats.reset();
topicsCount.reset();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TransactionAggregator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TransactionAggregator.java
index 3da061f6ffe..df2638b3bb8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TransactionAggregator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TransactionAggregator.java
@@ -56,7 +56,7 @@ public class TransactionAggregator {
if (includeTopicMetrics) {
- pulsar.getBrokerService().getMultiLayerTopicMap().forEach((namespace, bundlesMap) ->
+ pulsar.getBrokerService().getMultiLayerTopicsMap().forEach((namespace, bundlesMap) ->
bundlesMap.forEach((bundle, topicsMap) -> topicsMap.forEach((name, topic) -> {
if (topic instanceof PersistentTopic) {
topic.getSubscriptions().values().forEach(subscription -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
index df9862691d6..900babbecf4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -114,9 +114,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.*;
import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl;
import org.apache.pulsar.common.protocol.schema.SchemaData;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
-import org.awaitility.reflect.WhiteboxImpl;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -2913,8 +2911,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
private AtomicInteger injectSchemaCheckCounterForTopic(String topicName) {
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- WhiteboxImpl.getInternalState(pulsar.getBrokerService(), "topics");
+ final var topics = pulsar.getBrokerService().getTopics();
AbstractTopic topic = (AbstractTopic) topics.get(topicName).join().get();
AbstractTopic spyTopic = Mockito.spy(topic);
AtomicInteger counter = new AtomicInteger();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java
index bb6ef9d3636..9861ab57237 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTrackerFactoryTest.java
@@ -24,12 +24,10 @@ import org.apache.pulsar.broker.delayed.bucket.RecoverDelayedDeliveryTrackerExce
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Dispatcher;
import org.apache.pulsar.broker.service.Subscription;
-import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.*;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.mockito.Mockito;
import org.testng.Assert;
@@ -166,11 +164,7 @@ public class DelayedDeliveryTrackerFactoryTest extends ProducerConsumerBase {
Mockito.doReturn(brokerService).when(topic).getBrokerService();
// Set Mocked topic to BrokerService
- Field topics = BrokerService.class.getDeclaredField("topics");
- topics.setAccessible(true);
- @SuppressWarnings("unchecked")
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topicMap =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) topics.get(brokerService);
+ final var topicMap = brokerService.getTopics();
topicMap.put(topicName, CompletableFuture.completedFuture(Optional.of(topic)));
// Create consumer
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
index 6b2669275df..951247bd688 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
@@ -87,7 +87,6 @@ import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.util.ObjectMapperFactory;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.metadata.api.GetResult;
import org.apache.pulsar.metadata.api.MetadataCache;
import org.apache.pulsar.metadata.api.Notification;
@@ -300,8 +299,7 @@ public class NamespaceServiceTest extends BrokerTestBase {
final String topicName = "persistent://my-property/use/my-ns/my-topic1";
pulsarClient.newConsumer().topic(topicName).subscriptionName("my-subscriber-name").subscribe();
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService()
- .getTopics();
+ final var topics = pulsar.getBrokerService().getTopics();
Topic spyTopic = spy(topics.get(topicName).get().get());
topics.clear();
CompletableFuture<Optional<Topic>> topicFuture = CompletableFuture.completedFuture(Optional.of(spyTopic));
@@ -331,7 +329,7 @@ public class NamespaceServiceTest extends BrokerTestBase {
final String topicName = "persistent://my-property/use/my-ns/my-topic1";
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName("my-subscriber-name")
.subscribe();
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService().getTopics();
+ final var topics = pulsar.getBrokerService().getTopics();
Topic spyTopic = spy(topics.get(topicName).get().get());
topics.clear();
CompletableFuture<Optional<Topic>> topicFuture = CompletableFuture.completedFuture(Optional.of(spyTopic));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
index 7415a40ad55..374296e6867 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
@@ -28,6 +28,7 @@ import io.netty.channel.DefaultEventLoop;
import io.netty.util.internal.DefaultPriorityQueue;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -45,7 +46,6 @@ import org.apache.pulsar.client.impl.ProducerBuilderImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.common.policies.data.stats.ReplicatorStatsImpl;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
import org.testng.Assert;
@@ -74,7 +74,7 @@ public class AbstractReplicatorTest {
when(remoteClient.getCnxPool()).thenReturn(connectionPool);
final ProducerConfigurationData producerConf = new ProducerConfigurationData();
final ProducerBuilderImpl producerBuilder = mock(ProducerBuilderImpl.class);
- final ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics = new ConcurrentOpenHashMap<>();
+ final var topics = new ConcurrentHashMap<String, CompletableFuture<Optional<Topic>>>();
when(broker.executor()).thenReturn(eventLoopGroup);
when(broker.getTopics()).thenReturn(topics);
when(remoteClient.newProducer(any(Schema.class))).thenReturn(producerBuilder);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
index a4460187d23..3e735ee4c85 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
@@ -26,11 +26,11 @@ import static org.testng.Assert.fail;
import java.util.List;
-import java.util.Optional;
+import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import lombok.Cleanup;
import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.client.admin.ListNamespaceTopicsOptions;
@@ -45,7 +45,6 @@ import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicType;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -552,10 +551,9 @@ public class BrokerServiceAutoTopicCreationTest extends BrokerTestBase{
admin.topics().createNonPartitionedTopic(ExtensibleLoadManagerImpl.TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
// clear the topics to test the auto creation of non-persistent topics.
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- pulsar.getBrokerService().getTopics();
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> oldTopics = new ConcurrentOpenHashMap<>();
- topics.forEach((key, val) -> oldTopics.put(key, val));
+ final var topics = pulsar.getBrokerService().getTopics();
+ final var oldTopics = topics.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
+ Map.Entry::getValue));
topics.clear();
// The created persistent topic correctly can be found by
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 5398b5aa57b..17209c83c13 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -1630,7 +1630,7 @@ public class BrokerServiceTest extends BrokerTestBase {
producer1.close();
PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName.toString(), false).get().get();
persistentTopic.close().join();
- List<String> topics = new ArrayList<>(pulsar.getBrokerService().getTopics().keys());
+ List<String> topics = new ArrayList<>(pulsar.getBrokerService().getTopics().keySet());
topics.removeIf(item -> item.contains(SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME));
Assert.assertEquals(topics.size(), 0);
@Cleanup
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index d684b4af7c2..a8f8d7ecbbd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -92,7 +92,6 @@ import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.policies.data.impl.AutoTopicCreationOverrideImpl;
import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
import org.glassfish.jersey.client.JerseyClient;
@@ -293,8 +292,7 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
});
// Inject spy client.
- ConcurrentOpenHashMap<String, PulsarClient>
- replicationClients = WhiteboxImpl.getInternalState(brokerService, "replicationClients");
+ final var replicationClients = brokerService.getReplicationClients();
PulsarClientImpl internalClient = (PulsarClientImpl) replicationClients.get(cluster2);
PulsarClient spyClient = spy(internalClient);
assertTrue(replicationClients.remove(cluster2, internalClient));
@@ -1141,9 +1139,9 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
return t.startsWith(tp);
};
Awaitility.await().untilAsserted(() -> {
- List<String> topics1 = pulsar1.getBrokerService().getTopics().keys()
+ List<String> topics1 = pulsar1.getBrokerService().getTopics().keySet()
.stream().filter(topicNameFilter).collect(Collectors.toList());
- List<String> topics2 = pulsar2.getBrokerService().getTopics().keys()
+ List<String> topics2 = pulsar2.getBrokerService().getTopics().keySet()
.stream().filter(topicNameFilter).collect(Collectors.toList());
Collections.sort(topics1);
Collections.sort(topics2);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java
index a5d14ca0487..c214378fd94 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java
@@ -23,10 +23,8 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import java.util.List;
import java.util.Optional;
-import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.internal.PulsarAdminImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -55,7 +53,7 @@ public class ReplicatorAdminTlsTest extends ReplicatorTestBase {
ns.getClusterPulsarAdmin(cluster3, Optional.of(admin1.clusters().getCluster(cluster3)));
// verify the admin
- ConcurrentOpenHashMap<String, PulsarAdmin> clusterAdmins = ns.getClusterAdmins();
+ final var clusterAdmins = ns.getClusterAdmins();
assertFalse(clusterAdmins.isEmpty());
clusterAdmins.forEach((cluster, admin) -> {
ClientConfigurationData clientConfigData = ((PulsarAdminImpl) admin).getClientConfigData();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java
index 3d3eb3faa72..451bdd10eb1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java
@@ -23,10 +23,8 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import java.util.List;
import java.util.Optional;
-import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.internal.PulsarAdminImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -56,7 +54,7 @@ public class ReplicatorAdminTlsWithKeyStoreTest extends ReplicatorTestBase {
ns.getClusterPulsarAdmin(cluster3, Optional.of(admin1.clusters().getCluster(cluster3)));
// verify the admin
- ConcurrentOpenHashMap<String, PulsarAdmin> clusterAdmins = ns.getClusterAdmins();
+ final var clusterAdmins = ns.getClusterAdmins();
assertFalse(clusterAdmins.isEmpty());
clusterAdmins.forEach((cluster, admin) -> {
ClientConfigurationData clientConfigData = ((PulsarAdminImpl) admin).getClientConfigData();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
index 514e0207fbf..a1f147cbb62 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
@@ -34,7 +34,6 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.pulsar.common.naming.TopicName;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -169,9 +168,9 @@ public class ReplicatorGlobalNSTest extends ReplicatorTestBase {
Thread.sleep(1000L);
// Make sure that the internal replicators map contains remote cluster info
- ConcurrentOpenHashMap<String, PulsarClient> replicationClients1 = ns1.getReplicationClients();
- ConcurrentOpenHashMap<String, PulsarClient> replicationClients2 = ns2.getReplicationClients();
- ConcurrentOpenHashMap<String, PulsarClient> replicationClients3 = ns3.getReplicationClients();
+ final var replicationClients1 = ns1.getReplicationClients();
+ final var replicationClients2 = ns2.getReplicationClients();
+ final var replicationClients3 = ns3.getReplicationClients();
Assert.assertNotNull(replicationClients1.get("r2"));
Assert.assertNotNull(replicationClients1.get("r3"));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index 1c47abab775..8e115e14b37 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -234,11 +234,7 @@ public class ReplicatorTest extends ReplicatorTestBase {
final Method startRepl = PersistentTopic.class.getDeclaredMethod("startReplicator", String.class);
startRepl.setAccessible(true);
- Field replClientField = BrokerService.class.getDeclaredField("replicationClients");
- replClientField.setAccessible(true);
- ConcurrentOpenHashMap<String, PulsarClient> replicationClients =
- (ConcurrentOpenHashMap<String, PulsarClient>) replClientField
- .get(pulsar1.getBrokerService());
+ final var replicationClients = pulsar1.getBrokerService().getReplicationClients();
replicationClients.put("r3", pulsarClient);
admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2", "r3"));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java
index ddc5eeab1d2..f1940a28999 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java
@@ -27,7 +27,6 @@ import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
-import java.lang.reflect.Field;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -45,7 +44,6 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.TopicName;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
import org.testng.Assert;
@@ -591,12 +589,7 @@ public class TopicDuplicationTest extends ProducerConsumerBase {
return false;
});
- Field field2 = BrokerService.class.getDeclaredField("topics");
- field2.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>)
- field2.get(pulsar.getBrokerService());
-
+ final var topics = pulsar.getBrokerService().getTopics();
try {
pulsar.getBrokerService().getTopic(topic, false).join().get();
Assert.fail();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
index 4df2d36a953..a92f5a4acc2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -424,16 +424,16 @@ public class PrometheusMetricsTest extends BrokerTestBase {
// There should be 2 metrics with different tags for each topic
List<Metric> cm = (List<Metric>) metrics.get("pulsar_storage_write_latency_le_1");
assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
cm = (List<Metric>) metrics.get("pulsar_producers_count");
assertEquals(cm.size(), 2);
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
- assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
cm = (List<Metric>) metrics.get("pulsar_topic_load_times_count");
@@ -446,33 +446,33 @@ public class PrometheusMetricsTest extends BrokerTestBase {
cm = (List<Metric>) metrics.get("pulsar_in_bytes_total");
assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
cm = (List<Metric>) metrics.get("pulsar_in_messages_total");
assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
cm = (List<Metric>) metrics.get("pulsar_out_bytes_total");
assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
assertEquals(cm.get(0).tags.get("subscription"), "test");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
assertEquals(cm.get(1).tags.get("subscription"), "test");
cm = (List<Metric>) metrics.get("pulsar_out_messages_total");
assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
assertEquals(cm.get(0).tags.get("subscription"), "test");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
assertEquals(cm.get(1).tags.get("subscription"), "test");
@@ -1086,26 +1086,26 @@ public class PrometheusMetricsTest extends BrokerTestBase {
List<Metric> cm = (List<Metric>) metrics.get("pulsar_producer_msg_rate_in");
assertEquals(cm.size(), 2);
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
- assertEquals(cm.get(0).tags.get("producer_name"), "producer2");
- assertEquals(cm.get(0).tags.get("producer_id"), "1");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(0).tags.get("producer_name"), "producer1");
+ assertEquals(cm.get(0).tags.get("producer_id"), "0");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
- assertEquals(cm.get(1).tags.get("producer_name"), "producer1");
- assertEquals(cm.get(1).tags.get("producer_id"), "0");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(1).tags.get("producer_name"), "producer2");
+ assertEquals(cm.get(1).tags.get("producer_id"), "1");
cm = (List<Metric>) metrics.get("pulsar_producer_msg_throughput_in");
assertEquals(cm.size(), 2);
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
- assertEquals(cm.get(0).tags.get("producer_name"), "producer2");
- assertEquals(cm.get(0).tags.get("producer_id"), "1");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(0).tags.get("producer_name"), "producer1");
+ assertEquals(cm.get(0).tags.get("producer_id"), "0");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
- assertEquals(cm.get(1).tags.get("producer_name"), "producer1");
- assertEquals(cm.get(1).tags.get("producer_id"), "0");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(1).tags.get("producer_name"), "producer2");
+ assertEquals(cm.get(1).tags.get("producer_id"), "1");
p1.close();
p2.close();
@@ -1155,42 +1155,42 @@ public class PrometheusMetricsTest extends BrokerTestBase {
List<Metric> cm = (List<Metric>) metrics.get("pulsar_out_bytes_total");
assertEquals(cm.size(), 4);
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(0).tags.get("subscription"), "test");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(1).tags.get("subscription"), "test");
- assertEquals(cm.get(1).tags.get("consumer_id"), "1");
+ assertEquals(cm.get(1).tags.get("consumer_id"), "0");
assertEquals(cm.get(2).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(2).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(2).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(2).tags.get("subscription"), "test");
assertEquals(cm.get(3).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(3).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(3).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(3).tags.get("subscription"), "test");
- assertEquals(cm.get(3).tags.get("consumer_id"), "0");
+ assertEquals(cm.get(3).tags.get("consumer_id"), "1");
cm = (List<Metric>) metrics.get("pulsar_out_messages_total");
assertEquals(cm.size(), 4);
assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(0).tags.get("subscription"), "test");
assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
+ assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
assertEquals(cm.get(1).tags.get("subscription"), "test");
- assertEquals(cm.get(1).tags.get("consumer_id"), "1");
+ assertEquals(cm.get(1).tags.get("consumer_id"), "0");
assertEquals(cm.get(2).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(2).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(2).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(2).tags.get("subscription"), "test");
assertEquals(cm.get(3).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(3).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
+ assertEquals(cm.get(3).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
assertEquals(cm.get(3).tags.get("subscription"), "test");
- assertEquals(cm.get(3).tags.get("consumer_id"), "0");
+ assertEquals(cm.get(3).tags.get("consumer_id"), "1");
p1.close();
p2.close();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java
index cf923df0411..45e3fb253bf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregatorTest.java
@@ -21,6 +21,8 @@ package org.apache.pulsar.broker.stats.prometheus;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.when;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl;
import org.apache.bookkeeper.mledger.util.StatsBuckets;
@@ -47,17 +49,14 @@ import org.testng.annotations.Test;
public class NamespaceStatsAggregatorTest {
protected PulsarService pulsar;
private BrokerService broker;
- private ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>>>
- multiLayerTopicsMap;
+ private Map<String, Map<String, Map<String, Topic>>> multiLayerTopicsMap;
@BeforeMethod(alwaysRun = true)
public void setup() throws Exception {
- multiLayerTopicsMap = ConcurrentOpenHashMap.<String,
- ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>>>newBuilder()
- .build();
+ multiLayerTopicsMap = new ConcurrentHashMap<>();
pulsar = Mockito.mock(PulsarService.class);
broker = Mockito.mock(BrokerService.class);
- doReturn(multiLayerTopicsMap).when(broker).getMultiLayerTopicMap();
+ doReturn(multiLayerTopicsMap).when(broker).getMultiLayerTopicsMap();
Mockito.when(pulsar.getLocalMetadataStore()).thenReturn(Mockito.mock(ZKMetadataStore.class));
ServiceConfiguration mockConfig = Mockito.mock(ServiceConfiguration.class);
doReturn(mockConfig).when(pulsar).getConfiguration();
@@ -70,8 +69,8 @@ public class NamespaceStatsAggregatorTest {
final String namespace = "tenant/cluster/ns";
// prepare multi-layer topic map
- ConcurrentOpenHashMap bundlesMap = ConcurrentOpenHashMap.newBuilder().build();
- ConcurrentOpenHashMap topicsMap = ConcurrentOpenHashMap.newBuilder().build();
+ final var bundlesMap = new ConcurrentHashMap<String, Map<String, Topic>>();
+ final var topicsMap = new ConcurrentHashMap<String, Topic>();
ConcurrentOpenHashMap subscriptionsMaps = ConcurrentOpenHashMap.newBuilder().build();
bundlesMap.put("my-bundle", topicsMap);
multiLayerTopicsMap.put(namespace, bundlesMap);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java
index 3924281c094..f21e11b9802 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java
@@ -93,7 +93,6 @@ import org.apache.pulsar.common.events.EventType;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -201,20 +200,14 @@ public class TopicTransactionBufferRecoverTest extends TransactionTestBase {
Awaitility.await().until(() -> {
for (int i = 0; i < getPulsarServiceList().size(); i++) {
- Field field = BrokerService.class.getDeclaredField("topics");
- field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> completableFuture = topics.get("persistent://" + testTopic);
if (completableFuture != null) {
Optional<Topic> topic = completableFuture.get();
if (topic.isPresent()) {
PersistentTopic persistentTopic = (PersistentTopic) topic.get();
- field = PersistentTopic.class.getDeclaredField("transactionBuffer");
- field.setAccessible(true);
TopicTransactionBuffer topicTransactionBuffer =
- (TopicTransactionBuffer) field.get(persistentTopic);
+ (TopicTransactionBuffer) persistentTopic.getTransactionBuffer();
if (topicTransactionBuffer.checkIfReady()) {
return true;
} else {
@@ -455,17 +448,13 @@ public class TopicTransactionBufferRecoverTest extends TransactionTestBase {
assertTrue(((MessageIdImpl) messageId2).getLedgerId() != ((MessageIdImpl) messageId1).getLedgerId());
boolean exist = false;
for (int i = 0; i < getPulsarServiceList().size(); i++) {
- Field field = BrokerService.class.getDeclaredField("topics");
- field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> completableFuture = topics.get("persistent://" + ABORT_DELETE);
if (completableFuture != null) {
Optional<Topic> topic = completableFuture.get();
if (topic.isPresent()) {
PersistentTopic persistentTopic = (PersistentTopic) topic.get();
- field = ManagedLedgerImpl.class.getDeclaredField("ledgers");
+ var field = ManagedLedgerImpl.class.getDeclaredField("ledgers");
field.setAccessible(true);
NavigableMap<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo> ledgers
= (NavigableMap<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo>) field.get(persistentTopic.getManagedLedger());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
index 39f36f4d38c..14b1d563c11 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
@@ -390,7 +390,7 @@ public class TransactionProduceTest extends TransactionTestBase {
int pendingAckCount = 0;
for (PulsarService pulsarService : getPulsarServiceList()) {
- for (String key : pulsarService.getBrokerService().getTopics().keys()) {
+ for (String key : pulsarService.getBrokerService().getTopics().keySet()) {
if (key.contains(topic)) {
Field field = clazz.getDeclaredField("pendingAckHandle");
field.setAccessible(true);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
index cc09fa21219..5480b1a21d5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
@@ -145,7 +145,6 @@ import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl;
import org.apache.pulsar.common.schema.SchemaInfo;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.compaction.CompactionServiceFactory;
import org.apache.pulsar.compaction.PulsarCompactionServiceFactory;
import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
@@ -385,8 +384,7 @@ public class TransactionTest extends TransactionTestBase {
return true;
});
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- getPulsarServiceList().get(0).getBrokerService().getTopics();
+ final var topics = getPulsarServiceList().get(0).getBrokerService().getTopics();
Assert.assertNull(topics.get(TopicName.get(TopicDomain.persistent.value(),
NamespaceName.SYSTEM_NAMESPACE, TRANSACTION_LOG_PREFIX).toString() + 0));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java
index 2e6d9c61bde..818b854ffe9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionLowWaterMarkTest.java
@@ -33,7 +33,6 @@ import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.commons.collections4.map.LinkedMap;
-import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -54,7 +53,6 @@ import org.apache.pulsar.client.impl.transaction.TransactionImpl;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID;
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStore;
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStoreState;
@@ -215,18 +213,14 @@ public class TransactionLowWaterMarkTest extends TransactionTestBase {
LinkedMap<TxnID, HashMap<Position, Position>> individualAckOfTransaction = null;
for (int i = 0; i < getPulsarServiceList().size(); i++) {
- Field field = BrokerService.class.getDeclaredField("topics");
- field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> completableFuture = topics.get(TOPIC);
if (completableFuture != null) {
Optional<Topic> topic = completableFuture.get();
if (topic.isPresent()) {
PersistentSubscription persistentSubscription = (PersistentSubscription) topic.get()
.getSubscription(subName);
- field = PersistentSubscription.class.getDeclaredField("pendingAckHandle");
+ var field = PersistentSubscription.class.getDeclaredField("pendingAckHandle");
field.setAccessible(true);
PendingAckHandleImpl pendingAckHandle = (PendingAckHandleImpl) field.get(persistentSubscription);
field = PendingAckHandleImpl.class.getDeclaredField("individualAckOfTransaction");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java
index fd4e984b6c1..58cf59aa6b3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker.transaction.pendingack;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
-import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -35,7 +34,6 @@ import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.commons.collections4.map.LinkedMap;
import org.apache.commons.lang3.tuple.MutablePair;
-import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.transaction.TransactionTestBase;
@@ -48,7 +46,6 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.util.collections.BitSetRecyclable;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -120,18 +117,14 @@ public class PendingAckInMemoryDeleteTest extends TransactionTestBase {
int count = 0;
for (int i = 0; i < getPulsarServiceList().size(); i++) {
- Field field = BrokerService.class.getDeclaredField("topics");
- field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> completableFuture = topics.get("persistent://" + normalTopic);
if (completableFuture != null) {
Optional<Topic> topic = completableFuture.get();
if (topic.isPresent()) {
PersistentSubscription persistentSubscription = (PersistentSubscription) topic.get()
.getSubscription(subscriptionName);
- field = PersistentSubscription.class.getDeclaredField("pendingAckHandle");
+ var field = PersistentSubscription.class.getDeclaredField("pendingAckHandle");
field.setAccessible(true);
PendingAckHandleImpl pendingAckHandle = (PendingAckHandleImpl) field.get(persistentSubscription);
field = PendingAckHandleImpl.class.getDeclaredField("individualAckOfTransaction");
@@ -214,18 +207,14 @@ public class PendingAckInMemoryDeleteTest extends TransactionTestBase {
commitTxn.commit().get();
int count = 0;
for (int i = 0; i < getPulsarServiceList().size(); i++) {
- Field field = BrokerService.class.getDeclaredField("topics");
- field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> completableFuture = topics.get("persistent://" + normalTopic);
if (completableFuture != null) {
Optional<Topic> topic = completableFuture.get();
if (topic.isPresent()) {
PersistentSubscription testPersistentSubscription =
(PersistentSubscription) topic.get().getSubscription(subscriptionName);
- field = PersistentSubscription.class.getDeclaredField("pendingAckHandle");
+ var field = PersistentSubscription.class.getDeclaredField("pendingAckHandle");
field.setAccessible(true);
pendingAckHandle = (PendingAckHandleImpl) field.get(testPersistentSubscription);
field = PendingAckHandleImpl.class.getDeclaredField("individualAckOfTransaction");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
index 424081b904c..e04dde65fa8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
@@ -996,7 +996,7 @@ public class TopicReaderTest extends ProducerConsumerBase {
}
// cause broker to drop topic. Will be loaded next time we access it
- pulsar.getBrokerService().getTopics().keys().forEach(topicName -> {
+ pulsar.getBrokerService().getTopics().keySet().forEach(topicName -> {
try {
pulsar.getBrokerService().getTopicReference(topicName).get().close(false).get();
} catch (Exception e) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
index 06c6069ebae..1e8754a2d67 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
@@ -48,7 +48,6 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.NavigableMap;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
@@ -101,7 +100,6 @@ import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.protocol.PulsarHandler;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@@ -697,8 +695,7 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
@Test(expectedExceptions = PulsarClientException.TimeoutException.class)
public void testOperationTimeout() throws PulsarClientException {
final String topicName = "persistent://my-property/my-ns/my-topic1";
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService()
- .getTopics();
+ final var topics = pulsar.getBrokerService().getTopics();
// non-complete topic future so, create topic should timeout
topics.put(topicName, new CompletableFuture<>());
try (PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl.toString())
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
index 812f8fd571c..f490fa70539 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
@@ -73,7 +73,6 @@ import org.apache.pulsar.client.util.RetryMessageUtil;
import org.apache.pulsar.common.api.proto.CommandAck;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID;
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStore;
import org.apache.pulsar.transaction.coordinator.TransactionSubscription;
@@ -415,11 +414,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
boolean exist = false;
for (int i = 0; i < getPulsarServiceList().size(); i++) {
- Field field = BrokerService.class.getDeclaredField("topics");
- field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
if (topicFuture != null) {
@@ -722,9 +717,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
Field field = BrokerService.class.getDeclaredField("topics");
field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
if (topicFuture != null) {
@@ -1193,9 +1186,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
Field field = BrokerService.class.getDeclaredField("topics");
field.setAccessible(true);
- ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
- (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
- .get(getPulsarServiceList().get(i).getBrokerService());
+ final var topics = getPulsarServiceList().get(i).getBrokerService().getTopics();
CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
if (topicFuture != null) {