From a9dbf26b43912da1110cb78a8d6608f7600736e2 Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Thu, 9 Jun 2022 13:14:09 +0200 Subject: [PATCH 1/7] PIP-145: Notifications for faster topic discovery This commit introduces topic list watchers. By using these objects clients can observe the creation or deletion of topics closer to real-time. This reduces latency in consuming the first messages published to a topic when using a pattern-based subscription. Modifications: - New commands were added to the binary protocol to enable registering and deregistering watchers. - Pattern-based consumers create TopicListWatcher objects if the broker supports this feature. Otherwise, they fall back to polling only. - The watchers use ConnectionHandler to obtain a connection to a broker. - Once connected, watchers register and wait for updates. - ServerCnx uses the newly created TopicListService to manage watchers. - TopicListService listens to metadata notifications and sends updates. --- .../broker/resources/TopicResources.java | 43 +++ .../broker/resources/TopicResourcesTest.java | 109 +++++++ .../broker/service/PulsarCommandSender.java | 7 +- .../service/PulsarCommandSenderImpl.java | 24 +- .../pulsar/broker/service/ServerCnx.java | 63 +++- .../broker/service/TopicListService.java | 254 ++++++++++++++++ .../pulsar/broker/service/ServerCnxTest.java | 23 ++ .../broker/service/TopicListServiceTest.java | 128 ++++++++ .../broker/service/TopicListWatcherTest.java | 102 +++++++ .../service/utils/ClientChannelHelper.java | 6 + .../pulsar/client/api/ClientErrorsTest.java | 4 +- .../pulsar/client/api/MockBrokerService.java | 2 +- .../client/impl/BinaryProtoLookupService.java | 5 + .../apache/pulsar/client/impl/ClientCnx.java | 82 +++++ .../pulsar/client/impl/ConnectionHandler.java | 10 +- .../apache/pulsar/client/impl/HttpClient.java | 4 + .../pulsar/client/impl/HttpLookupService.java | 5 + .../pulsar/client/impl/LookupService.java | 7 + .../impl/PatternMultiTopicsConsumerImpl.java | 31 +- .../pulsar/client/impl/PulsarClientImpl.java | 21 ++ .../pulsar/client/impl/TopicListWatcher.java | 287 ++++++++++++++++++ .../pulsar/client/impl/ClientCnxTest.java | 108 +++++++ .../client/impl/PulsarClientImplTest.java | 3 +- .../client/impl/TopicListWatcherTest.java | 111 +++++++ .../pulsar/common/protocol/Commands.java | 62 +++- .../pulsar/common/protocol/PulsarDecoder.java | 42 +++ .../pulsar/common/topics/TopicList.java | 2 + pulsar-common/src/main/proto/PulsarApi.proto | 40 +++ .../pulsar/proxy/server/ProxyConnection.java | 5 +- 29 files changed, 1566 insertions(+), 24 deletions(-) create mode 100644 pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java index bea67943f8268..ee537431280f3 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java @@ -19,22 +19,35 @@ package org.apache.pulsar.broker.resources; import static org.apache.pulsar.common.util.Codec.decode; +import java.util.EnumSet; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.BiConsumer; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import java.util.stream.Collectors; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.metadata.api.MetadataStore; +import org.apache.pulsar.metadata.api.Notification; +import org.apache.pulsar.metadata.api.NotificationType; public class TopicResources { private static final String MANAGED_LEDGER_PATH = "/managed-ledgers"; private final MetadataStore store; + private final Map, Pattern> topicListeners; + public TopicResources(MetadataStore store) { this.store = store; + topicListeners = new ConcurrentHashMap<>(); + store.registerListener(this::handleNotification); } public CompletableFuture> listPersistentTopicsAsync(NamespaceName ns) { @@ -110,4 +123,34 @@ public CompletableFuture clearTenantPersistence(String tenant) { } }); } + + void handleNotification(Notification notification) { + if (notification.getPath().startsWith(MANAGED_LEDGER_PATH) + && EnumSet.of(NotificationType.Created, NotificationType.Deleted).contains(notification.getType())) { + for (Map.Entry, Pattern> entry : + new HashMap<>(topicListeners).entrySet()) { + Matcher matcher = entry.getValue().matcher(notification.getPath()); + if (matcher.matches()) { + TopicName topicName = TopicName.get( + matcher.group(2), NamespaceName.get(matcher.group(1)), matcher.group(3)); + entry.getKey().accept(topicName.toString(), notification.getType()); + } + } + } + } + + Pattern namespaceNameToTopicNamePattern(NamespaceName namespaceName) { + return Pattern.compile( + MANAGED_LEDGER_PATH + "/(" + namespaceName + ")/(" + TopicDomain.persistent + ")/(" + "[^/]+)"); + } + + public void registerPersistentTopicListener( + NamespaceName namespaceName, BiConsumer listener) { + topicListeners.put(listener, namespaceNameToTopicNamePattern(namespaceName)); + } + + public void deregisterPersistentTopicListener(BiConsumer listener) { + topicListeners.remove(listener); + } + } diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java new file mode 100644 index 0000000000000..cfafc5107d2de --- /dev/null +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/TopicResourcesTest.java @@ -0,0 +1,109 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.resources; + +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.metadata.api.MetadataStore; +import org.apache.pulsar.metadata.api.Notification; +import org.apache.pulsar.metadata.api.NotificationType; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import java.util.function.BiConsumer; + +public class TopicResourcesTest { + + private MetadataStore metadataStore; + private TopicResources topicResources; + + @BeforeMethod + public void setup() { + metadataStore = mock(MetadataStore.class); + topicResources = new TopicResources(metadataStore); + } + + @Test + public void testConstructorRegistersAsListener() { + verify(metadataStore).registerListener(any()); + } + + @Test + public void testListenerInvokedWhenTopicCreated() { + BiConsumer listener = mock(BiConsumer.class); + topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener); + topicResources.handleNotification(new Notification(NotificationType.Created, "/managed-ledgers/tenant/namespace/persistent/topic")); + verify(listener).accept("persistent://tenant/namespace/topic", NotificationType.Created); + } + + @Test + public void testListenerInvokedWhenTopicV1Created() { + BiConsumer listener = mock(BiConsumer.class); + topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/cluster/namespace"), listener); + topicResources.handleNotification(new Notification(NotificationType.Created, "/managed-ledgers/tenant/cluster/namespace/persistent/topic")); + verify(listener).accept("persistent://tenant/cluster/namespace/topic", NotificationType.Created); + } + + @Test + public void testListenerInvokedWhenTopicDeleted() { + BiConsumer listener = mock(BiConsumer.class); + topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener); + topicResources.handleNotification(new Notification(NotificationType.Deleted, "/managed-ledgers/tenant/namespace/persistent/topic")); + verify(listener).accept("persistent://tenant/namespace/topic", NotificationType.Deleted); + } + + @Test + public void testListenerNotInvokedWhenSubscriptionCreated() { + BiConsumer listener = mock(BiConsumer.class); + topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener); + topicResources.handleNotification(new Notification(NotificationType.Created, "/managed-ledgers/tenant/namespace/persistent/topic/subscription")); + verifyNoInteractions(listener); + } + + @Test + public void testListenerNotInvokedWhenTopicCreatedInOtherNamespace() { + BiConsumer listener = mock(BiConsumer.class); + topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener); + topicResources.handleNotification(new Notification(NotificationType.Created, "/managed-ledgers/tenant/namespace2/persistent/topic")); + verifyNoInteractions(listener); + } + + @Test + public void testListenerNotInvokedWhenTopicModified() { + BiConsumer listener = mock(BiConsumer.class); + topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener); + topicResources.handleNotification(new Notification(NotificationType.Modified, "/managed-ledgers/tenant/namespace/persistent/topic")); + verifyNoInteractions(listener); + } + + @Test + public void testListenerNotInvokedAfterDeregistered() { + BiConsumer listener = mock(BiConsumer.class); + topicResources.registerPersistentTopicListener(NamespaceName.get("tenant/namespace"), listener); + topicResources.handleNotification(new Notification(NotificationType.Created, "/managed-ledgers/tenant/namespace/persistent/topic")); + verify(listener).accept("persistent://tenant/namespace/topic", NotificationType.Created); + topicResources.deregisterPersistentTopicListener(listener); + topicResources.handleNotification(new Notification(NotificationType.Created, "/managed-ledgers/tenant/namespace/persistent/topic2")); + verifyNoMoreInteractions(listener); + } + +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java index b267f725986e9..8934d2e6ce684 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSender.java @@ -61,7 +61,7 @@ void sendGetTopicsOfNamespaceResponse(List topics, String topicsHash, bo void sendGetOrCreateSchemaErrorResponse(long requestId, ServerError error, String errorMessage); - void sendConnectedResponse(int clientProtocolVersion, int maxMessageSize); + void sendConnectedResponse(int clientProtocolVersion, int maxMessageSize, boolean supportsTopicWatchers); void sendLookupResponse(String brokerServiceUrl, String brokerServiceUrlTls, boolean authoritative, CommandLookupTopicResponse.LookupType response, long requestId, @@ -92,4 +92,9 @@ Future sendMessagesToConsumer(long consumerId, String topicName, Subscript void sendEndTxnResponse(long requestId, TxnID txnID, int txnAction); void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError error, String message); + + void sendWatchTopicListSuccess(long requestId, long watcherId, String topicsHash, List topics); + + void sendWatchTopicListUpdate(long watcherId, + List newTopics, List deletedTopics, String topicsHash); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java index 4cdc0a76e8c96..71a6d4935cddf 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java @@ -162,8 +162,9 @@ public void sendGetOrCreateSchemaErrorResponse(long requestId, ServerError error } @Override - public void sendConnectedResponse(int clientProtocolVersion, int maxMessageSize) { - BaseCommand command = Commands.newConnectedCommand(clientProtocolVersion, maxMessageSize); + public void sendConnectedResponse(int clientProtocolVersion, int maxMessageSize, boolean supportsTopicWatchers) { + BaseCommand command = Commands.newConnectedCommand( + clientProtocolVersion, maxMessageSize, supportsTopicWatchers); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); cnx.ctx().writeAndFlush(outBuf); @@ -346,6 +347,25 @@ public void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError err } } + @Override + public void sendWatchTopicListSuccess(long requestId, long watcherId, String topicsHash, List topics) { + BaseCommand command = Commands.newWatchTopicListSuccess(requestId, watcherId, topicsHash, topics); + interceptAndWriteCommand(command); + } + + @Override + public void sendWatchTopicListUpdate(long watcherId, + List newTopics, List deletedTopics, String topicsHash) { + BaseCommand command = Commands.newWatchTopicUpdate(watcherId, newTopics, deletedTopics, topicsHash); + interceptAndWriteCommand(command); + } + + private void interceptAndWriteCommand(BaseCommand command) { + safeIntercept(command, cnx); + ByteBuf outBuf = Commands.serializeWithSize(command); + cnx.ctx().writeAndFlush(outBuf); + } + private void safeIntercept(BaseCommand command, ServerCnx cnx) { try { this.interceptor.onPulsarCommand(command, cnx); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 74b37213fbd55..42eb46c69408d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -119,6 +119,8 @@ import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.CommandTcClientConnectRequest; import org.apache.pulsar.common.api.proto.CommandUnsubscribe; +import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicList; import org.apache.pulsar.common.api.proto.FeatureFlags; import org.apache.pulsar.common.api.proto.KeySharedMeta; import org.apache.pulsar.common.api.proto.KeySharedMode; @@ -164,6 +166,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx { private final ConcurrentLongHashMap> consumers; private final boolean enableSubscriptionPatternEvaluation; private final int maxSubscriptionPatternLength; + private final TopicListService topicListService; private State state; private volatile boolean isActive = true; String authRole = null; @@ -272,6 +275,8 @@ public ServerCnx(PulsarService pulsar, String listenerName) { this.connectionController = new ConnectionController.DefaultConnectionController(conf); this.enableSubscriptionPatternEvaluation = conf.isEnableBrokerSideSubscriptionPatternEvaluation(); this.maxSubscriptionPatternLength = conf.getSubscriptionPatternMaxLength(); + this.topicListService = new TopicListService(pulsar, this, + enableSubscriptionPatternEvaluation, maxSubscriptionPatternLength); } @Override @@ -334,6 +339,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { } } }); + this.topicListService.inactivate(); this.service.getPulsarStats().recordConnectionClose(); } @@ -623,8 +629,8 @@ ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) { } // complete the connect and sent newConnected command - private void completeConnect(int clientProtoVersion, String clientVersion) { - ctx.writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize)); + private void completeConnect(int clientProtoVersion, String clientVersion, boolean supportsTopicWatchers) { + ctx.writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers)); state = State.Connected; service.getPulsarStats().recordConnectionCreateSuccess(); if (log.isDebugEnabled()) { @@ -683,7 +689,7 @@ private State doAuthentication(AuthData clientData, if (state != State.Connected) { // First time authentication is done - completeConnect(clientProtocolVersion, clientVersion); + completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation); } else { // If the connection was already ready, it means we're doing a refresh if (!StringUtils.isEmpty(authRole)) { @@ -790,7 +796,7 @@ protected void handleConnect(CommandConnect connect) { } if (!service.isAuthenticationEnabled()) { - completeConnect(clientProtocolVersion, clientVersion); + completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation); return; } @@ -818,7 +824,7 @@ protected void handleConnect(CommandConnect connect) { authRole = getBrokerService().getAuthenticationService().getAnonymousUserRole() .orElseThrow(() -> new AuthenticationException("No anonymous role, and no authentication provider configured")); - completeConnect(clientProtocolVersion, clientVersion); + completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation); return; } @@ -2513,6 +2519,53 @@ protected void handleAddSubscriptionToTxn(CommandAddSubscriptionToTxn command) { })); } + protected void handleCommandWatchTopicList(CommandWatchTopicList commandWatchTopicList) { + final long requestId = commandWatchTopicList.getRequestId(); + final long watcherId = commandWatchTopicList.getWatcherId(); + final NamespaceName namespaceName = NamespaceName.get(commandWatchTopicList.getNamespace()); + + final Semaphore lookupSemaphore = service.getLookupRequestSemaphore(); + if (lookupSemaphore.tryAcquire()) { + if (invalidOriginalPrincipal(originalPrincipal)) { + final String msg = "Valid Proxy Client role should be provided for watchTopicListRequest "; + log.warn("[{}] {} with role {} and proxyClientAuthRole {} on namespace {}", remoteAddress, msg, + authRole, originalPrincipal, namespaceName); + commandSender.sendErrorResponse(watcherId, ServerError.AuthorizationError, msg); + lookupSemaphore.release(); + return; + } + isNamespaceOperationAllowed(namespaceName, NamespaceOperation.GET_TOPICS).thenApply(isAuthorized -> { + if (isAuthorized) { + topicListService.handleWatchTopicList(commandWatchTopicList, lookupSemaphore); + } else { + final String msg = "Proxy Client is not authorized to watchTopicList"; + log.warn("[{}] {} with role {} on namespace {}", remoteAddress, msg, getPrincipal(), namespaceName); + commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, msg); + lookupSemaphore.release(); + } + return null; + }).exceptionally(ex -> { + logNamespaceNameAuthException(remoteAddress, "watchTopicList", getPrincipal(), + Optional.of(namespaceName), ex); + final String msg = "Exception occurred while trying to handle command WatchTopicList"; + commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, msg); + lookupSemaphore.release(); + return null; + }); + } else { + if (log.isDebugEnabled()) { + log.debug("[{}] Failed WatchTopicList due to too many lookup-requests {}", remoteAddress, + namespaceName); + } + commandSender.sendErrorResponse(requestId, ServerError.TooManyRequests, + "Failed due to too many pending lookup requests"); + } + } + + protected void handleCommandUnwatchTopicList(CommandUnwatchTopicList commandUnwatchTopicList) { + topicListService.handleUnwatchTopicList(commandUnwatchTopicList); + } + @Override protected boolean isHandshakeCompleted() { return state == State.Connected; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java new file mode 100644 index 0000000000000..b3e232955cca8 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java @@ -0,0 +1,254 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Semaphore; +import java.util.function.BiConsumer; +import java.util.regex.Pattern; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.namespace.NamespaceService; +import org.apache.pulsar.broker.resources.TopicResources; +import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicList; +import org.apache.pulsar.common.api.proto.ServerError; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.topics.TopicList; +import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap; +import org.apache.pulsar.metadata.api.NotificationType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class TopicListService { + + + public static class TopicListWatcher implements BiConsumer { + + private final List matchingTopics; + private final TopicListService topicListService; + private final long id; + private final Pattern topicsPattern; + + public TopicListWatcher(TopicListService topicListService, long id, + Pattern topicsPattern, List topics) { + this.topicListService = topicListService; + this.id = id; + this.topicsPattern = topicsPattern; + this.matchingTopics = TopicList.filterTopics(topics, topicsPattern); + } + + public List getMatchingTopics() { + return matchingTopics; + } + + @Override + public void accept(String topicName, NotificationType notificationType) { + if (topicsPattern.matcher(topicName).matches()) { + List newTopics; + List deletedTopics; + if (notificationType == NotificationType.Deleted) { + newTopics = Collections.emptyList(); + deletedTopics = Collections.singletonList(topicName); + matchingTopics.remove(topicName); + } else { + deletedTopics = Collections.emptyList(); + newTopics = Collections.singletonList(topicName); + matchingTopics.add(topicName); + } + String hash = TopicList.calculateHash(matchingTopics); + topicListService.sendTopicListUpdate(id, hash, deletedTopics, newTopics); + } + } + } + + + private static final Logger log = LoggerFactory.getLogger(TopicListService.class); + + private final NamespaceService namespaceService; + private final TopicResources topicResources; + private final ServerCnx connection; + private final boolean enableSubscriptionPatternEvaluation; + private final int maxSubscriptionPatternLength; + private final ConcurrentLongHashMap> watchers; + + + public TopicListService(PulsarService pulsar, ServerCnx connection, + boolean enableSubscriptionPatternEvaluation, int maxSubscriptionPatternLength) { + this.namespaceService = pulsar.getNamespaceService(); + this.connection = connection; + this.enableSubscriptionPatternEvaluation = enableSubscriptionPatternEvaluation; + this.maxSubscriptionPatternLength = maxSubscriptionPatternLength; + this.watchers = ConcurrentLongHashMap.>newBuilder() + .expectedItems(8) + .concurrencyLevel(1) + .build(); + this.topicResources = pulsar.getPulsarResources().getTopicResources(); + } + + public void inactivate() { + for (Long watcherId : new HashSet<>(watchers.keys())) { + deleteTopicListWatcher(watcherId); + } + } + + public void handleWatchTopicList(CommandWatchTopicList commandWatchTopicList, Semaphore lookupSemaphore) { + String namespace = commandWatchTopicList.getNamespace(); + NamespaceName namespaceName = NamespaceName.get(namespace); + long watcherId = commandWatchTopicList.getWatcherId(); + long requestId = commandWatchTopicList.getRequestId(); + + Pattern topicsPattern = Pattern.compile(commandWatchTopicList.hasTopicsPattern() + ? commandWatchTopicList.getTopicsPattern() : TopicList.ALL_TOPICS_PATTERN); + if (!enableSubscriptionPatternEvaluation || topicsPattern.pattern().length() > maxSubscriptionPatternLength) { + String msg = "Unable to create topic list watcher: "; + if (!enableSubscriptionPatternEvaluation) { + msg += "Evaluating subscription patterns is disabled."; + } else { + msg += "Pattern longer than maximum: " + maxSubscriptionPatternLength; + } + log.warn("[{}] {} on namespace {}", connection.getRemoteAddress(), msg, namespaceName); + connection.getCommandSender().sendErrorResponse(requestId, ServerError.NotAllowedError, msg); + lookupSemaphore.release(); + return; + } + CompletableFuture watcherFuture = new CompletableFuture<>(); + CompletableFuture existingWatcherFuture = watchers.putIfAbsent(watcherId, watcherFuture); + + if (existingWatcherFuture != null) { + if (existingWatcherFuture.isDone() && !existingWatcherFuture.isCompletedExceptionally()) { + TopicListWatcher watcher = existingWatcherFuture.getNow(null); + log.info("[{}] Watcher with the same id is already created:" + + " watcherId={}, watcher={}", + connection.getRemoteAddress(), watcherId, watcher); + watcherFuture = existingWatcherFuture; + } else { + // There was an early request to create a watcher with the same watcherId. This can happen when + // client timeout is lower the broker timeouts. We need to wait until the previous watcher + // creation request either completes or fails. + log.warn("[{}] Watcher with id is already present on the connection," + + " consumerId={}", connection.getRemoteAddress(), watcherId); + ServerError error; + if (!existingWatcherFuture.isDone()) { + error = ServerError.ServiceNotReady; + } else { + error = ServerError.UnknownError; + watchers.remove(watcherId, existingWatcherFuture); + } + connection.getCommandSender().sendErrorResponse(requestId, error, + "Topic list watcher is already present on the connection"); + lookupSemaphore.release(); + return; + } + } else { + initializeTopicsListWatcher(watcherFuture, namespaceName, watcherId, topicsPattern); + } + + + watcherFuture.thenAccept(watcher -> { + List topicList = watcher.getMatchingTopics(); + String hash = TopicList.calculateHash(topicList); + if (commandWatchTopicList.hasTopicsHash() + && hash.equals(commandWatchTopicList.getTopicsHash())) { + topicList = Collections.emptyList(); + } + if (log.isDebugEnabled()) { + log.debug( + "[{}] Received WatchTopicList for namespace [//{}] by {}", + connection.getRemoteAddress(), namespace, requestId); + } + connection.getCommandSender().sendWatchTopicListSuccess(requestId, watcherId, hash, topicList); + lookupSemaphore.release(); + }) + .exceptionally(ex -> { + log.warn("[{}] Error WatchTopicList for namespace [//{}] by {}", + connection.getRemoteAddress(), namespace, requestId); + connection.getCommandSender().sendErrorResponse(requestId, + BrokerServiceException.getClientErrorCode( + new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage()); + lookupSemaphore.release(); + return null; + }); + } + + + public void initializeTopicsListWatcher(CompletableFuture watcherFuture, + NamespaceName namespace, long watcherId, Pattern topicsPattern) { + namespaceService.getListOfPersistentTopics(namespace). + thenApply(topics -> { + TopicListWatcher watcher = new TopicListWatcher(this, watcherId, topicsPattern, topics); + topicResources.registerPersistentTopicListener(namespace, watcher); + return watcher; + }). + whenComplete((watcher, exception) -> { + if (exception != null) { + watcherFuture.completeExceptionally(exception); + } else { + watcherFuture.complete(watcher); + } + }); + } + + + public void handleUnwatchTopicList(CommandUnwatchTopicList commandUnwatchTopicList) { + long requestId = commandUnwatchTopicList.getRequestId(); + long watcherId = commandUnwatchTopicList.getWatcherId(); + deleteTopicListWatcher(watcherId); + connection.getCommandSender().sendWatchTopicListSuccess(requestId, watcherId, null, null); + } + + public void deleteTopicListWatcher(Long watcherId) { + CompletableFuture watcherFuture = watchers.get(watcherId); + if (watcherFuture == null) { + log.info("[{}] TopicListWatcher was not registered on the connection: {}", + watcherId, connection.getRemoteAddress()); + return; + } + + if (!watcherFuture.isDone() && watcherFuture + .completeExceptionally(new IllegalStateException("Closed watcher before creation was complete"))) { + // We have received a request to close the watcher before it was actually completed, we have marked the + // watcher future as failed and we can tell the client the close operation was successful. When the actual + // create operation will complete, the new watcher will be discarded. + log.info("[{}] Closed watcher before its creation was completed. watcherId={}", + connection.getRemoteAddress(), watcherId); + return; + } + + if (watcherFuture.isCompletedExceptionally()) { + log.info("[{}] Closed watcher that already failed to be created. watcherId={}", + connection.getRemoteAddress(), watcherId); + return; + } + + // Proceed with normal watcher close + topicResources.deregisterPersistentTopicListener(watcherFuture.getNow(null)); + watchers.remove(watcherId); + log.info("[{}] Closed watcher, watcherId={}", connection.getRemoteAddress(), watcherId); + } + + public void sendTopicListUpdate(long watcherId, String topicsHash, List deletedTopics, + List newTopics) { + connection.getCommandSender().sendWatchTopicListUpdate(watcherId, newTopics, deletedTopics, topicsHash); + } + + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java index cffc82bcf73c6..3b96c80bd3c45 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java @@ -102,6 +102,7 @@ import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.CommandSuccess; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.api.proto.ProtocolVersion; import org.apache.pulsar.common.api.proto.ServerError; @@ -216,6 +217,8 @@ public void setup() throws Exception { doReturn(CompletableFuture.completedFuture(true)).when(namespaceService).checkTopicOwnership(any()); doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfTopics( NamespaceName.get("use", "ns-abc"), CommandGetTopicsOfNamespace.Mode.ALL); + doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfPersistentTopics( + NamespaceName.get("use", "ns-abc")); setupMLAsyncCallbackMocks(); @@ -1960,6 +1963,26 @@ public void testGetTopicsOfNamespaceNoChange() throws Exception { channel.finish(); } + @Test + public void testWatchTopicList() throws Exception { + svcConfig.setEnableBrokerSideSubscriptionPatternEvaluation(true); + resetChannel(); + setChannelConnected(); + BaseCommand command = Commands.newWatchTopicList(1, 3, "use/ns-abc", "use/ns-abc/topic-.*", null); + ByteBuf serializedCommand = Commands.serializeWithSize(command); + + channel.writeInbound(serializedCommand); + Object resp = getResponse(); + System.out.println(resp); + CommandWatchTopicListSuccess response = (CommandWatchTopicListSuccess) resp; + + assertEquals(response.getTopicsList(), matchingTopics); + assertEquals(response.getTopicsHash(), TopicList.calculateHash(matchingTopics)); + assertEquals(response.getWatcherId(), 3); + + channel.finish(); + } + @Test public void testNeverDelayConsumerFutureWhenNotFail() throws Exception{ // Mock ServerCnx.field: consumers diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java new file mode 100644 index 0000000000000..494bb734bd482 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java @@ -0,0 +1,128 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import org.apache.pulsar.broker.PulsarServerException; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.namespace.NamespaceService; +import org.apache.pulsar.broker.resources.PulsarResources; +import org.apache.pulsar.broker.resources.TopicResources; +import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicList; +import org.apache.pulsar.common.api.proto.ServerError; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.topics.TopicList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + + +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import java.net.InetSocketAddress; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Semaphore; + +public class TopicListServiceTest { + + private TopicListService topicListService; + private ServerCnx connection; + private CompletableFuture> topicListFuture; + private Semaphore lookupSemaphore; + private TopicResources topicResources; + + @BeforeMethod(alwaysRun = true) + public void setup() throws Exception { + lookupSemaphore = new Semaphore(1); + lookupSemaphore.acquire(); + topicListFuture = new CompletableFuture<>(); + topicResources = mock(TopicResources.class); + + PulsarService pulsar = mock(PulsarService.class); + when(pulsar.getNamespaceService()).thenReturn(mock(NamespaceService.class)); + when(pulsar.getPulsarResources()).thenReturn(mock(PulsarResources.class)); + when(pulsar.getPulsarResources().getTopicResources()).thenReturn(topicResources); + when(pulsar.getNamespaceService().getListOfPersistentTopics(any())).thenReturn(topicListFuture); + + + connection = mock(ServerCnx.class); + when(connection.getRemoteAddress()).thenReturn(new InetSocketAddress(10000)); + when(connection.getCommandSender()).thenReturn(mock(PulsarCommandSender.class)); + + topicListService = new TopicListService(pulsar, connection, true, 30); + + } + + @Test + public void testCommandWatchSuccessResponse() { + CommandWatchTopicList command = new CommandWatchTopicList() + .setRequestId(7) + .setWatcherId(13) + .setNamespace("tenant/ns") + .setTopicsPattern("persistent://tenant/ns/topic\\d"); + topicListService.handleWatchTopicList(command, lookupSemaphore); + List topics = Collections.singletonList("persistent://tenant/ns/topic1"); + String hash = TopicList.calculateHash(topics); + topicListFuture.complete(topics); + Assert.assertEquals(1, lookupSemaphore.availablePermits()); + verify(topicResources).registerPersistentTopicListener( + eq(NamespaceName.get("tenant/ns")), any(TopicListService.TopicListWatcher.class)); + verify(connection.getCommandSender()).sendWatchTopicListSuccess(7, 13, hash, topics); + } + + @Test + public void testCommandWatchErrorResponse() { + CommandWatchTopicList command = new CommandWatchTopicList() + .setRequestId(7) + .setWatcherId(13) + .setNamespace("tenant/ns") + .setTopicsPattern("persistent://tenant/ns/topic\\d"); + topicListService.handleWatchTopicList(command, lookupSemaphore); + topicListFuture.completeExceptionally(new PulsarServerException("Error")); + Assert.assertEquals(1, lookupSemaphore.availablePermits()); + verifyNoInteractions(topicResources); + verify(connection.getCommandSender()).sendErrorResponse(eq(7L), any(ServerError.class), + eq(PulsarServerException.class.getCanonicalName() + ": Error")); + } + + @Test + public void testCommandUnwatchRemovesListener() { + CommandWatchTopicList commandWatch = new CommandWatchTopicList() + .setRequestId(7) + .setWatcherId(13) + .setNamespace("tenant/ns") + .setTopicsPattern("persistent://tenant/ns/topic\\d"); + topicListService.handleWatchTopicList(commandWatch, lookupSemaphore); + List topics = Collections.singletonList("persistent://tenant/ns/topic1"); + topicListFuture.complete(topics); + + CommandUnwatchTopicList commandUnwatch = new CommandUnwatchTopicList() + .setRequestId(8) + .setWatcherId(13); + topicListService.handleUnwatchTopicList(commandUnwatch); + verify(topicResources).deregisterPersistentTopicListener(any(TopicListService.TopicListWatcher.class)); + } + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java new file mode 100644 index 0000000000000..1a2e192d1f5fa --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListWatcherTest.java @@ -0,0 +1,102 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import org.apache.pulsar.common.topics.TopicList; +import org.apache.pulsar.metadata.api.NotificationType; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.regex.Pattern; + +public class TopicListWatcherTest { + + private static final List INITIAL_TOPIC_LIST = Arrays.asList( + "persistent://tenant/ns/topic1", + "persistent://tenant/ns/topic2", + "persistent://tenant/ns/t3" + ); + + private static final long ID = 7; + private static final Pattern PATTERN = Pattern.compile("persistent://tenant/ns/topic\\d+"); + + + private TopicListService topicListService; + private TopicListService.TopicListWatcher watcher; + + + + @BeforeMethod(alwaysRun = true) + public void setup() { + topicListService = mock(TopicListService.class); + watcher = new TopicListService.TopicListWatcher(topicListService, ID, PATTERN, INITIAL_TOPIC_LIST); + } + + @Test + public void testGetMatchingTopicsReturnsFilteredList() { + Assert.assertEquals( + Arrays.asList("persistent://tenant/ns/topic1", "persistent://tenant/ns/topic2"), + watcher.getMatchingTopics()); + } + + @Test + public void testAcceptSendsNotificationAndRemembersTopic() { + String newTopic = "persistent://tenant/ns/topic3"; + watcher.accept(newTopic, NotificationType.Created); + + List allMatchingTopics = Arrays.asList( + "persistent://tenant/ns/topic1", "persistent://tenant/ns/topic2", newTopic); + String hash = TopicList.calculateHash(allMatchingTopics); + verify(topicListService).sendTopicListUpdate(ID, hash, Collections.emptyList(), + Collections.singletonList(newTopic)); + Assert.assertEquals( + allMatchingTopics, + watcher.getMatchingTopics()); + } + + @Test + public void testAcceptSendsNotificationAndForgetsTopic() { + String deletedTopic = "persistent://tenant/ns/topic1"; + watcher.accept(deletedTopic, NotificationType.Deleted); + + List allMatchingTopics = Collections.singletonList("persistent://tenant/ns/topic2"); + String hash = TopicList.calculateHash(allMatchingTopics); + verify(topicListService).sendTopicListUpdate(ID, hash, + Collections.singletonList(deletedTopic), Collections.emptyList()); + Assert.assertEquals( + allMatchingTopics, + watcher.getMatchingTopics()); + } + + @Test + public void testAcceptIgnoresNonMatching() { + watcher.accept("persistent://tenant/ns/mytopic", NotificationType.Created); + verifyNoInteractions(topicListService); + Assert.assertEquals( + Arrays.asList("persistent://tenant/ns/topic1", "persistent://tenant/ns/topic2"), + watcher.getMatchingTopics()); + } + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java index 92f2617cc680b..a178227896661 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java @@ -21,6 +21,7 @@ import java.util.Queue; import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespaceResponse; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; import org.apache.pulsar.common.protocol.PulsarDecoder; import org.apache.pulsar.common.api.proto.CommandAck; import org.apache.pulsar.common.api.proto.CommandCloseConsumer; @@ -156,6 +157,11 @@ protected void handleLookupResponse(CommandLookupTopicResponse connection) { protected void handleGetTopicsOfNamespaceSuccess(CommandGetTopicsOfNamespaceResponse response) { queue.offer(new CommandGetTopicsOfNamespaceResponse().copyFrom(response)); } + + @Override + protected void handleCommandWatchTopicListSuccess(CommandWatchTopicListSuccess commandWatchTopicListSuccess) { + queue.offer(new CommandWatchTopicListSuccess().copyFrom(commandWatchTopicListSuccess)); + } }; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java index 4353297666a9e..8c8e4b96addb4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java @@ -719,7 +719,7 @@ public void testProducerReconnect() throws Exception { AtomicBoolean msgSent = new AtomicBoolean(); mockBrokerService.setHandleConnect((ctx, connect) -> { channelCtx.set(ctx); - ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion())); + ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion(), false)); if (numOfConnections.incrementAndGet() == 2) { // close the cnx immediately when trying to connect the 2nd time ctx.channel().close(); @@ -759,7 +759,7 @@ public void testConsumerReconnect() throws Exception { CountDownLatch latch = new CountDownLatch(1); mockBrokerService.setHandleConnect((ctx, connect) -> { channelCtx.set(ctx); - ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion())); + ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion(), false)); if (numOfConnections.incrementAndGet() == 2) { // close the cnx immediately when trying to connect the 2nd time ctx.channel().close(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java index b16fac427aa7c..7cba7405ddbf6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java @@ -143,7 +143,7 @@ protected void handleConnect(CommandConnect connect) { return; } // default - ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion())); + ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion(), false)); } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java index 2106b7a5c6f35..18bffba8dc627 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java @@ -256,6 +256,11 @@ public String getServiceUrl() { return serviceNameResolver.getServiceUrl(); } + @Override + public InetSocketAddress resolveHost() { + return serviceNameResolver.resolveHost(); + } + @Override public CompletableFuture getTopicsUnderNamespace(NamespaceName namespace, Mode mode, diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java index e0eee02e95746..eb9cac0e82a8e 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java @@ -44,6 +44,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import lombok.AccessLevel; import lombok.Getter; import org.apache.commons.lang3.exception.ExceptionUtils; import org.apache.commons.lang3.tuple.Pair; @@ -59,6 +60,7 @@ import org.apache.pulsar.client.impl.transaction.TransactionBufferHandler; import org.apache.pulsar.client.util.TimedCompletableFuture; import org.apache.pulsar.common.api.AuthData; +import org.apache.pulsar.common.api.proto.BaseCommand; import org.apache.pulsar.common.api.proto.CommandAckResponse; import org.apache.pulsar.common.api.proto.CommandActiveConsumerChange; import org.apache.pulsar.common.api.proto.CommandAddPartitionToTxnResponse; @@ -85,6 +87,8 @@ import org.apache.pulsar.common.api.proto.CommandSendReceipt; import org.apache.pulsar.common.api.proto.CommandSuccess; import org.apache.pulsar.common.api.proto.CommandTcClientConnectResponse; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; +import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.lookup.GetTopicsResult; import org.apache.pulsar.common.protocol.Commands; @@ -128,6 +132,12 @@ public class ClientCnx extends PulsarHandler { .expectedItems(16) .concurrencyLevel(1) .build(); + @Getter(AccessLevel.PACKAGE) + private final ConcurrentLongHashMap topicListWatchers = + ConcurrentLongHashMap.newBuilder() + .expectedItems(16) + .concurrencyLevel(1) + .build(); private final CompletableFuture connectionFuture = new CompletableFuture(); private final ConcurrentLinkedQueue requestTimeoutQueue = new ConcurrentLinkedQueue<>(); @@ -160,6 +170,7 @@ public class ClientCnx extends PulsarHandler { @Getter protected AuthenticationDataProvider authenticationDataProvider; private TransactionBufferHandler transactionBufferHandler; + private boolean supportsTopicWatchers; enum State { None, SentConnectFrame, Ready, Failed, Connecting @@ -281,12 +292,14 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { producers.forEach((id, producer) -> producer.connectionClosed(this)); consumers.forEach((id, consumer) -> consumer.connectionClosed(this)); transactionMetaStoreHandlers.forEach((id, handler) -> handler.connectionClosed(this)); + topicListWatchers.forEach((__, watcher) -> watcher.connectionClosed(this)); pendingRequests.clear(); waitingLookupRequests.clear(); producers.clear(); consumers.clear(); + topicListWatchers.clear(); timeoutTask.cancel(true); } @@ -330,6 +343,10 @@ protected void handleConnected(CommandConnected connected) { if (log.isDebugEnabled()) { log.debug("{} Connection is ready", ctx.channel()); } + + supportsTopicWatchers = + connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTopicWatchers(); + // set remote protocol version to the correct version before we complete the connection future setRemoteEndpointProtocolVersion(connected.getProtocolVersion()); connectionFuture.complete(null); @@ -1036,6 +1053,62 @@ private TransactionBufferHandler checkAndGetTransactionBufferHandler() { return transactionBufferHandler; } + public CompletableFuture newWatchTopicList( + BaseCommand commandWatchTopicList, long requestId) { + if (!supportsTopicWatchers) { + return FutureUtil.failedFuture( + new PulsarClientException.NotAllowedException( + "Broker does not allow broker side pattern evaluation.")); + } + return sendRequestAndHandleTimeout(Commands.serializeWithSize(commandWatchTopicList), requestId, + RequestType.Command, true); + } + + public CompletableFuture newUnwatchTopicList( + BaseCommand commandUnwatchTopicList, long requestId) { + if (!supportsTopicWatchers) { + return FutureUtil.failedFuture( + new PulsarClientException.NotAllowedException( + "Broker does not allow broker side pattern evaluation.")); + } + return sendRequestAndHandleTimeout(Commands.serializeWithSize(commandUnwatchTopicList), requestId, + RequestType.Command, true); + } + + protected void handleCommandWatchTopicListSuccess(CommandWatchTopicListSuccess commandWatchTopicListSuccess) { + checkArgument(state == State.Ready); + + if (log.isDebugEnabled()) { + log.debug("{} Received watchTopicListSuccess response from server: {}", + ctx.channel(), commandWatchTopicListSuccess.getRequestId()); + } + long requestId = commandWatchTopicListSuccess.getRequestId(); + CompletableFuture requestFuture = + (CompletableFuture) pendingRequests.remove(requestId); + if (requestFuture != null) { + requestFuture.complete(commandWatchTopicListSuccess); + } else { + log.warn("{} Received unknown request id from server: {}", + ctx.channel(), commandWatchTopicListSuccess.getRequestId()); + } + } + + protected void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate commandWatchTopicUpdate) { + checkArgument(state == State.Ready); + + if (log.isDebugEnabled()) { + log.debug("{} Received watchTopicUpdate command from server: {}", + ctx.channel(), commandWatchTopicUpdate.getWatcherId()); + } + long watcherId = commandWatchTopicUpdate.getWatcherId(); + TopicListWatcher watcher = topicListWatchers.get(watcherId); + if (watcher != null) { + watcher.handleCommandWatchTopicUpdate(commandWatchTopicUpdate); + } else { + log.warn("{} Received topic list update for unknown watcher from server: {}", ctx.channel(), watcherId); + } + } + /** * check serverError and take appropriate action. *
    @@ -1082,6 +1155,11 @@ void registerTransactionMetaStoreHandler(final long transactionMetaStoreId, transactionMetaStoreHandlers.put(transactionMetaStoreId, handler); } + void registerTopicListWatcher(final long watcherId, final TopicListWatcher watcher) { + topicListWatchers.put(watcherId, watcher); + + } + public void registerTransactionBufferHandler(final TransactionBufferHandler handler) { transactionBufferHandler = handler; } @@ -1094,6 +1172,10 @@ void removeConsumer(final long consumerId) { consumers.remove(consumerId); } + void removeTopicListWatcher(final long watcherId) { + topicListWatchers.remove(watcherId); + } + void setTargetBroker(InetSocketAddress targetBrokerAddress) { this.proxyToTargetBrokerAddress = String.format("%s:%d", targetBrokerAddress.getHostString(), targetBrokerAddress.getPort()); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java index a951d7b2cb836..6c5a5be200fb5 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.client.impl; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLongFieldUpdater; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; @@ -69,8 +70,13 @@ protected void grabCnx() { } try { - state.client.getConnection(state.topic) // - .thenAccept(cnx -> connection.connectionOpened(cnx)) // + CompletableFuture cnxFuture; + if (state.topic == null) { + cnxFuture = state.client.getConnectionToServiceUrl(); + } else { + cnxFuture = state.client.getConnection(state.topic); // + } + cnxFuture.thenAccept(cnx -> connection.connectionOpened(cnx)) // .exceptionally(this::handleConnectionError); } catch (Throwable t) { log.warn("[{}] [{}] Exception thrown while getting connection: ", state.topic, state.getHandlerName(), t); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java index 82530661be06d..61d0c2bd64ecc 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java @@ -155,6 +155,10 @@ String getServiceUrl() { return this.serviceNameResolver.getServiceUrl(); } + public InetSocketAddress resolveHost() { + return serviceNameResolver.resolveHost(); + } + void setServiceUrl(String serviceUrl) throws PulsarClientException { this.serviceNameResolver.updateServiceUrl(serviceUrl); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java index 6d04c5fab3102..def19c45affb0 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java @@ -123,6 +123,11 @@ public String getServiceUrl() { return httpClient.getServiceUrl(); } + @Override + public InetSocketAddress resolveHost() { + return httpClient.resolveHost(); + } + @Override public CompletableFuture getTopicsUnderNamespace(NamespaceName namespace, Mode mode, String topicsPattern, String topicsHash) { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java index e7d358148c0c9..c1c93f36c6e56 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java @@ -90,6 +90,13 @@ public interface LookupService extends AutoCloseable { */ String getServiceUrl(); + /** + * Resolves pulsar service url. + * + * @return the service url resolved to a socket address + */ + InetSocketAddress resolveHost(); + /** * Returns all the topics name for a given namespace. * diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index 9d00b2c282760..79a7c6a3ae62b 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -21,6 +21,7 @@ import static com.google.common.base.Preconditions.checkArgument; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Lists; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import io.netty.util.Timeout; import io.netty.util.TimerTask; import java.util.ArrayList; @@ -47,6 +48,7 @@ public class PatternMultiTopicsConsumerImpl extends MultiTopicsConsumerImpl watcherFuture; protected NamespaceName namespaceName; private volatile Timeout recheckPatternTimeout = null; private volatile String topicsHash; @@ -74,6 +76,19 @@ public PatternMultiTopicsConsumerImpl(Pattern topicsPattern, this.topicsChangeListener = new PatternTopicsChangedListener(); this.recheckPatternTimeout = client.timer() .newTimeout(this, Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS); + this.watcherFuture = new CompletableFuture<>(); + if (subscriptionMode == Mode.PERSISTENT) { + long watcherId = client.newTopicListWatcherId(); + new TopicListWatcher(topicsChangeListener, client, topicsPattern, watcherId, + namespaceName, topicsHash, watcherFuture); + watcherFuture.exceptionally(ex -> { + log.debug("Unable to create topic list watcher. Falling back to only polling for new topics", ex); + return null; + }); + } else { + log.debug("Not creating topic list watcher for subscription mode {}", subscriptionMode); + watcherFuture.complete(null); + } } public static NamespaceName getNameSpaceFromPattern(Pattern pattern) { @@ -172,7 +187,7 @@ public CompletableFuture onTopicsRemoved(Collection removedTopics) FutureUtil.waitForAll(futures) .thenAccept(finalFuture -> removeFuture.complete(null)) .exceptionally(ex -> { - log.warn("[{}] Failed to subscribe topics: {}", topic, ex.getMessage()); + log.warn("[{}] Failed to unsubscribe from topics: {}", topic, ex.getMessage()); removeFuture.completeExceptionally(ex); return null; }); @@ -194,7 +209,7 @@ public CompletableFuture onTopicsAdded(Collection addedTopics) { FutureUtil.waitForAll(futures) .thenAccept(finalFuture -> addFuture.complete(null)) .exceptionally(ex -> { - log.warn("[{}] Failed to unsubscribe topics: {}", topic, ex.getMessage()); + log.warn("[{}] Failed to subscribe to topics: {}", topic, ex.getMessage()); addFuture.completeExceptionally(ex); return null; }); @@ -203,13 +218,23 @@ public CompletableFuture onTopicsAdded(Collection addedTopics) { } @Override + @SuppressFBWarnings public CompletableFuture closeAsync() { Timeout timeout = recheckPatternTimeout; if (timeout != null) { timeout.cancel(); recheckPatternTimeout = null; } - return super.closeAsync(); + List> closeFutures = new ArrayList<>(2); + if (watcherFuture.isDone() && !watcherFuture.isCompletedExceptionally()) { + TopicListWatcher watcher = watcherFuture.getNow(null); + // watcher can be null when subscription mode is not persistent + if (watcher != null) { + closeFutures.add(watcher.closeAsync()); + } + } + closeFutures.add(super.closeAsync()); + return FutureUtil.waitForAll(closeFutures); } @VisibleForTesting diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index 8a7d50f0d4b87..d7b7f630f101d 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -119,9 +119,11 @@ public enum State { // These sets are updated from multiple threads, so they require a threadsafe data structure private final Set> producers = Collections.newSetFromMap(new ConcurrentHashMap<>()); private final Set> consumers = Collections.newSetFromMap(new ConcurrentHashMap<>()); + private final Set topicListWatchers = Collections.newSetFromMap(new ConcurrentHashMap<>()); private final AtomicLong producerIdGenerator = new AtomicLong(); private final AtomicLong consumerIdGenerator = new AtomicLong(); + private final AtomicLong topicListWatcherIdGenerator = new AtomicLong(); private final AtomicLong requestIdGenerator = new AtomicLong(ThreadLocalRandom.current().nextLong(0, Long.MAX_VALUE / 2)); @@ -739,6 +741,12 @@ public CompletableFuture closeAsync() { } return null; }))); + topicListWatchers.forEach(c -> futures.add(c.closeAsync().handle((__, t) -> { + if (t != null) { + log.error("Error closing topic list watcher {}", c, t); + } + return null; + }))); // Need to run the shutdown sequence in a separate thread to prevent deadlocks // If there are consumers or producers that need to be shutdown we cannot use the same thread @@ -929,6 +937,11 @@ public CompletableFuture getConnection(final String topic) { .thenCompose(pair -> getConnection(pair.getLeft(), pair.getRight())); } + public CompletableFuture getConnectionToServiceUrl() { + InetSocketAddress address = lookup.resolveHost(); + return getConnection(address, address); + } + public CompletableFuture getConnection(final InetSocketAddress logicalAddress, final InetSocketAddress physicalAddress) { return cnxPool.getConnection(logicalAddress, physicalAddress); @@ -951,6 +964,10 @@ long newConsumerId() { return consumerIdGenerator.getAndIncrement(); } + long newTopicListWatcherId() { + return topicListWatcherIdGenerator.getAndIncrement(); + } + public long newRequestId() { return requestIdGenerator.getAndIncrement(); } @@ -1063,6 +1080,10 @@ void cleanupConsumer(ConsumerBase consumer) { consumers.remove(consumer); } + void cleanupTopicListWatcher(TopicListWatcher watcher) { + topicListWatchers.remove(watcher); + } + @VisibleForTesting int producersCount() { return producers.size(); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java new file mode 100644 index 0000000000000..90d930708aa90 --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java @@ -0,0 +1,287 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.impl; + +import io.netty.channel.ChannelHandlerContext; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Pattern; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.common.api.proto.BaseCommand; +import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.protocol.Commands; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class TopicListWatcher extends HandlerState implements ConnectionHandler.Connection { + + private static final Logger log = LoggerFactory.getLogger(TopicListWatcher.class); + + private static final AtomicLongFieldUpdater CREATE_WATCHER_DEADLINE_UPDATER = + AtomicLongFieldUpdater + .newUpdater(TopicListWatcher.class, "createWatcherDeadline"); + + private final PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener; + private final String name; + private final ConnectionHandler connectionHandler; + private final Pattern topicsPattern; + private final long watcherId; + private volatile long createWatcherDeadline = 0; + private final NamespaceName namespace; + // TODO maintain the value based on updates from broker and warn the user if inconsistent with hash from polling + private String topicsHash; + private final CompletableFuture watcherFuture; + + private final List previousExceptions = new CopyOnWriteArrayList<>(); + private final AtomicReference clientCnxUsedForWatcherRegistration = new AtomicReference<>(); + + + public TopicListWatcher(PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener, + PulsarClientImpl client, Pattern topicsPattern, long watcherId, + NamespaceName namespace, String topicsHash, + CompletableFuture watcherFuture) { + super(client, null); + this.topicsChangeListener = topicsChangeListener; + this.name = "Watcher(" + topicsPattern + ")"; + this.connectionHandler = new ConnectionHandler(this, + new BackoffBuilder() + .setInitialTime(client.getConfiguration().getInitialBackoffIntervalNanos(), + TimeUnit.NANOSECONDS) + .setMax(client.getConfiguration().getMaxBackoffIntervalNanos(), TimeUnit.NANOSECONDS) + .setMandatoryStop(0, TimeUnit.MILLISECONDS) + .create(), + this); + this.topicsPattern = topicsPattern; + this.watcherId = watcherId; + this.namespace = namespace; + this.topicsHash = topicsHash; + this.watcherFuture = watcherFuture; + + connectionHandler.grabCnx(); + } + + @Override + public void connectionFailed(PulsarClientException exception) { + boolean nonRetriableError = !PulsarClientException.isRetriableError(exception); + if (nonRetriableError) { + exception.setPreviousExceptions(previousExceptions); + if (watcherFuture.completeExceptionally(exception)) { + setState(State.Failed); + log.info("[{}] Watcher creation failed for {} with non-retriable error {}", + topic, name, exception); + deregisterFromClientCnx(); + client.cleanupTopicListWatcher(this); + } + } else { + previousExceptions.add(exception); + } + } + + @Override + public void connectionOpened(ClientCnx cnx) { + previousExceptions.clear(); + + if (getState() == State.Closing || getState() == State.Closed) { + setState(State.Closed); + deregisterFromClientCnx(); + client.cleanupTopicListWatcher(this); + return; + } + + log.info("[{}][{}] Creating topic list watcher on cnx {}, watcherId {}", + topic, getHandlerName(), cnx.ctx().channel(), watcherId); + + long requestId = client.newRequestId(); + + CREATE_WATCHER_DEADLINE_UPDATER + .compareAndSet(this, 0L, System.currentTimeMillis() + + client.getConfiguration().getOperationTimeoutMs()); + + // synchronized this, because redeliverUnAckMessage eliminate the epoch inconsistency between them + synchronized (this) { + setClientCnx(cnx); + BaseCommand watchRequest = Commands.newWatchTopicList(requestId, watcherId, namespace.toString(), + topicsPattern.pattern(), topicsHash); + + cnx.newWatchTopicList(watchRequest, requestId) + + .thenAccept(response -> { + synchronized (TopicListWatcher.this) { + if (!changeToReadyState()) { + // Watcher was closed while reconnecting, close the connection to make sure the broker + // drops the watcher on its side + setState(State.Closed); + deregisterFromClientCnx(); + client.cleanupTopicListWatcher(this); + + cnx.channel().close(); + return; + } + } + + this.connectionHandler.resetBackoff(); + + watcherFuture.complete(this); + + }).exceptionally((e) -> { + deregisterFromClientCnx(); + if (getState() == State.Closing || getState() == State.Closed) { + // Watcher was closed while reconnecting, close the connection to make sure the broker + // drops the watcher on its side + cnx.channel().close(); + return null; + } + log.warn("[{}][{}] Failed to subscribe to topic on {}", topic, + getHandlerName(), cnx.channel().remoteAddress()); + + if (e.getCause() instanceof PulsarClientException + && PulsarClientException.isRetriableError(e.getCause()) + && System.currentTimeMillis() + < CREATE_WATCHER_DEADLINE_UPDATER.get(TopicListWatcher.this)) { + reconnectLater(e.getCause()); + } else if (!watcherFuture.isDone()) { + // unable to create new watcher, fail operation + setState(State.Failed); + watcherFuture.completeExceptionally( + PulsarClientException.wrap(e, String.format("Failed to create topic list watcher %s" + + "when connecting to the broker", getHandlerName()))); + client.cleanupTopicListWatcher(this); + + } else { + // watcher was subscribed and connected, but we got some error, keep trying + reconnectLater(e.getCause()); + } + return null; + }); + } + } + + @Override + String getHandlerName() { + return name; + } + + public boolean isConnected() { + return getClientCnx() != null && (getState() == State.Ready); + } + + public ClientCnx getClientCnx() { + return this.connectionHandler.cnx(); + } + + public CompletableFuture closeAsync() { + + CompletableFuture closeFuture = new CompletableFuture<>(); + + if (getState() == State.Closing || getState() == State.Closed) { + closeFuture.complete(null); + return closeFuture; + } + + if (!isConnected()) { + log.info("[{}] [{}] Closed watcher (not connected)", topic, getHandlerName()); + setState(State.Closed); + deregisterFromClientCnx(); + client.cleanupTopicListWatcher(this); + + closeFuture.complete(null); + return closeFuture; + } + + setState(State.Closing); + + + long requestId = client.newRequestId(); + + ClientCnx cnx = cnx(); + if (null == cnx) { + cleanupAtClose(closeFuture, null); + } else { + BaseCommand cmd = Commands.newUnwatchTopicList(watcherId, requestId); + cnx.sendRequestWithId(Commands.serializeWithSize(cmd), requestId).handle((v, exception) -> { + final ChannelHandlerContext ctx = cnx.ctx(); + boolean ignoreException = ctx == null || !ctx.channel().isActive(); + if (ignoreException && exception != null) { + log.debug("Exception ignored in closing watcher", exception); + } + cleanupAtClose(closeFuture, ignoreException ? null : exception); + return null; + }); + } + + return closeFuture; + } + + // wrapper for connection methods + ClientCnx cnx() { + return this.connectionHandler.cnx(); + } + + public void connectionClosed(ClientCnx clientCnx) { + this.connectionHandler.connectionClosed(clientCnx); + } + + void setClientCnx(ClientCnx clientCnx) { + if (clientCnx != null) { + this.connectionHandler.setClientCnx(clientCnx); + clientCnx.registerTopicListWatcher(watcherId, this); + } + ClientCnx previousClientCnx = clientCnxUsedForWatcherRegistration.getAndSet(clientCnx); + if (previousClientCnx != null && previousClientCnx != clientCnx) { + previousClientCnx.removeTopicListWatcher(watcherId); + } + } + + void deregisterFromClientCnx() { + setClientCnx(null); + } + + void reconnectLater(Throwable exception) { + this.connectionHandler.reconnectLater(exception); + } + + + private void cleanupAtClose(CompletableFuture closeFuture, Throwable exception) { + log.info("[{}] Closed topic list watcher", getHandlerName()); + setState(State.Closed); + deregisterFromClientCnx(); + client.cleanupTopicListWatcher(this); + if (exception != null) { + closeFuture.completeExceptionally(exception); + } else { + closeFuture.complete(null); + } + } + + public void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate update) { + List deleted = update.getDeletedTopicsList(); + if (!deleted.isEmpty()) { + topicsChangeListener.onTopicsRemoved(deleted); + } + List added = update.getNewTopicsList(); + if (!added.isEmpty()) { + topicsChangeListener.onTopicsAdded(added); + } + } +} diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java index 6ce4afecd02bd..07c8136f1cd48 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java @@ -20,8 +20,10 @@ import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; import io.netty.buffer.ByteBuf; @@ -34,12 +36,16 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ThreadFactory; +import java.util.function.Consumer; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.PulsarClientException.BrokerMetadataException; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.common.api.proto.CommandCloseConsumer; import org.apache.pulsar.common.api.proto.CommandCloseProducer; +import org.apache.pulsar.common.api.proto.CommandConnected; import org.apache.pulsar.common.api.proto.CommandError; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; +import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.PulsarHandler; @@ -190,4 +196,106 @@ public void testHandleCloseProducer() { eventLoop.shutdownGracefully(); } + + @Test + public void testNoWatchersWhenNoServerSupport() { + withConnection("testNoWatchersWhenNoServerSupport", cnx -> { + cnx.handleConnected(new CommandConnected() + .setServerVersion("Some old Server") + .setProtocolVersion(1)); + + CompletableFuture result = + cnx.newWatchTopicList(Commands.newWatchTopicList(7, 5, "tenant/ns", + ".*", null), 7); + assertTrue(result.isCompletedExceptionally()); + }); + } + + @Test + public void testCreateWatcher() { + withConnection("testCreateWatcher", cnx -> { + CommandConnected connected = new CommandConnected() + .setServerVersion("Some strange Server") + .setProtocolVersion(1); + connected.setFeatureFlags().setSupportsTopicWatchers(true); + cnx.handleConnected(connected); + + CompletableFuture result = + cnx.newWatchTopicList(Commands.newWatchTopicList(7, 5, "tenant/ns", + ".*", null), 7); + verify(cnx.ctx()).writeAndFlush(any(ByteBuf.class)); + assertFalse(result.isDone()); + + CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess() + .setRequestId(7) + .setWatcherId(5).setTopicsHash("f00"); + cnx.handleCommandWatchTopicListSuccess(success); + assertEquals(result.getNow(null), success); + }); + } + + + + @Test + public void testUpdateWatcher() { + withConnection("testUpdateWatcher", cnx -> { + CommandConnected connected = new CommandConnected() + .setServerVersion("Some Strange Server") + .setProtocolVersion(1); + connected.setFeatureFlags().setSupportsTopicWatchers(true); + cnx.handleConnected(connected); + + cnx.newWatchTopicList(Commands.newWatchTopicList(7, 5, "tenant/ns", ".*", null), 7); + + CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess() + .setRequestId(7) + .setWatcherId(5).setTopicsHash("f00"); + cnx.handleCommandWatchTopicListSuccess(success); + + TopicListWatcher watcher = mock(TopicListWatcher.class); + cnx.registerTopicListWatcher(5, watcher); + + CommandWatchTopicUpdate update = new CommandWatchTopicUpdate() + .setWatcherId(5) + .setTopicsHash("ADD"); + update.addNewTopic("persistent://tenant/ns/topic"); + cnx.handleCommandWatchTopicUpdate(update); + verify(watcher).handleCommandWatchTopicUpdate(update); + }); + } + + private void withConnection(String testName, Consumer test) { + ThreadFactory threadFactory = new DefaultThreadFactory(testName); + EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, threadFactory); + try { + + ClientConfigurationData conf = new ClientConfigurationData(); + ClientCnx cnx = new ClientCnx(conf, eventLoop); + + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + Channel channel = mock(Channel.class); + when(ctx.channel()).thenReturn(channel); + + ChannelFuture listenerFuture = mock(ChannelFuture.class); + when(listenerFuture.addListener(any())).thenReturn(listenerFuture); + when(ctx.writeAndFlush(any())).thenReturn(listenerFuture); + + Field ctxField = PulsarHandler.class.getDeclaredField("ctx"); + ctxField.setAccessible(true); + ctxField.set(cnx, ctx); + + // set connection as SentConnectFrame + Field cnxField = ClientCnx.class.getDeclaredField("state"); + cnxField.setAccessible(true); + cnxField.set(cnx, ClientCnx.State.SentConnectFrame); + + test.accept(cnx); + + } catch (NoSuchFieldException | IllegalAccessException e) { + fail("Error using reflection on ClientCnx", e); + } finally { + eventLoop.shutdownGracefully(); + } + } + } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java index 4c174ff0ac3f1..9d907ca83236a 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java @@ -20,6 +20,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyObject; import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -119,7 +120,7 @@ public void testConsumerIsClosed() throws Exception { new GetTopicsResult(Collections.emptyList(), null, false, true))); when(lookup.getPartitionedTopicMetadata(any(TopicName.class))) .thenReturn(CompletableFuture.completedFuture(new PartitionedTopicMetadata())); - when(lookup.getBroker(any(TopicName.class))) + when(lookup.getBroker(any())) .thenReturn(CompletableFuture.completedFuture( Pair.of(mock(InetSocketAddress.class), mock(InetSocketAddress.class)))); ConnectionPool pool = mock(ConnectionPool.class); diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java new file mode 100644 index 0000000000000..1d245350c82cb --- /dev/null +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java @@ -0,0 +1,111 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.impl; + +import io.netty.channel.ChannelHandlerContext; +import org.apache.pulsar.client.impl.PatternMultiTopicsConsumerImpl.TopicsChangedListener; +import org.apache.pulsar.client.impl.conf.ClientConfigurationData; +import org.apache.pulsar.common.api.proto.BaseCommand; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; +import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; +import org.apache.pulsar.common.naming.NamespaceName; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertTrue; +import org.mockito.ArgumentCaptor; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import java.util.Collections; +import java.util.concurrent.CompletableFuture; +import java.util.regex.Pattern; + +public class TopicListWatcherTest { + + private CompletableFuture clientCnxFuture; + private TopicListWatcher watcher; + private PulsarClientImpl client; + private CompletableFuture watcherFuture; + private TopicsChangedListener listener; + + @BeforeMethod(alwaysRun = true) + public void setup() { + listener = mock(TopicsChangedListener.class); + client = mock(PulsarClientImpl.class); + when(client.getConfiguration()).thenReturn(new ClientConfigurationData()); + clientCnxFuture = new CompletableFuture<>(); + when(client.getConnectionToServiceUrl()).thenReturn(clientCnxFuture); + watcherFuture = new CompletableFuture<>(); + watcher = new TopicListWatcher(listener, client, + Pattern.compile("persistent://tenant/ns/topic\\d+"), 7, + NamespaceName.get("tenant/ns"), null, watcherFuture); + } + + @Test + public void testWatcherGrabsConnection() { + verify(client).getConnectionToServiceUrl(); + } + + @Test + public void testWatcherCreatesBrokerSideObjectWhenConnected() { + ClientCnx clientCnx = mock(ClientCnx.class); + CompletableFuture responseFuture = new CompletableFuture<>(); + ArgumentCaptor commandCaptor = ArgumentCaptor.forClass(BaseCommand.class); + when(clientCnx.newWatchTopicList(any(BaseCommand.class), anyLong())).thenReturn(responseFuture); + when(clientCnx.ctx()).thenReturn(mock(ChannelHandlerContext.class)); + clientCnxFuture.complete(clientCnx); + verify(clientCnx).newWatchTopicList(commandCaptor.capture(), anyLong()); + CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess() + .setWatcherId(7) + .setRequestId(commandCaptor.getValue().getWatchTopicList().getRequestId()) + .setTopicsHash("FEED"); + success.addTopic("persistent://tenant/ns/topic11"); + responseFuture.complete(success); + assertTrue(watcherFuture.isDone() && !watcherFuture.isCompletedExceptionally()); + } + + @Test + public void testWatcherCallsListenerOnUpdate() { + ClientCnx clientCnx = mock(ClientCnx.class); + CompletableFuture responseFuture = new CompletableFuture<>(); + ArgumentCaptor commandCaptor = ArgumentCaptor.forClass(BaseCommand.class); + when(clientCnx.newWatchTopicList(any(BaseCommand.class), anyLong())).thenReturn(responseFuture); + when(clientCnx.ctx()).thenReturn(mock(ChannelHandlerContext.class)); + clientCnxFuture.complete(clientCnx); + verify(clientCnx).newWatchTopicList(commandCaptor.capture(), anyLong()); + CommandWatchTopicListSuccess success = new CommandWatchTopicListSuccess() + .setWatcherId(7) + .setRequestId(commandCaptor.getValue().getWatchTopicList().getRequestId()) + .setTopicsHash("FEED"); + success.addTopic("persistent://tenant/ns/topic11"); + responseFuture.complete(success); + + CommandWatchTopicUpdate update = new CommandWatchTopicUpdate() + .setTopicsHash("F33D") + .setWatcherId(7) + .addAllNewTopics(Collections.singleton("persistent://tenant/ns/topic12")); + + watcher.handleCommandWatchTopicUpdate(update); + verify(listener).onTopicsAdded(Collections.singletonList("persistent://tenant/ns/topic12")); + } + + +} diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index 46c22375b778e..4da97eff815d9 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -263,11 +263,12 @@ public static ByteBuf newConnect(String authMethodName, AuthData authData, int p return serializeWithSize(cmd); } - public static ByteBuf newConnected(int clientProtocoVersion) { - return newConnected(clientProtocoVersion, INVALID_MAX_MESSAGE_SIZE); + public static ByteBuf newConnected(int clientProtocoVersion, boolean supportsTopicWatchers) { + return newConnected(clientProtocoVersion, INVALID_MAX_MESSAGE_SIZE, supportsTopicWatchers); } - public static BaseCommand newConnectedCommand(int clientProtocolVersion, int maxMessageSize) { + public static BaseCommand newConnectedCommand(int clientProtocolVersion, int maxMessageSize, + boolean supportsTopicWatchers) { BaseCommand cmd = localCmd(Type.CONNECTED); CommandConnected connected = cmd.setConnected() .setServerVersion("Pulsar Server" + PulsarVersion.getVersion()); @@ -282,11 +283,13 @@ public static BaseCommand newConnectedCommand(int clientProtocolVersion, int max int versionToAdvertise = Math.min(currentProtocolVersion, clientProtocolVersion); connected.setProtocolVersion(versionToAdvertise); + + connected.setFeatureFlags().setSupportsTopicWatchers(supportsTopicWatchers); return cmd; } - public static ByteBuf newConnected(int clientProtocolVersion, int maxMessageSize) { - return serializeWithSize(newConnectedCommand(clientProtocolVersion, maxMessageSize)); + public static ByteBuf newConnected(int clientProtocolVersion, int maxMessageSize, boolean supportsTopicWatchers) { + return serializeWithSize(newConnectedCommand(clientProtocolVersion, maxMessageSize, supportsTopicWatchers)); } public static ByteBuf newAuthChallenge(String authMethod, AuthData brokerData, int clientProtocolVersion) { @@ -1474,6 +1477,55 @@ public static ByteBuf newEndTxnOnSubscriptionResponse(long requestId, long txnId return serializeWithSize(cmd); } + public static BaseCommand newWatchTopicList( + long requestId, long watcherId, String namespace, String topicsPattern, String topicsHash) { + BaseCommand cmd = localCmd(Type.WATCH_TOPIC_LIST); + cmd.setWatchTopicList() + .setRequestId(requestId) + .setNamespace(namespace) + .setTopicsPattern(topicsPattern) + .setWatcherId(watcherId); + if (topicsHash != null) { + cmd.getWatchTopicList() + .setTopicsHash(topicsHash); + } + return cmd; + } + + public static BaseCommand newWatchTopicListSuccess(long requestId, long watcherId, String topicsHash, + List topics) { + BaseCommand cmd = localCmd(Type.WATCH_TOPIC_LIST_SUCCESS); + cmd.setWatchTopicListSuccess() + .setRequestId(requestId) + .setWatcherId(watcherId); + if (topicsHash != null) { + cmd.getWatchTopicListSuccess().setTopicsHash(topicsHash); + } + if (topics != null && !topics.isEmpty()) { + cmd.getWatchTopicListSuccess().addAllTopics(topics); + } + return cmd; + } + + public static BaseCommand newWatchTopicUpdate(long watcherId, + List newTopics, List deletedTopics, String topicsHash) { + BaseCommand cmd = localCmd(Type.WATCH_TOPIC_UPDATE); + cmd.setWatchTopicUpdate() + .setWatcherId(watcherId) + .setTopicsHash(topicsHash) + .addAllNewTopics(newTopics) + .addAllDeletedTopics(deletedTopics); + return cmd; + } + + public static BaseCommand newUnwatchTopicList(long watcherId, long requestId) { + BaseCommand cmd = localCmd(Type.UNWATCH_TOPIC_LIST); + cmd.setUnwatchTopicList() + .setRequestId(requestId) + .setWatcherId(watcherId); + return cmd; + } + public static ByteBuf serializeWithSize(BaseCommand cmd) { // / Wire format // [TOTAL_SIZE] [CMD_SIZE][CMD] diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java index 9fb22d1a289a1..9b0a81a7db112 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java @@ -77,6 +77,10 @@ import org.apache.pulsar.common.api.proto.CommandTcClientConnectRequest; import org.apache.pulsar.common.api.proto.CommandTcClientConnectResponse; import org.apache.pulsar.common.api.proto.CommandUnsubscribe; +import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; +import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.intercept.InterceptException; import org.slf4j.Logger; @@ -432,6 +436,27 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception checkArgument(cmd.hasEndTxnOnSubscriptionResponse()); handleEndTxnOnSubscriptionResponse(cmd.getEndTxnOnSubscriptionResponse()); break; + + case WATCH_TOPIC_LIST: + checkArgument(cmd.hasWatchTopicList()); + handleCommandWatchTopicList(cmd.getWatchTopicList()); + break; + + case WATCH_TOPIC_LIST_SUCCESS: + checkArgument(cmd.hasWatchTopicListSuccess()); + handleCommandWatchTopicListSuccess(cmd.getWatchTopicListSuccess()); + break; + + case WATCH_TOPIC_UPDATE: + checkArgument(cmd.hasWatchTopicUpdate()); + handleCommandWatchTopicUpdate(cmd.getWatchTopicUpdate()); + break; + + case UNWATCH_TOPIC_LIST: + checkArgument(cmd.hasUnwatchTopicList()); + handleCommandUnwatchTopicList(cmd.getUnwatchTopicList()); + break; + default: break; } @@ -672,5 +697,22 @@ protected void handleEndTxnOnSubscriptionResponse( throw new UnsupportedOperationException(); } + protected void handleCommandWatchTopicList(CommandWatchTopicList commandWatchTopicList) { + throw new UnsupportedOperationException(); + } + + protected void handleCommandWatchTopicListSuccess( + CommandWatchTopicListSuccess commandWatchTopicListSuccess) { + throw new UnsupportedOperationException(); + } + + protected void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate commandWatchTopicUpdate) { + throw new UnsupportedOperationException(); + } + + protected void handleCommandUnwatchTopicList(CommandUnwatchTopicList commandUnwatchTopicList) { + throw new UnsupportedOperationException(); + } + private static final Logger log = LoggerFactory.getLogger(PulsarDecoder.class); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java b/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java index ed37386789869..574e4d5bd2637 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java @@ -32,6 +32,8 @@ @UtilityClass public class TopicList { + public static final String ALL_TOPICS_PATTERN = ".*"; + private static final String SCHEME_SEPARATOR = "://"; private static final Pattern SCHEME_SEPARATOR_PATTERN = Pattern.compile(Pattern.quote(SCHEME_SEPARATOR)); diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index fc8e5a5b7c66c..e94563d700ad2 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -294,12 +294,14 @@ message FeatureFlags { optional bool supports_auth_refresh = 1 [default = false]; optional bool supports_broker_entry_metadata = 2 [default = false]; optional bool supports_partial_producer = 3 [default = false]; + optional bool supports_topic_watchers = 4 [default = false]; } message CommandConnected { required string server_version = 1; optional int32 protocol_version = 2 [default = 0]; optional int32 max_message_size = 3; + optional FeatureFlags feature_flags = 4; } message CommandAuthResponse { @@ -757,6 +759,34 @@ message CommandGetTopicsOfNamespaceResponse { optional bool changed = 5 [default = true]; } +message CommandWatchTopicList { + required uint64 request_id = 1; + required uint64 watcher_id = 2; + required string namespace = 3; + required string topics_pattern = 4; + // Only present when the client reconnects: + optional string topics_hash = 5; +} + +message CommandWatchTopicListSuccess { + required uint64 request_id = 1; + required uint64 watcher_id = 2; + repeated string topic = 3; + required string topics_hash = 4; +} + +message CommandWatchTopicUpdate { + required uint64 watcher_id = 1; + repeated string new_topics = 2; + repeated string deleted_topics = 3; + required string topics_hash = 4; +} + +message CommandUnwatchTopicList { + required uint64 request_id = 1; + required uint64 watcher_id = 2; +} + message CommandGetSchema { required uint64 request_id = 1; required string topic = 2; @@ -987,6 +1017,11 @@ message BaseCommand { TC_CLIENT_CONNECT_REQUEST = 62; TC_CLIENT_CONNECT_RESPONSE = 63; + WATCH_TOPIC_LIST = 64; + WATCH_TOPIC_LIST_SUCCESS = 65; + WATCH_TOPIC_UPDATE = 66; + UNWATCH_TOPIC_LIST = 67; + } @@ -1063,4 +1098,9 @@ message BaseCommand { optional CommandEndTxnOnSubscriptionResponse endTxnOnSubscriptionResponse = 61; optional CommandTcClientConnectRequest tcClientConnectRequest = 62; optional CommandTcClientConnectResponse tcClientConnectResponse = 63; + + optional CommandWatchTopicList watchTopicList = 64; + optional CommandWatchTopicListSuccess watchTopicListSuccess = 65; + optional CommandWatchTopicUpdate watchTopicUpdate = 66; + optional CommandUnwatchTopicList unwatchTopicList = 67; } diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index d770e63dcee2e..c1bf0fa96bc25 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -330,7 +330,7 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie // partitions metadata lookups state = State.ProxyLookupRequests; lookupProxyHandler = new LookupProxyHandler(service, this); - ctx.writeAndFlush(Commands.newConnected(protocolVersionToAdvertise)) + ctx.writeAndFlush(Commands.newConnected(protocolVersionToAdvertise, false)) .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); } } @@ -342,7 +342,8 @@ private void handleBrokerConnected(DirectProxyHandler directProxyHandler, Comman state = State.ProxyConnectionToBroker; int maxMessageSize = connected.hasMaxMessageSize() ? connected.getMaxMessageSize() : Commands.INVALID_MAX_MESSAGE_SIZE; - ctx.writeAndFlush(Commands.newConnected(connected.getProtocolVersion(), maxMessageSize)) + ctx.writeAndFlush(Commands.newConnected(connected.getProtocolVersion(), maxMessageSize, + connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTopicWatchers())) .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); } else { LOG.warn("[{}] Channel is {}. ProxyConnection is in {}. " From 1f4d9ade317f6224fb50bd83074c7d98c2fcbbbd Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Wed, 15 Jun 2022 08:24:27 +0200 Subject: [PATCH 2/7] Fix checkstyle violation --- .../java/org/apache/pulsar/client/impl/PulsarClientImplTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java index 9d907ca83236a..4d9977f547803 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java @@ -20,7 +20,6 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; -import static org.mockito.ArgumentMatchers.anyObject; import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; From 577c717fb270570027a9f6890ff60d83cee150f2 Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Wed, 15 Jun 2022 10:22:49 +0200 Subject: [PATCH 3/7] Fix cpp client compile error --- pulsar-client-cpp/lib/Commands.cc | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/pulsar-client-cpp/lib/Commands.cc b/pulsar-client-cpp/lib/Commands.cc index 33134a6c04bef..041cad4c4eccc 100644 --- a/pulsar-client-cpp/lib/Commands.cc +++ b/pulsar-client-cpp/lib/Commands.cc @@ -669,6 +669,18 @@ std::string Commands::messageType(BaseCommand_Type type) { case BaseCommand::TC_CLIENT_CONNECT_RESPONSE: return "TC_CLIENT_CONNECT_RESPONSE"; break; + case BaseCommand::WATCH_TOPIC_LIST: + return "WATCH_TOPIC_LIST"; + break; + case BaseCommand::WATCH_TOPIC_LIST_SUCCESS: + return "WATCH_TOPIC_LIST_SUCCESS"; + break; + case BaseCommand::WATCH_TOPIC_UPDATE: + return "WATCH_TOPIC_UPDATE"; + break; + case BaseCommand::UNWATCH_TOPIC_LIST: + return "UNWATCH_TOPIC_LIST"; + break; }; BOOST_THROW_EXCEPTION(std::logic_error("Invalid BaseCommand enumeration value")); } From 3a97b697f3188f57232f017bf1b59c0ff6d4ec09 Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Thu, 23 Jun 2022 14:02:39 +0200 Subject: [PATCH 4/7] Remove unused code, remove failed watchers --- .../pulsar/broker/service/TopicListService.java | 6 +++++- .../java/org/apache/pulsar/client/impl/ClientCnx.java | 11 ----------- .../apache/pulsar/client/impl/PulsarClientImpl.java | 11 ----------- .../apache/pulsar/client/impl/TopicListWatcher.java | 9 --------- .../org/apache/pulsar/client/impl/ClientCnxTest.java | 1 + 5 files changed, 6 insertions(+), 32 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java index b3e232955cca8..4764ec87b0c29 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java @@ -163,7 +163,8 @@ public void handleWatchTopicList(CommandWatchTopicList commandWatchTopicList, Se } - watcherFuture.thenAccept(watcher -> { + CompletableFuture finalWatcherFuture = watcherFuture; + finalWatcherFuture.thenAccept(watcher -> { List topicList = watcher.getMatchingTopics(); String hash = TopicList.calculateHash(topicList); if (commandWatchTopicList.hasTopicsHash() @@ -184,6 +185,7 @@ public void handleWatchTopicList(CommandWatchTopicList commandWatchTopicList, Se connection.getCommandSender().sendErrorResponse(requestId, BrokerServiceException.getClientErrorCode( new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage()); + watchers.remove(watcherId, finalWatcherFuture); lookupSemaphore.release(); return null; }); @@ -230,12 +232,14 @@ public void deleteTopicListWatcher(Long watcherId) { // create operation will complete, the new watcher will be discarded. log.info("[{}] Closed watcher before its creation was completed. watcherId={}", connection.getRemoteAddress(), watcherId); + watchers.remove(watcherId); return; } if (watcherFuture.isCompletedExceptionally()) { log.info("[{}] Closed watcher that already failed to be created. watcherId={}", connection.getRemoteAddress(), watcherId); + watchers.remove(watcherId); return; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java index eb9cac0e82a8e..94cfbd7dd2778 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java @@ -1064,17 +1064,6 @@ public CompletableFuture newWatchTopicList( RequestType.Command, true); } - public CompletableFuture newUnwatchTopicList( - BaseCommand commandUnwatchTopicList, long requestId) { - if (!supportsTopicWatchers) { - return FutureUtil.failedFuture( - new PulsarClientException.NotAllowedException( - "Broker does not allow broker side pattern evaluation.")); - } - return sendRequestAndHandleTimeout(Commands.serializeWithSize(commandUnwatchTopicList), requestId, - RequestType.Command, true); - } - protected void handleCommandWatchTopicListSuccess(CommandWatchTopicListSuccess commandWatchTopicListSuccess) { checkArgument(state == State.Ready); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index d7b7f630f101d..24ecd9b2f93ac 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -119,7 +119,6 @@ public enum State { // These sets are updated from multiple threads, so they require a threadsafe data structure private final Set> producers = Collections.newSetFromMap(new ConcurrentHashMap<>()); private final Set> consumers = Collections.newSetFromMap(new ConcurrentHashMap<>()); - private final Set topicListWatchers = Collections.newSetFromMap(new ConcurrentHashMap<>()); private final AtomicLong producerIdGenerator = new AtomicLong(); private final AtomicLong consumerIdGenerator = new AtomicLong(); @@ -741,12 +740,6 @@ public CompletableFuture closeAsync() { } return null; }))); - topicListWatchers.forEach(c -> futures.add(c.closeAsync().handle((__, t) -> { - if (t != null) { - log.error("Error closing topic list watcher {}", c, t); - } - return null; - }))); // Need to run the shutdown sequence in a separate thread to prevent deadlocks // If there are consumers or producers that need to be shutdown we cannot use the same thread @@ -1080,10 +1073,6 @@ void cleanupConsumer(ConsumerBase consumer) { consumers.remove(consumer); } - void cleanupTopicListWatcher(TopicListWatcher watcher) { - topicListWatchers.remove(watcher); - } - @VisibleForTesting int producersCount() { return producers.size(); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java index 90d930708aa90..79acb17640965 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java @@ -91,7 +91,6 @@ public void connectionFailed(PulsarClientException exception) { log.info("[{}] Watcher creation failed for {} with non-retriable error {}", topic, name, exception); deregisterFromClientCnx(); - client.cleanupTopicListWatcher(this); } } else { previousExceptions.add(exception); @@ -105,7 +104,6 @@ public void connectionOpened(ClientCnx cnx) { if (getState() == State.Closing || getState() == State.Closed) { setState(State.Closed); deregisterFromClientCnx(); - client.cleanupTopicListWatcher(this); return; } @@ -133,8 +131,6 @@ public void connectionOpened(ClientCnx cnx) { // drops the watcher on its side setState(State.Closed); deregisterFromClientCnx(); - client.cleanupTopicListWatcher(this); - cnx.channel().close(); return; } @@ -166,8 +162,6 @@ public void connectionOpened(ClientCnx cnx) { watcherFuture.completeExceptionally( PulsarClientException.wrap(e, String.format("Failed to create topic list watcher %s" + "when connecting to the broker", getHandlerName()))); - client.cleanupTopicListWatcher(this); - } else { // watcher was subscribed and connected, but we got some error, keep trying reconnectLater(e.getCause()); @@ -203,8 +197,6 @@ public CompletableFuture closeAsync() { log.info("[{}] [{}] Closed watcher (not connected)", topic, getHandlerName()); setState(State.Closed); deregisterFromClientCnx(); - client.cleanupTopicListWatcher(this); - closeFuture.complete(null); return closeFuture; } @@ -266,7 +258,6 @@ private void cleanupAtClose(CompletableFuture closeFuture, Throwable excep log.info("[{}] Closed topic list watcher", getHandlerName()); setState(State.Closed); deregisterFromClientCnx(); - client.cleanupTopicListWatcher(this); if (exception != null) { closeFuture.completeExceptionally(exception); } else { diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java index 07c8136f1cd48..c6eba43fb7a16 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java @@ -208,6 +208,7 @@ public void testNoWatchersWhenNoServerSupport() { cnx.newWatchTopicList(Commands.newWatchTopicList(7, 5, "tenant/ns", ".*", null), 7); assertTrue(result.isCompletedExceptionally()); + assertFalse(cnx.getTopicListWatchers().containsKey(5)); }); } From a177adeba83c441e26b740a3dc7d68f8c385e555 Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Fri, 24 Jun 2022 17:32:08 +0200 Subject: [PATCH 5/7] Rename command Unwatch, extract fields from command to avoid concurrent modification --- .../pulsar/broker/service/ServerCnx.java | 15 ++++-- .../broker/service/TopicListService.java | 25 ++++------ .../broker/service/TopicListServiceTest.java | 50 ++++++++++--------- .../pulsar/client/impl/TopicListWatcher.java | 2 +- .../pulsar/common/protocol/Commands.java | 6 +-- .../pulsar/common/protocol/PulsarDecoder.java | 10 ++-- pulsar-common/src/main/proto/PulsarApi.proto | 6 +-- 7 files changed, 59 insertions(+), 55 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 42eb46c69408d..5405291b72a01 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -50,6 +50,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; import java.util.stream.Collectors; import javax.naming.AuthenticationException; import javax.net.ssl.SSLSession; @@ -119,8 +120,8 @@ import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.CommandTcClientConnectRequest; import org.apache.pulsar.common.api.proto.CommandUnsubscribe; -import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; import org.apache.pulsar.common.api.proto.CommandWatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListClose; import org.apache.pulsar.common.api.proto.FeatureFlags; import org.apache.pulsar.common.api.proto.KeySharedMeta; import org.apache.pulsar.common.api.proto.KeySharedMode; @@ -2524,6 +2525,11 @@ protected void handleCommandWatchTopicList(CommandWatchTopicList commandWatchTop final long watcherId = commandWatchTopicList.getWatcherId(); final NamespaceName namespaceName = NamespaceName.get(commandWatchTopicList.getNamespace()); + Pattern topicsPattern = Pattern.compile(commandWatchTopicList.hasTopicsPattern() + ? commandWatchTopicList.getTopicsPattern() : TopicList.ALL_TOPICS_PATTERN); + String topicsHash = commandWatchTopicList.hasTopicsHash() + ? commandWatchTopicList.getTopicsHash() : null; + final Semaphore lookupSemaphore = service.getLookupRequestSemaphore(); if (lookupSemaphore.tryAcquire()) { if (invalidOriginalPrincipal(originalPrincipal)) { @@ -2536,7 +2542,8 @@ protected void handleCommandWatchTopicList(CommandWatchTopicList commandWatchTop } isNamespaceOperationAllowed(namespaceName, NamespaceOperation.GET_TOPICS).thenApply(isAuthorized -> { if (isAuthorized) { - topicListService.handleWatchTopicList(commandWatchTopicList, lookupSemaphore); + topicListService.handleWatchTopicList(namespaceName, watcherId, requestId, topicsPattern, + topicsHash, lookupSemaphore); } else { final String msg = "Proxy Client is not authorized to watchTopicList"; log.warn("[{}] {} with role {} on namespace {}", remoteAddress, msg, getPrincipal(), namespaceName); @@ -2562,8 +2569,8 @@ protected void handleCommandWatchTopicList(CommandWatchTopicList commandWatchTop } } - protected void handleCommandUnwatchTopicList(CommandUnwatchTopicList commandUnwatchTopicList) { - topicListService.handleUnwatchTopicList(commandUnwatchTopicList); + protected void handleCommandWatchTopicListClose(CommandWatchTopicListClose commandWatchTopicListClose) { + topicListService.handleWatchTopicListClose(commandWatchTopicListClose); } @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java index 4764ec87b0c29..78ee45223ebc0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java @@ -28,8 +28,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.resources.TopicResources; -import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; -import org.apache.pulsar.common.api.proto.CommandWatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListClose; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.topics.TopicList; @@ -110,14 +109,9 @@ public void inactivate() { } } - public void handleWatchTopicList(CommandWatchTopicList commandWatchTopicList, Semaphore lookupSemaphore) { - String namespace = commandWatchTopicList.getNamespace(); - NamespaceName namespaceName = NamespaceName.get(namespace); - long watcherId = commandWatchTopicList.getWatcherId(); - long requestId = commandWatchTopicList.getRequestId(); + public void handleWatchTopicList(NamespaceName namespaceName, long watcherId, long requestId, Pattern topicsPattern, + String topicsHash, Semaphore lookupSemaphore) { - Pattern topicsPattern = Pattern.compile(commandWatchTopicList.hasTopicsPattern() - ? commandWatchTopicList.getTopicsPattern() : TopicList.ALL_TOPICS_PATTERN); if (!enableSubscriptionPatternEvaluation || topicsPattern.pattern().length() > maxSubscriptionPatternLength) { String msg = "Unable to create topic list watcher: "; if (!enableSubscriptionPatternEvaluation) { @@ -167,21 +161,20 @@ public void handleWatchTopicList(CommandWatchTopicList commandWatchTopicList, Se finalWatcherFuture.thenAccept(watcher -> { List topicList = watcher.getMatchingTopics(); String hash = TopicList.calculateHash(topicList); - if (commandWatchTopicList.hasTopicsHash() - && hash.equals(commandWatchTopicList.getTopicsHash())) { + if (hash.equals(topicsHash)) { topicList = Collections.emptyList(); } if (log.isDebugEnabled()) { log.debug( "[{}] Received WatchTopicList for namespace [//{}] by {}", - connection.getRemoteAddress(), namespace, requestId); + connection.getRemoteAddress(), namespaceName, requestId); } connection.getCommandSender().sendWatchTopicListSuccess(requestId, watcherId, hash, topicList); lookupSemaphore.release(); }) .exceptionally(ex -> { log.warn("[{}] Error WatchTopicList for namespace [//{}] by {}", - connection.getRemoteAddress(), namespace, requestId); + connection.getRemoteAddress(), namespaceName, requestId); connection.getCommandSender().sendErrorResponse(requestId, BrokerServiceException.getClientErrorCode( new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage()); @@ -210,9 +203,9 @@ public void initializeTopicsListWatcher(CompletableFuture watc } - public void handleUnwatchTopicList(CommandUnwatchTopicList commandUnwatchTopicList) { - long requestId = commandUnwatchTopicList.getRequestId(); - long watcherId = commandUnwatchTopicList.getWatcherId(); + public void handleWatchTopicListClose(CommandWatchTopicListClose commandWatchTopicListClose) { + long requestId = commandWatchTopicListClose.getRequestId(); + long watcherId = commandWatchTopicListClose.getWatcherId(); deleteTopicListWatcher(watcherId); connection.getCommandSender().sendWatchTopicListSuccess(requestId, watcherId, null, null); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java index 494bb734bd482..924019f7a546d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicListServiceTest.java @@ -23,8 +23,7 @@ import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.resources.PulsarResources; import org.apache.pulsar.broker.resources.TopicResources; -import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; -import org.apache.pulsar.common.api.proto.CommandWatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListClose; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.topics.TopicList; @@ -44,6 +43,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Semaphore; +import java.util.regex.Pattern; public class TopicListServiceTest { @@ -77,12 +77,14 @@ public void setup() throws Exception { @Test public void testCommandWatchSuccessResponse() { - CommandWatchTopicList command = new CommandWatchTopicList() - .setRequestId(7) - .setWatcherId(13) - .setNamespace("tenant/ns") - .setTopicsPattern("persistent://tenant/ns/topic\\d"); - topicListService.handleWatchTopicList(command, lookupSemaphore); + + topicListService.handleWatchTopicList( + NamespaceName.get("tenant/ns"), + 13, + 7, + Pattern.compile("persistent://tenant/ns/topic\\d"), + null, + lookupSemaphore); List topics = Collections.singletonList("persistent://tenant/ns/topic1"); String hash = TopicList.calculateHash(topics); topicListFuture.complete(topics); @@ -94,12 +96,13 @@ public void testCommandWatchSuccessResponse() { @Test public void testCommandWatchErrorResponse() { - CommandWatchTopicList command = new CommandWatchTopicList() - .setRequestId(7) - .setWatcherId(13) - .setNamespace("tenant/ns") - .setTopicsPattern("persistent://tenant/ns/topic\\d"); - topicListService.handleWatchTopicList(command, lookupSemaphore); + topicListService.handleWatchTopicList( + NamespaceName.get("tenant/ns"), + 13, + 7, + Pattern.compile("persistent://tenant/ns/topic\\d"), + null, + lookupSemaphore); topicListFuture.completeExceptionally(new PulsarServerException("Error")); Assert.assertEquals(1, lookupSemaphore.availablePermits()); verifyNoInteractions(topicResources); @@ -108,20 +111,21 @@ public void testCommandWatchErrorResponse() { } @Test - public void testCommandUnwatchRemovesListener() { - CommandWatchTopicList commandWatch = new CommandWatchTopicList() - .setRequestId(7) - .setWatcherId(13) - .setNamespace("tenant/ns") - .setTopicsPattern("persistent://tenant/ns/topic\\d"); - topicListService.handleWatchTopicList(commandWatch, lookupSemaphore); + public void testCommandWatchTopicListCloseRemovesListener() { + topicListService.handleWatchTopicList( + NamespaceName.get("tenant/ns"), + 13, + 7, + Pattern.compile("persistent://tenant/ns/topic\\d"), + null, + lookupSemaphore); List topics = Collections.singletonList("persistent://tenant/ns/topic1"); topicListFuture.complete(topics); - CommandUnwatchTopicList commandUnwatch = new CommandUnwatchTopicList() + CommandWatchTopicListClose watchTopicListClose = new CommandWatchTopicListClose() .setRequestId(8) .setWatcherId(13); - topicListService.handleUnwatchTopicList(commandUnwatch); + topicListService.handleWatchTopicListClose(watchTopicListClose); verify(topicResources).deregisterPersistentTopicListener(any(TopicListService.TopicListWatcher.class)); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java index 79acb17640965..9cd6b003d7da1 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java @@ -210,7 +210,7 @@ public CompletableFuture closeAsync() { if (null == cnx) { cleanupAtClose(closeFuture, null); } else { - BaseCommand cmd = Commands.newUnwatchTopicList(watcherId, requestId); + BaseCommand cmd = Commands.newWatchTopicListClose(watcherId, requestId); cnx.sendRequestWithId(Commands.serializeWithSize(cmd), requestId).handle((v, exception) -> { final ChannelHandlerContext ctx = cnx.ctx(); boolean ignoreException = ctx == null || !ctx.channel().isActive(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index 4da97eff815d9..d815157935342 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -1518,9 +1518,9 @@ public static BaseCommand newWatchTopicUpdate(long watcherId, return cmd; } - public static BaseCommand newUnwatchTopicList(long watcherId, long requestId) { - BaseCommand cmd = localCmd(Type.UNWATCH_TOPIC_LIST); - cmd.setUnwatchTopicList() + public static BaseCommand newWatchTopicListClose(long watcherId, long requestId) { + BaseCommand cmd = localCmd(Type.WATCH_TOPIC_LIST_CLOSE); + cmd.setWatchTopicListClose() .setRequestId(requestId) .setWatcherId(watcherId); return cmd; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java index 9b0a81a7db112..d0b261034610a 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java @@ -77,8 +77,8 @@ import org.apache.pulsar.common.api.proto.CommandTcClientConnectRequest; import org.apache.pulsar.common.api.proto.CommandTcClientConnectResponse; import org.apache.pulsar.common.api.proto.CommandUnsubscribe; -import org.apache.pulsar.common.api.proto.CommandUnwatchTopicList; import org.apache.pulsar.common.api.proto.CommandWatchTopicList; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListClose; import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; import org.apache.pulsar.common.api.proto.ServerError; @@ -452,9 +452,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception handleCommandWatchTopicUpdate(cmd.getWatchTopicUpdate()); break; - case UNWATCH_TOPIC_LIST: - checkArgument(cmd.hasUnwatchTopicList()); - handleCommandUnwatchTopicList(cmd.getUnwatchTopicList()); + case WATCH_TOPIC_LIST_CLOSE: + checkArgument(cmd.hasWatchTopicListClose()); + handleCommandWatchTopicListClose(cmd.getWatchTopicListClose()); break; default: @@ -710,7 +710,7 @@ protected void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate commandWatc throw new UnsupportedOperationException(); } - protected void handleCommandUnwatchTopicList(CommandUnwatchTopicList commandUnwatchTopicList) { + protected void handleCommandWatchTopicListClose(CommandWatchTopicListClose commandWatchTopicListClose) { throw new UnsupportedOperationException(); } diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index e94563d700ad2..03cc6cc54daa4 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -782,7 +782,7 @@ message CommandWatchTopicUpdate { required string topics_hash = 4; } -message CommandUnwatchTopicList { +message CommandWatchTopicListClose { required uint64 request_id = 1; required uint64 watcher_id = 2; } @@ -1020,7 +1020,7 @@ message BaseCommand { WATCH_TOPIC_LIST = 64; WATCH_TOPIC_LIST_SUCCESS = 65; WATCH_TOPIC_UPDATE = 66; - UNWATCH_TOPIC_LIST = 67; + WATCH_TOPIC_LIST_CLOSE = 67; } @@ -1102,5 +1102,5 @@ message BaseCommand { optional CommandWatchTopicList watchTopicList = 64; optional CommandWatchTopicListSuccess watchTopicListSuccess = 65; optional CommandWatchTopicUpdate watchTopicUpdate = 66; - optional CommandUnwatchTopicList unwatchTopicList = 67; + optional CommandWatchTopicListClose watchTopicListClose = 67; } From da7d5c4237b0e79679eb93deaf07068056657e71 Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Fri, 24 Jun 2022 17:39:35 +0200 Subject: [PATCH 6/7] Fix cpp client compile error --- pulsar-client-cpp/lib/Commands.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-client-cpp/lib/Commands.cc b/pulsar-client-cpp/lib/Commands.cc index 041cad4c4eccc..6ef8c46338ab6 100644 --- a/pulsar-client-cpp/lib/Commands.cc +++ b/pulsar-client-cpp/lib/Commands.cc @@ -679,7 +679,7 @@ std::string Commands::messageType(BaseCommand_Type type) { return "WATCH_TOPIC_UPDATE"; break; case BaseCommand::UNWATCH_TOPIC_LIST: - return "UNWATCH_TOPIC_LIST"; + return "WATCH_TOPIC_LIST_CLOSE"; break; }; BOOST_THROW_EXCEPTION(std::logic_error("Invalid BaseCommand enumeration value")); From c71bb64b9666c38688ebc115501ea0ebdd8f309f Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Fri, 24 Jun 2022 17:46:42 +0200 Subject: [PATCH 7/7] Fix cpp client compile error --- pulsar-client-cpp/lib/Commands.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-client-cpp/lib/Commands.cc b/pulsar-client-cpp/lib/Commands.cc index 6ef8c46338ab6..3a00d302e8679 100644 --- a/pulsar-client-cpp/lib/Commands.cc +++ b/pulsar-client-cpp/lib/Commands.cc @@ -678,7 +678,7 @@ std::string Commands::messageType(BaseCommand_Type type) { case BaseCommand::WATCH_TOPIC_UPDATE: return "WATCH_TOPIC_UPDATE"; break; - case BaseCommand::UNWATCH_TOPIC_LIST: + case BaseCommand::WATCH_TOPIC_LIST_CLOSE: return "WATCH_TOPIC_LIST_CLOSE"; break; };