You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/04/26 20:37:07 UTC
[GitHub] sijie closed pull request #1652: Refactored BrokerService to use
Optional in topics map
sijie closed pull request #1652: Refactored BrokerService to use Optional in topics map
URL: https://github.com/apache/incubator-pulsar/pull/1652
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
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 f47796d93..683bb0331 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
@@ -137,7 +137,7 @@
private final int port;
private final int tlsPort;
- private final ConcurrentOpenHashMap<String, CompletableFuture<Topic>> topics;
+ private final ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics;
private final ConcurrentOpenHashMap<String, PulsarClient> replicationClients;
@@ -154,7 +154,7 @@
private static final ConcurrentOpenHashMap<String, ConfigField> dynamicConfigurationMap = prepareDynamicConfigurationMap();
private final ConcurrentOpenHashMap<String, Consumer<?>> configRegisteredListeners;
- private final ConcurrentLinkedQueue<Pair<String, CompletableFuture<Topic>>> pendingTopicLoadingQueue;
+ private final ConcurrentLinkedQueue<Pair<String, CompletableFuture<Optional<Topic>>>> pendingTopicLoadingQueue;
private AuthorizationService authorizationService = null;
private final ScheduledExecutorService statsUpdater;
@@ -422,19 +422,19 @@ public void unloadNamespaceBundlesGracefully() {
}
public CompletableFuture<Optional<Topic>> getTopicIfExists(final String topic) {
- return getTopic(topic, false /* createIfMissing */ ).thenApply(t -> Optional.ofNullable(t));
+ return getTopic(topic, false /* createIfMissing */ );
}
public CompletableFuture<Topic> getOrCreateTopic(final String topic) {
- return getTopic(topic, true /* createIfMissing */ );
+ return getTopic(topic, true /* createIfMissing */ ).thenApply(Optional::get);
}
- private CompletableFuture<Topic> getTopic(final String topic, boolean createIfMissing) {
+ private CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean createIfMissing) {
try {
- CompletableFuture<Topic> topicFuture = topics.get(topic);
+ CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
if (topicFuture != null) {
if (topicFuture.isCompletedExceptionally()
- || (topicFuture.isDone() && topicFuture.getNow(null) == null)) {
+ || (topicFuture.isDone() && !topicFuture.getNow(Optional.empty()).isPresent())) {
// Exceptional topics should be recreated.
topics.remove(topic, topicFuture);
} else {
@@ -461,8 +461,8 @@ public void unloadNamespaceBundlesGracefully() {
}
}
- private CompletableFuture<Topic> createNonPersistentTopic(String topic) {
- CompletableFuture<Topic> topicFuture = new CompletableFuture<Topic>();
+ private CompletableFuture<Optional<Topic>> createNonPersistentTopic(String topic) {
+ CompletableFuture<Optional<Topic>> topicFuture = new CompletableFuture<>();
if (!pulsar.getConfiguration().isEnableNonPersistentTopics()) {
if (log.isDebugEnabled()) {
@@ -480,7 +480,7 @@ public void unloadNamespaceBundlesGracefully() {
long topicLoadLatencyMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()) - topicCreateTimeMs;
pulsarStats.recordTopicLoadTimeValue(topic, topicLoadLatencyMs);
addTopicToStatsMaps(TopicName.get(topic), nonPersistentTopic);
- topicFuture.complete(nonPersistentTopic);
+ topicFuture.complete(Optional.of(nonPersistentTopic));
});
replicationFuture.exceptionally((ex) -> {
log.warn("Replication check failed. Removing topic from topics list {}, {}", topic, ex);
@@ -549,10 +549,10 @@ public PulsarClient getReplicationClient(String cluster) {
* @return CompletableFuture<Topic>
* @throws RuntimeException
*/
- protected CompletableFuture<Topic> loadOrCreatePersistentTopic(final String topic, boolean createIfMissing) throws RuntimeException {
+ protected CompletableFuture<Optional<Topic>> loadOrCreatePersistentTopic(final String topic, boolean createIfMissing) throws RuntimeException {
checkTopicNsOwnership(topic);
- final CompletableFuture<Topic> topicFuture = new CompletableFuture<>();
+ final CompletableFuture<Optional<Topic>> topicFuture = new CompletableFuture<>();
if (!pulsar.getConfiguration().isEnablePersistentTopics()) {
if (log.isDebugEnabled()) {
log.debug("Broker is unable to load persistent topic {}", topic);
@@ -572,7 +572,7 @@ public PulsarClient getReplicationClient(String cluster) {
return null;
});
} else {
- pendingTopicLoadingQueue.add(new ImmutablePair<String, CompletableFuture<Topic>>(topic, topicFuture));
+ pendingTopicLoadingQueue.add(new ImmutablePair<String, CompletableFuture<Optional<Topic>>>(topic, topicFuture));
if (log.isDebugEnabled()) {
log.debug("topic-loading for {} added into pending queue", topic);
}
@@ -580,7 +580,7 @@ public PulsarClient getReplicationClient(String cluster) {
return topicFuture;
}
- private void createPersistentTopic(final String topic, boolean createIfMissing, CompletableFuture<Topic> topicFuture) {
+ private void createPersistentTopic(final String topic, boolean createIfMissing, CompletableFuture<Optional<Topic>> topicFuture) {
final long topicCreateTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
TopicName topicName = TopicName.get(topic);
@@ -615,7 +615,7 @@ public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
- topicCreateTimeMs;
pulsarStats.recordTopicLoadTimeValue(topic, topicLoadLatencyMs);
addTopicToStatsMaps(topicName, persistentTopic);
- topicFuture.complete(persistentTopic);
+ topicFuture.complete(Optional.of(persistentTopic));
}).exceptionally((ex) -> {
log.warn(
"Replication or dedup check failed. Removing topic from topics list {}, {}",
@@ -638,7 +638,7 @@ public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
if (!createIfMissing && exception instanceof ManagedLedgerNotFoundException) {
// We were just trying to load a topic and the topic doesn't exist
- topicFuture.complete(null);
+ topicFuture.complete(Optional.empty());
} else {
log.warn("Failed to create topic {}", topic, exception);
pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
@@ -782,9 +782,9 @@ public void invalidateOfflineTopicStatCache(TopicName topicName) {
* This method will not make the broker attempt to load the topic if it's not already.
*/
public Optional<Topic> getTopicReference(String topic) {
- CompletableFuture<Topic> future = topics.get(topic);
+ CompletableFuture<Optional<Topic>> future = topics.get(topic);
if (future != null && future.isDone() && !future.isCompletedExceptionally()) {
- return Optional.ofNullable(future.join());
+ return future.join();
} else {
return Optional.empty();
}
@@ -814,27 +814,27 @@ public Semaphore getLookupRequestSemaphore() {
public void checkGC(int gcIntervalInSeconds) {
topics.forEach((n, t) -> {
- Topic topic = t.isCompletedExceptionally() ? null : t.getNow(null);
- if (topic != null) {
- topic.checkGC(gcIntervalInSeconds);
+ Optional<Topic> topic = extractTopic(t);
+ if (topic.isPresent()) {
+ topic.get().checkGC(gcIntervalInSeconds);
}
});
}
public void checkMessageExpiry() {
topics.forEach((n, t) -> {
- Topic topic = t.getNow(null);
- if (topic != null) {
- topic.checkMessageExpiry();
+ Optional<Topic> topic = extractTopic(t);
+ if (topic.isPresent()) {
+ topic.get().checkMessageExpiry();
}
});
}
public void checkMessageDeduplicationInfo() {
topics.forEach((n, t) -> {
- Topic topic = t.getNow(null);
- if (topic != null) {
- topic.checkMessageDeduplicationInfo();
+ Optional<Topic> topic = extractTopic(t);
+ if (topic.isPresent()) {
+ topic.get().checkMessageDeduplicationInfo();
}
});
}
@@ -868,8 +868,9 @@ public boolean isBacklogExceeded(PersistentTopic topic) {
public void monitorBacklogQuota() {
topics.forEach((n, t) -> {
try {
- if (t.getNow(null) != null && t.getNow(null) instanceof PersistentTopic) {
- PersistentTopic topic = (PersistentTopic) t.getNow(null);
+ Optional<Topic> optionalTopic = extractTopic(t);
+ if (optionalTopic.isPresent() && optionalTopic.get() instanceof PersistentTopic) {
+ PersistentTopic topic = (PersistentTopic) optionalTopic.get();
if (isBacklogExceeded(topic)) {
getBacklogQuotaManager().handleExceededBacklogQuota(topic);
} else if (topic == null) {
@@ -921,7 +922,8 @@ void checkTopicNsOwnership(final String topic) throws RuntimeException {
if (serviceUnit.includes(topicName)) {
// Topic needs to be unloaded
log.info("[{}] Unloading topic", topicName);
- closeFutures.add(topicFuture.thenCompose(Topic::close));
+ closeFutures.add(topicFuture
+ .thenCompose(t -> t.isPresent() ? t.get().close() : CompletableFuture.completedFuture(null)));
}
});
CompletableFuture<Void> aggregator = FutureUtil.waitForAll(closeFutures);
@@ -978,7 +980,7 @@ public int getNumberOfNamespaceBundles() {
return this.numberOfNamespaceBundles;
}
- public ConcurrentOpenHashMap<String, CompletableFuture<Topic>> getTopics() {
+ public ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> getTopics() {
return topics;
}
@@ -996,7 +998,8 @@ public void onUpdate(String path, Policies data, Stat stat) {
if (log.isDebugEnabled()) {
log.debug("Notifying topic that policies have changed: {}", name);
}
- topic.onPoliciesUpdate(data);
+
+ topic.ifPresent(t -> t.onPoliciesUpdate(data));
});
}
});
@@ -1033,9 +1036,9 @@ public String generateUniqueProducerName() {
public Map<String, PersistentTopicStats> getTopicStats() {
HashMap<String, PersistentTopicStats> stats = new HashMap<>();
topics.forEach((name, topicFuture) -> {
- Topic currentTopic = topicFuture.getNow(null);
- if (currentTopic != null) {
- stats.put(name, currentTopic.getStats());
+ Optional<Topic> topic = extractTopic(topicFuture);
+ if (topic.isPresent()) {
+ stats.put(name, topic.get().getStats());
}
});
return stats;
@@ -1128,18 +1131,12 @@ private void updateTopicMessageDispatchRate() {
this.pulsar().getExecutor().execute(() -> {
// update message-rate for each topic
topics.forEach((name, topicFuture) -> {
- if (topicFuture.isDone()) {
- String topicName = null;
- try {
- if (topicFuture.get() instanceof PersistentTopic) {
- PersistentTopic topic = (PersistentTopic) topicFuture.get();
- topicName = topicFuture.get().getName();
- // it first checks namespace-policy rate and if not present then applies broker-config
- topic.getDispatchRateLimiter().updateDispatchRate();
- }
- } catch (Exception e) {
- log.warn("[{}] failed to update message-dispatch rate {}", topicName, e.getMessage());
- }
+ Optional<Topic> topic = extractTopic(topicFuture);
+
+ if (topic.isPresent() && topic.get() instanceof PersistentTopic) {
+ PersistentTopic persistentTopic = (PersistentTopic) topic.get();
+ // it first checks namespace-policy rate and if not present then applies broker-config
+ persistentTopic.getDispatchRateLimiter().updateDispatchRate();
}
});
});
@@ -1149,22 +1146,19 @@ private void updateSubscriptionMessageDispatchRate() {
this.pulsar().getExecutor().submit(() -> {
// update message-rate for each topic subscription
topics.forEach((name, topicFuture) -> {
- if (topicFuture.isDone()) {
- try {
- topicFuture.get().getSubscriptions().forEach((subName, persistentSubscription) -> {
- if (persistentSubscription
- .getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
- ((PersistentDispatcherMultipleConsumers) persistentSubscription
- .getDispatcher()).getDispatchRateLimiter().updateDispatchRate();
- } else if (persistentSubscription
+ Optional<Topic> topic = extractTopic(topicFuture);
+
+ if (topic.isPresent()) {
+ topic.get().getSubscriptions().forEach((subName, persistentSubscription) -> {
+ if (persistentSubscription.getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
+ ((PersistentDispatcherMultipleConsumers) persistentSubscription.getDispatcher())
+ .getDispatchRateLimiter().updateDispatchRate();
+ } else if (persistentSubscription
.getDispatcher() instanceof PersistentDispatcherSingleActiveConsumer) {
- ((PersistentDispatcherSingleActiveConsumer) persistentSubscription
- .getDispatcher()).getDispatchRateLimiter().updateDispatchRate();
- }
- });
- } catch (Exception e) {
- log.warn("Failed to get topic from future while update subscription dispatch rate ", e);
- }
+ ((PersistentDispatcherSingleActiveConsumer) persistentSubscription.getDispatcher())
+ .getDispatchRateLimiter().updateDispatchRate();
+ }
+ });
}
});
});
@@ -1177,11 +1171,13 @@ private void updateManagedLedgerConfig() {
if (topicFuture.isDone()) {
String topicName = null;
try {
- if (topicFuture.getNow(null) instanceof PersistentTopic) {
- PersistentTopic topic = (PersistentTopic) topicFuture.get();
- topicName = topicFuture.get().getName();
+ Optional<Topic> topic = extractTopic(topicFuture);
+
+ if (topic.isPresent() && topic.get() instanceof PersistentTopic) {
+ PersistentTopic persistentTopic = (PersistentTopic) topic.get();
+ topicName = persistentTopic.getName();
// update skipNonRecoverableLedger configuration
- topic.getManagedLedger().getConfig().setAutoSkipNonRecoverableData(
+ persistentTopic.getManagedLedger().getConfig().setAutoSkipNonRecoverableData(
pulsar.getConfiguration().isAutoSkipNonRecoverableData());
}
} catch (Exception e) {
@@ -1340,7 +1336,7 @@ public static boolean validateDynamicConfiguration(String key, String value) {
* permit if it was successful to acquire it.
*/
private void createPendingLoadTopic() {
- Pair<String, CompletableFuture<Topic>> pendingTopic = pendingTopicLoadingQueue.poll();
+ Pair<String, CompletableFuture<Optional<Topic>>> pendingTopic = pendingTopicLoadingQueue.poll();
if (pendingTopic == null) {
return;
}
@@ -1348,7 +1344,7 @@ private void createPendingLoadTopic() {
final String topic = pendingTopic.getLeft();
try {
checkTopicNsOwnership(topic);
- CompletableFuture<Topic> pendingFuture = pendingTopic.getRight();
+ CompletableFuture<Optional<Topic>> pendingFuture = pendingTopic.getRight();
final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();
final boolean acquiredPermit = topicLoadSemaphore.tryAcquire();
createPersistentTopic(topic, true, pendingFuture);
@@ -1442,25 +1438,21 @@ private void blockDispatchersWithLargeUnAckMessages() {
lock.readLock().lock();
try {
topics.forEach((name, topicFuture) -> {
- if (topicFuture.isDone()) {
- try {
- topicFuture.get().getSubscriptions().forEach((subName, persistentSubscription) -> {
- if (persistentSubscription
- .getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
- PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) persistentSubscription
- .getDispatcher();
- int dispatcherUnAckMsgs = dispatcher.getTotalUnackedMessages();
- if (dispatcherUnAckMsgs > maxUnackedMsgsPerDispatcher) {
- log.info("[{}] Blocking dispatcher due to reached max broker limit {}",
- dispatcher.getName(), dispatcher.getTotalUnackedMessages());
- dispatcher.blockDispatcherOnUnackedMsgs();
- blockedDispatchers.add(dispatcher);
- }
+ Optional<Topic> topic = extractTopic(topicFuture);
+ if (topic.isPresent()) {
+ topic.get().getSubscriptions().forEach((subName, persistentSubscription) -> {
+ if (persistentSubscription.getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
+ PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) persistentSubscription
+ .getDispatcher();
+ int dispatcherUnAckMsgs = dispatcher.getTotalUnackedMessages();
+ if (dispatcherUnAckMsgs > maxUnackedMsgsPerDispatcher) {
+ log.info("[{}] Blocking dispatcher due to reached max broker limit {}",
+ dispatcher.getName(), dispatcher.getTotalUnackedMessages());
+ dispatcher.blockDispatcherOnUnackedMsgs();
+ blockedDispatchers.add(dispatcher);
}
- });
- } catch (Exception e) {
- log.warn("Failed to get topic from future ", e);
- }
+ }
+ });
}
});
} finally {
@@ -1496,4 +1488,15 @@ public ConfigField(Field field) {
this.field = field;
}
}
+
+ /**
+ * Safely extract optional topic instance from a future, in a way to avoid unchecked exceptions and race conditions.
+ */
+ public static Optional<Topic> extractTopic(CompletableFuture<Optional<Topic>> topicFuture) {
+ if (topicFuture.isDone() && !topicFuture.isCompletedExceptionally()) {
+ return topicFuture.join();
+ } else {
+ return Optional.empty();
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BookieClientStatsGenerator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BookieClientStatsGenerator.java
index c2ffc5341..914a25c7e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BookieClientStatsGenerator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BookieClientStatsGenerator.java
@@ -19,9 +19,12 @@
package org.apache.pulsar.broker.stats;
import java.util.Map;
+import java.util.Optional;
import org.apache.bookkeeper.mledger.proto.PendingBookieOpsStats;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.common.naming.TopicName;
@@ -47,8 +50,9 @@ public BookieClientStatsGenerator(PulsarService pulsar) {
private Map<String, Map<String, PendingBookieOpsStats>> generate() throws Exception {
if (pulsar.getBrokerService() != null && pulsar.getBrokerService().getTopics() != null) {
pulsar.getBrokerService().getTopics().forEach((name, topicFuture) -> {
- PersistentTopic persistentTopic = (PersistentTopic) topicFuture.getNow(null);
- if (persistentTopic != null) {
+ Optional<Topic> topic = BrokerService.extractTopic(topicFuture);
+ if (topic.isPresent() && topic.get() instanceof PersistentTopic) {
+ PersistentTopic persistentTopic = (PersistentTopic) topic.get();
TopicName topicName = TopicName.get(persistentTopic.getName());
put(topicName, persistentTopic.getManagedLedger().getStats().getPendingBookieOpsStats());
}
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 01e388a33..a2ef39726 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
@@ -32,11 +32,16 @@
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
+import com.google.common.collect.Lists;
+import com.google.common.hash.Hashing;
+
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.net.URI;
import java.util.HashSet;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -75,10 +80,6 @@
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
-import com.google.common.collect.Lists;
-import com.google.common.hash.Hashing;
-
public class NamespaceServiceTest extends BrokerTestBase {
@BeforeMethod
@@ -260,10 +261,11 @@ public void testUnloadNamespaceBundleFailure() throws Exception {
final String topicName = "persistent://my-property/use/my-ns/my-topic1";
pulsarClient.newConsumer().topic(topicName).subscriptionName("my-subscriber-name").subscribe();
- ConcurrentOpenHashMap<String, CompletableFuture<Topic>> topics = pulsar.getBrokerService().getTopics();
- Topic spyTopic = spy(topics.get(topicName).get());
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService()
+ .getTopics();
+ Topic spyTopic = spy(topics.get(topicName).get().get());
topics.clear();
- CompletableFuture<Topic> topicFuture = CompletableFuture.completedFuture(spyTopic);
+ CompletableFuture<Optional<Topic>> topicFuture = CompletableFuture.completedFuture(Optional.of(spyTopic));
// add mock topic
topics.put(topicName, topicFuture);
doAnswer(new Answer<CompletableFuture<Void>>() {
@@ -300,10 +302,10 @@ public void testUnloadNamespaceBundleWithStuckTopic() throws Exception {
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<Topic>> topics = pulsar.getBrokerService().getTopics();
- Topic spyTopic = spy(topics.get(topicName).get());
+ ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService().getTopics();
+ Topic spyTopic = spy(topics.get(topicName).get().get());
topics.clear();
- CompletableFuture<Topic> topicFuture = CompletableFuture.completedFuture(spyTopic);
+ CompletableFuture<Optional<Topic>> topicFuture = CompletableFuture.completedFuture(Optional.of(spyTopic));
// add mock topic
topics.put(topicName, topicFuture);
// return uncompleted future as close-topic result.
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 0918bc7c9..31773bbd6 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
@@ -746,7 +746,7 @@ public void testTopicLoadingOnDisableNamespaceBundle() throws Exception {
pulsar.getNamespaceService().getOwnershipCache().updateBundleState(bundle, false);
// try to create topic which should fail as bundle is disable
- CompletableFuture<Topic> futureResult = pulsar.getBrokerService().loadOrCreatePersistentTopic(topicName, true);
+ CompletableFuture<Optional<Topic>> futureResult = pulsar.getBrokerService().loadOrCreatePersistentTopic(topicName, true);
try {
futureResult.get();
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services