diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java index 310354dcd3b47..5e74158c9c297 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.NavigableMap; import java.util.Objects; +import java.util.Optional; import java.util.TreeMap; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArrayList; @@ -32,6 +33,7 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException; import org.apache.pulsar.client.impl.Murmur3Hash32; @@ -80,8 +82,6 @@ public AbstractDispatcherSingleActiveConsumer(SubType subscriptionType, int part protected abstract void scheduleReadOnActiveConsumer(); - protected abstract void readMoreEntries(Consumer consumer); - protected abstract void cancelPendingRead(); protected void notifyActiveConsumerChanged(Consumer activeConsumer) { @@ -257,9 +257,12 @@ public synchronized boolean canUnsubscribe(Consumer consumer) { return (consumers.size() == 1) && Objects.equals(consumer, ACTIVE_CONSUMER_UPDATER.get(this)); } - public CompletableFuture close() { + @Override + public CompletableFuture close(boolean disconnectConsumers, + Optional assignedBrokerLookupData) { IS_CLOSED_UPDATER.set(this, TRUE); - return disconnectAllConsumers(); + return disconnectConsumers + ? disconnectAllConsumers(false, assignedBrokerLookupData) : CompletableFuture.completedFuture(null); } public boolean isClosed() { @@ -268,15 +271,23 @@ public boolean isClosed() { /** * Disconnect all consumers on this dispatcher (server side close). This triggers channelInactive on the inbound - * handler which calls dispatcher.removeConsumer(), where the closeFuture is completed + * handler which calls dispatcher.removeConsumer(), where the closeFuture is completed. * - * @return + * @param isResetCursor + * Specifies if the cursor has been reset. + * @param assignedBrokerLookupData + * Optional target broker redirect information. Allows the consumer to quickly reconnect to a broker + * during bundle unloading. + * + * @return CompletableFuture indicating the completion of the operation. */ - public synchronized CompletableFuture disconnectAllConsumers(boolean isResetCursor) { + @Override + public synchronized CompletableFuture disconnectAllConsumers( + boolean isResetCursor, Optional assignedBrokerLookupData) { closeFuture = new CompletableFuture<>(); if (!consumers.isEmpty()) { - consumers.forEach(consumer -> consumer.disconnect(isResetCursor)); + consumers.forEach(consumer -> consumer.disconnect(isResetCursor, assignedBrokerLookupData)); cancelPendingRead(); } else { // no consumer connected, complete disconnect immediately diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index ee4fcff3ad1aa..5ec76d07feb42 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -44,6 +44,7 @@ import org.apache.commons.lang3.mutable.MutableInt; import org.apache.commons.lang3.tuple.MutablePair; import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.client.api.MessageId; @@ -407,8 +408,12 @@ public void disconnect() { } public void disconnect(boolean isResetCursor) { + disconnect(isResetCursor, Optional.empty()); + } + + public void disconnect(boolean isResetCursor, Optional assignedBrokerLookupData) { log.info("Disconnecting consumer: {}", this); - cnx.closeConsumer(this); + cnx.closeConsumer(this, assignedBrokerLookupData); try { close(isResetCursor); } catch (BrokerServiceException e) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java index 3ca06dc83d9aa..bdea106171b82 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java @@ -22,6 +22,7 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.MessageMetadata; @@ -49,7 +50,11 @@ public interface Dispatcher { * * @return */ - CompletableFuture close(); + default CompletableFuture close() { + return close(true, Optional.empty()); + } + + CompletableFuture close(boolean disconnectClients, Optional assignedBrokerLookupData); boolean isClosed(); @@ -63,12 +68,17 @@ public interface Dispatcher { * * @return */ - CompletableFuture disconnectAllConsumers(boolean isResetCursor); + default CompletableFuture disconnectAllConsumers(boolean isResetCursor) { + return disconnectAllConsumers(isResetCursor, Optional.empty()); + } default CompletableFuture disconnectAllConsumers() { return disconnectAllConsumers(false); } + CompletableFuture disconnectAllConsumers(boolean isResetCursor, + Optional assignedBrokerLookupData); + void resetCloseFuture(); /** diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 5b6a723a250f7..b0b4fe98b0c5f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -497,6 +497,7 @@ public void completed(Exception exception, long ledgerId, long entryId) { } else if (!(exception instanceof TopicClosedException)) { // For TopicClosed exception there's no need to send explicit error, since the client was // already notified + // For TopicClosingOrDeleting exception, a notification will be sent separately long callBackSequenceId = Math.max(highestSequenceId, sequenceId); producer.cnx.getCommandSender().sendSendError(producer.producerId, callBackSequenceId, serverError, exception.getMessage()); @@ -718,7 +719,7 @@ public CompletableFuture disconnect() { */ public CompletableFuture disconnect(Optional assignedBrokerLookupData) { if (!closeFuture.isDone() && isDisconnecting.compareAndSet(false, true)) { - log.info("Disconnecting producer: {}", this); + log.info("Disconnecting producer: {}, assignedBrokerLookupData: {}", this, assignedBrokerLookupData); cnx.execute(() -> { cnx.closeProducer(this, assignedBrokerLookupData); closeNow(true); 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 b58b74b3ea225..ab48f52dee7f2 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 @@ -28,6 +28,7 @@ import static org.apache.pulsar.broker.service.persistent.PersistentTopic.getMigratedClusterUrl; import static org.apache.pulsar.common.api.proto.ProtocolVersion.v5; import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH; +import static org.apache.pulsar.common.protocol.Commands.newCloseConsumer; import static org.apache.pulsar.common.protocol.Commands.newLookupErrorResponse; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Strings; @@ -1321,7 +1322,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { topicName, remoteAddress, consumerId); } consumers.remove(consumerId, consumerFuture); - closeConsumer(consumerId); + closeConsumer(consumerId, Optional.empty()); return null; } } else if (exception.getCause() instanceof BrokerServiceException) { @@ -1766,7 +1767,7 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) { // if the topic is transferring, we ignore send msg. if (producer.getTopic().isTransferring()) { long ignoredMsgCount = ExtensibleLoadManagerImpl.get(pulsar) - .getIgnoredSendMsgCounter().incrementAndGet(); + .getIgnoredSendMsgCounter().addAndGet(send.getNumMessages()); if (log.isDebugEnabled()) { log.debug("Ignored send msg from:{}:{} to fenced topic:{} while transferring." + " Ignored message count:{}.", @@ -1842,6 +1843,15 @@ protected void handleAck(CommandAck ack) { if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) { Consumer consumer = consumerFuture.getNow(null); + Subscription subscription = consumer.getSubscription(); + if (subscription.getTopic().isTransferring()) { + // Message acks are silently ignored during topic transfer. + if (log.isDebugEnabled()) { + log.debug("[{}] [{}] Ignoring message acknowledgment during topic transfer, ack count: {}", + subscription, consumerId, ack.getMessageIdsCount()); + } + return; + } consumer.messageAcked(ack).thenRun(() -> { if (hasRequestId) { writeAndFlush(Commands.newAckResponse( @@ -3071,15 +3081,17 @@ private void closeProducer(long producerId, long epoch, Optional assignedBrokerLookupData) { // removes consumer-connection from map and send close command to consumer safelyRemoveConsumer(consumer); - closeConsumer(consumer.consumerId()); + closeConsumer(consumer.consumerId(), assignedBrokerLookupData); } - private void closeConsumer(long consumerId) { + private void closeConsumer(long consumerId, Optional assignedBrokerLookupData) { if (getRemoteEndpointProtocolVersion() >= v5.getValue()) { - writeAndFlush(Commands.newCloseConsumer(consumerId, -1L)); + writeAndFlush(newCloseConsumer(consumerId, -1L, + assignedBrokerLookupData.map(BrokerLookupData::pulsarServiceUrl).orElse(null), + assignedBrokerLookupData.map(BrokerLookupData::pulsarServiceUrlTls).orElse(null))); } else { close(); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java index 9deeafdb272f5..6805d19752126 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java @@ -26,6 +26,7 @@ import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.common.api.proto.CommandAck.AckType; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.ReplicatedSubscriptionsSnapshot; @@ -64,13 +65,13 @@ default long getNumberOfEntriesDelayed() { List getConsumers(); - CompletableFuture close(); - CompletableFuture delete(); CompletableFuture deleteForcefully(); - CompletableFuture disconnect(); + CompletableFuture disconnect(Optional assignedBrokerLookupData); + + CompletableFuture close(boolean disconnectConsumers, Optional assignedBrokerLookupData); CompletableFuture doUnsubscribe(Consumer consumer); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java index c09d63a9232eb..a644bb70de8b0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java @@ -73,7 +73,7 @@ public interface TransportCnx { void removedConsumer(Consumer consumer); - void closeConsumer(Consumer consumer); + void closeConsumer(Consumer consumer, Optional assignedBrokerLookupData); boolean isPreciseDispatcherFlowControl(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java index c106b1603f6bd..0a9ec9fbb57f9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java @@ -19,10 +19,12 @@ package org.apache.pulsar.broker.service.nonpersistent; import java.util.List; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import org.apache.bookkeeper.mledger.Entry; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; @@ -126,9 +128,11 @@ public synchronized boolean canUnsubscribe(Consumer consumer) { } @Override - public CompletableFuture close() { + public CompletableFuture close(boolean disconnectConsumers, + Optional assignedBrokerLookupData) { IS_CLOSED_UPDATER.set(this, TRUE); - return disconnectAllConsumers(); + return disconnectConsumers + ? disconnectAllConsumers(false, assignedBrokerLookupData) : CompletableFuture.completedFuture(null); } @Override @@ -147,12 +151,13 @@ public synchronized void consumerFlow(Consumer consumer, int additionalNumberOfM } @Override - public synchronized CompletableFuture disconnectAllConsumers(boolean isResetCursor) { + public synchronized CompletableFuture disconnectAllConsumers( + boolean isResetCursor, Optional assignedBrokerLookupData) { closeFuture = new CompletableFuture<>(); if (consumerList.isEmpty()) { closeFuture.complete(null); } else { - consumerList.forEach(Consumer::disconnect); + consumerList.forEach(consumer -> consumer.disconnect(isResetCursor, assignedBrokerLookupData)); } return closeFuture; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java index 25e3e2894daa1..5e8eda2ab70e6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java @@ -101,11 +101,6 @@ protected void scheduleReadOnActiveConsumer() { // No-op } - @Override - protected void readMoreEntries(Consumer consumer) { - // No-op - } - @Override protected void cancelPendingRead() { // No-op diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java index 6ec969c927a8c..28ea9f39ac86e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java @@ -30,6 +30,7 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.commons.collections4.CollectionUtils; import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.AbstractSubscription; import org.apache.pulsar.broker.service.AnalyzeBacklogResult; import org.apache.pulsar.broker.service.BrokerServiceException; @@ -277,40 +278,67 @@ public boolean isSubscriptionMigrated() { return topic.isMigrated(); } + /** + * Disconnect all consumers from this subscription. + * + * @return CompletableFuture indicating the completion of the operation. + */ @Override - public CompletableFuture close() { + public synchronized CompletableFuture disconnect(Optional assignedBrokerLookupData) { + CompletableFuture closeFuture = new CompletableFuture<>(); + + (dispatcher != null + ? dispatcher.disconnectAllConsumers(false, assignedBrokerLookupData) + : CompletableFuture.completedFuture(null)) + .thenRun(() -> { + log.info("[{}][{}] Successfully disconnected subscription consumers", topicName, subName); + closeFuture.complete(null); + }).exceptionally(exception -> { + log.error("[{}][{}] Error disconnecting subscription consumers", topicName, subName, exception); + closeFuture.completeExceptionally(exception); + return null; + }); + + return closeFuture; + + } + + private CompletableFuture fence() { IS_FENCED_UPDATER.set(this, TRUE); return CompletableFuture.completedFuture(null); } + /** - * Disconnect all consumers attached to the dispatcher and close this subscription. + * Fence this subscription and optionally disconnect all consumers. * - * @return CompletableFuture indicating the completion of disconnect operation + * @return CompletableFuture indicating the completion of the operation. */ @Override - public synchronized CompletableFuture disconnect() { - CompletableFuture disconnectFuture = new CompletableFuture<>(); + public synchronized CompletableFuture close(boolean disconnectConsumers, + Optional assignedBrokerLookupData) { + CompletableFuture closeFuture = new CompletableFuture<>(); // block any further consumers on this subscription IS_FENCED_UPDATER.set(this, TRUE); - (dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null)).thenCompose(v -> close()) + (dispatcher != null + ? dispatcher.close(disconnectConsumers, assignedBrokerLookupData) + : CompletableFuture.completedFuture(null)) .thenRun(() -> { - log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName); - disconnectFuture.complete(null); + log.info("[{}][{}] Successfully closed subscription", topicName, subName); + closeFuture.complete(null); }).exceptionally(exception -> { IS_FENCED_UPDATER.set(this, FALSE); if (dispatcher != null) { dispatcher.reset(); } - log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName, - exception); - disconnectFuture.completeExceptionally(exception); + log.error("[{}][{}] Error closing subscription", topicName, subName, exception); + closeFuture.completeExceptionally(exception); return null; }); - return disconnectFuture; + return closeFuture; } /** @@ -349,7 +377,7 @@ private CompletableFuture delete(boolean closeIfConsumersConnected) { CompletableFuture closeSubscriptionFuture = new CompletableFuture<>(); if (closeIfConsumersConnected) { - this.disconnect().thenRun(() -> { + this.close(true, Optional.empty()).thenRun(() -> { closeSubscriptionFuture.complete(null); }).exceptionally(ex -> { log.error("[{}][{}] Error disconnecting and closing subscription", topicName, subName, ex); @@ -357,7 +385,7 @@ private CompletableFuture delete(boolean closeIfConsumersConnected) { return null; }); } else { - this.close().thenRun(() -> { + this.fence().thenRun(() -> { closeSubscriptionFuture.complete(null); }).exceptionally(exception -> { log.error("[{}][{}] Error closing subscription", topicName, subName, exception); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 67a338d08bbe3..c8f2d4ce62e42 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -422,7 +422,7 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, boolean c List> futures = new ArrayList<>(); replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect())); producers.values().forEach(producer -> futures.add(producer.disconnect())); - subscriptions.forEach((s, sub) -> futures.add(sub.disconnect())); + subscriptions.forEach((s, sub) -> futures.add(sub.close(true, Optional.empty()))); FutureUtil.waitForAll(futures).thenRun(() -> { closeClientFuture.complete(null); }).exceptionally(ex -> { @@ -526,14 +526,25 @@ public CompletableFuture close( replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect())); if (disconnectClients) { futures.add(ExtensibleLoadManagerImpl.getAssignedBrokerLookupData( - brokerService.getPulsar(), topic).thenAccept(lookupData -> - producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData))) + brokerService.getPulsar(), topic).thenAccept(lookupData -> { + producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData))); + // Topics unloaded due to the ExtensibleLoadManager undergo closing twice: first with + // disconnectClients = false, second with disconnectClients = true. The check below identifies the + // cases when Topic.close is called outside the scope of the ExtensibleLoadManager. In these + // situations, we must pursue the regular Subscription.close, as Topic.close is invoked just once. + if (isTransferring()) { + subscriptions.forEach((s, sub) -> futures.add(sub.disconnect(lookupData))); + } else { + subscriptions.forEach((s, sub) -> futures.add(sub.close(true, lookupData))); + } + } )); + } else { + subscriptions.forEach((s, sub) -> futures.add(sub.close(false, Optional.empty()))); } if (topicPublishRateLimiter != null) { topicPublishRateLimiter.close(); } - subscriptions.forEach((s, sub) -> futures.add(sub.disconnect())); if (this.resourceGroupPublishLimiter != null) { this.resourceGroupPublishLimiter.unregisterRateLimitFunction(this.getName()); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index b3d48252efe58..30643b7058e83 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java @@ -51,6 +51,7 @@ import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerFactory; import org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker; import org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; @@ -275,6 +276,10 @@ public synchronized void readMoreEntries() { if (shouldPauseDeliveryForDelayTracker()) { return; } + if (topic.isTransferring()) { + // Do not deliver messages for topics that are undergoing transfer, as the acknowledgments would be ignored. + return; + } // totalAvailablePermits may be updated by other threads int firstAvailableConsumerPermits = getFirstAvailableConsumerPermits(); @@ -484,7 +489,8 @@ public synchronized boolean canUnsubscribe(Consumer consumer) { } @Override - public CompletableFuture close() { + public CompletableFuture close(boolean disconnectConsumers, + Optional assignedBrokerLookupData) { IS_CLOSED_UPDATER.set(this, TRUE); Optional delayedDeliveryTracker; @@ -494,19 +500,20 @@ public CompletableFuture close() { } delayedDeliveryTracker.ifPresent(DelayedDeliveryTracker::close); - dispatchRateLimiter.ifPresent(DispatchRateLimiter::close); - return disconnectAllConsumers(); + return disconnectConsumers + ? disconnectAllConsumers(false, assignedBrokerLookupData) : CompletableFuture.completedFuture(null); } @Override - public synchronized CompletableFuture disconnectAllConsumers(boolean isResetCursor) { + public synchronized CompletableFuture disconnectAllConsumers( + boolean isResetCursor, Optional assignedBrokerLookupData) { closeFuture = new CompletableFuture<>(); if (consumerList.isEmpty()) { closeFuture.complete(null); } else { - consumerList.forEach(consumer -> consumer.disconnect(isResetCursor)); + consumerList.forEach(consumer -> consumer.disconnect(isResetCursor, assignedBrokerLookupData)); cancelPendingRead(); } return closeFuture; @@ -665,15 +672,9 @@ protected synchronized boolean trySendMessagesToConsumers(ReadType readType, Lis long totalEntries = 0; int avgBatchSizePerMsg = remainingMessages > 0 ? Math.max(remainingMessages / entries.size(), 1) : 1; - int firstAvailableConsumerPermits, currentTotalAvailablePermits; - boolean dispatchMessage; - while (entriesToDispatch > 0) { - firstAvailableConsumerPermits = getFirstAvailableConsumerPermits(); - currentTotalAvailablePermits = Math.max(totalAvailablePermits, firstAvailableConsumerPermits); - dispatchMessage = currentTotalAvailablePermits > 0 && firstAvailableConsumerPermits > 0; - if (!dispatchMessage) { - break; - } + // If the dispatcher is closed, firstAvailableConsumerPermits will be 0, which skips dispatching the + // messages. + while (entriesToDispatch > 0 && isAtleastOneConsumerAvailable()) { Consumer c = getNextConsumer(); if (c == null) { // Do nothing, cursor will be rewind at reconnection diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java index 5e9183df0b1df..aea9b1c9b9e16 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java @@ -154,7 +154,7 @@ public void readEntriesComplete(final List entries, Object obj) { executor.execute(() -> internalReadEntriesComplete(entries, obj)); } - public synchronized void internalReadEntriesComplete(final List entries, Object obj) { + private synchronized void internalReadEntriesComplete(final List entries, Object obj) { ReadEntriesCtx readEntriesCtx = (ReadEntriesCtx) obj; Consumer readConsumer = readEntriesCtx.getConsumer(); long epoch = readEntriesCtx.getEpoch(); @@ -194,11 +194,17 @@ public synchronized void internalReadEntriesComplete(final List entries, } } - if (currentConsumer == null || readConsumer != currentConsumer) { - // Active consumer has changed since the read request has been issued. We need to rewind the cursor and - // re-issue the read request for the new consumer + if (currentConsumer == null || readConsumer != currentConsumer || topic.isTransferring()) { + // Active consumer has changed since the read request has been issued, or the topic is being transferred to + // another broker. We need to rewind the cursor and re-issue the read request for the new consumer. if (log.isDebugEnabled()) { - log.debug("[{}] rewind because no available consumer found", name); + if (currentConsumer == null) { + log.debug("[{}] rewind because no available consumer found", name); + } else if (readConsumer != currentConsumer) { + log.debug("[{}] rewind because active consumer changed", name); + } else { + log.debug("[{}] rewind because topic is transferring", name); + } } entries.forEach(Entry::release); cursor.rewind(); @@ -309,8 +315,7 @@ public void redeliverUnacknowledgedMessages(Consumer consumer, List 0) { synchronized (this) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index 0397eef8aa86c..86e3558f550cd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -57,6 +57,7 @@ import org.apache.commons.lang3.tuple.MutablePair; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.AbstractSubscription; import org.apache.pulsar.broker.service.AnalyzeBacklogResult; import org.apache.pulsar.broker.service.BrokerServiceException; @@ -307,8 +308,9 @@ public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor topic.getManagedLedger().removeWaitingCursor(cursor); if (!cursor.isDurable()) { - // If cursor is not durable, we need to clean up the subscription as well - this.close().thenRun(() -> { + // If cursor is not durable, we need to clean up the subscription as well. No need to check for active + // consumers since we already validated that there are no consumers on this dispatcher. + this.closeCursor(false).thenRun(() -> { synchronized (this) { if (dispatcher != null) { dispatcher.close().thenRun(() -> { @@ -885,49 +887,77 @@ public int getTotalNonContiguousDeletedMessagesRange() { } /** - * Close the cursor ledger for this subscription. Requires that there are no active consumers on the dispatcher + * Close the cursor ledger for this subscription. Optionally verifies that there are no active consumers on the + * dispatcher. * - * @return CompletableFuture indicating the completion of delete operation + * @return CompletableFuture indicating the completion of close operation */ - @Override - public CompletableFuture close() { - synchronized (this) { - if (dispatcher != null && dispatcher.isConsumerConnected()) { - return FutureUtil.failedFuture(new SubscriptionBusyException("Subscription has active consumers")); - } - return this.pendingAckHandle.closeAsync().thenAccept(v -> { - IS_FENCED_UPDATER.set(this, TRUE); - log.info("[{}][{}] Successfully closed subscription [{}]", topicName, subName, cursor); - }); + private synchronized CompletableFuture closeCursor(boolean checkActiveConsumers) { + if (checkActiveConsumers && dispatcher != null && dispatcher.isConsumerConnected()) { + return FutureUtil.failedFuture(new SubscriptionBusyException("Subscription has active consumers")); } + return this.pendingAckHandle.closeAsync().thenAccept(v -> { + IS_FENCED_UPDATER.set(this, TRUE); + log.info("[{}][{}] Successfully closed subscription [{}]", topicName, subName, cursor); + }); } + /** - * Disconnect all consumers attached to the dispatcher and close this subscription. + * Disconnect all consumers from this subscription. * - * @return CompletableFuture indicating the completion of disconnect operation + * @return CompletableFuture indicating the completion of the operation. */ @Override - public synchronized CompletableFuture disconnect() { - if (fenceFuture != null){ + public synchronized CompletableFuture disconnect(Optional assignedBrokerLookupData) { + CompletableFuture disconnectFuture = new CompletableFuture<>(); + + (dispatcher != null + ? dispatcher.disconnectAllConsumers(false, assignedBrokerLookupData) + : CompletableFuture.completedFuture(null)) + .thenRun(() -> { + log.info("[{}][{}] Successfully disconnected subscription consumers", topicName, subName); + disconnectFuture.complete(null); + }).exceptionally(exception -> { + log.error("[{}][{}] Error disconnecting subscription consumers", topicName, subName, exception); + disconnectFuture.completeExceptionally(exception); + return null; + }); + + return disconnectFuture; + } + + /** + * Fence this subscription and optionally disconnect all consumers. + * + * @return CompletableFuture indicating the completion of the operation. + */ + @Override + public synchronized CompletableFuture close(boolean disconnectConsumers, + Optional assignedBrokerLookupData) { + if (fenceFuture != null) { return fenceFuture; } + fenceFuture = new CompletableFuture<>(); // block any further consumers on this subscription IS_FENCED_UPDATER.set(this, TRUE); - (dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null)) - .thenCompose(v -> close()).thenRun(() -> { - log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName); + (dispatcher != null + ? dispatcher.close(disconnectConsumers, assignedBrokerLookupData) + : CompletableFuture.completedFuture(null)) + // checkActiveConsumers is false since we just closed all of them if we wanted. + .thenCompose(__ -> closeCursor(false)).thenRun(() -> { + log.info("[{}][{}] Successfully closed the subscription", topicName, subName); fenceFuture.complete(null); }).exceptionally(exception -> { - log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName, - exception); + log.error("[{}][{}] Error closing the subscription", topicName, subName, exception); fenceFuture.completeExceptionally(exception); resumeAfterFence(); return null; }); + return fenceFuture; } @@ -935,7 +965,7 @@ public synchronized CompletableFuture disconnect() { * Resume subscription after topic deletion or close failure. */ public synchronized void resumeAfterFence() { - // If "fenceFuture" is null, it means that "disconnect" has never been called. + // If "fenceFuture" is null, it means that "close" has never been called. if (fenceFuture != null) { fenceFuture.whenComplete((ignore, ignoreEx) -> { synchronized (PersistentSubscription.this) { @@ -992,7 +1022,7 @@ private CompletableFuture delete(boolean closeIfConsumersConnected) { CompletableFuture closeSubscriptionFuture = new CompletableFuture<>(); if (closeIfConsumersConnected) { - this.disconnect().thenRun(() -> { + this.close(true, Optional.empty()).thenRun(() -> { closeSubscriptionFuture.complete(null); }).exceptionally(ex -> { log.error("[{}][{}] Error disconnecting and closing subscription", topicName, subName, ex); @@ -1000,7 +1030,7 @@ private CompletableFuture delete(boolean closeIfConsumersConnected) { return null; }); } else { - this.close().thenRun(() -> { + this.closeCursor(true).thenRun(() -> { closeSubscriptionFuture.complete(null); }).exceptionally(exception -> { log.error("[{}][{}] Error closing subscription", topicName, subName, exception); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index f22865ed550ee..4b506799e16bd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -477,7 +477,7 @@ public CompletableFuture unloadSubscription(@Nonnull String subName) { new UnsupportedSubscriptionException(String.format("Unsupported subscription: %s", subName))); } // Fence old subscription -> Rewind cursor -> Replace with a new subscription. - return sub.disconnect().thenCompose(ignore -> { + return sub.close(true, Optional.empty()).thenCompose(ignore -> { if (!lock.writeLock().tryLock()) { return CompletableFuture.failedFuture(new SubscriptionConflictUnloadException(String.format("Conflict" + " topic-close, topic-delete, another-subscribe-unload, cannot unload subscription %s now", @@ -529,8 +529,7 @@ public void publishMessage(ByteBuf headersAndPayload, PublishContext publishCont return; } if (isExceedMaximumMessageSize(headersAndPayload.readableBytes(), publishContext)) { - publishContext.completed(new NotAllowedException("Exceed maximum message size") - , -1, -1); + publishContext.completed(new NotAllowedException("Exceed maximum message size"), -1, -1); decrementPendingWriteOpsAndCheck(); return; } @@ -1361,7 +1360,7 @@ private CompletableFuture delete(boolean failIfHasSubscriptions, CompletableFuture closeClientFuture = new CompletableFuture<>(); List> futures = new ArrayList<>(); - subscriptions.forEach((s, sub) -> futures.add(sub.disconnect())); + subscriptions.forEach((s, sub) -> futures.add(sub.close(true, Optional.empty()))); if (closeIfClientsConnected) { replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect())); shadowReplicators.forEach((__, replicator) -> futures.add(replicator.disconnect())); @@ -1509,14 +1508,25 @@ public CompletableFuture close( shadowReplicators.forEach((__, replicator) -> futures.add(replicator.disconnect())); if (disconnectClients) { futures.add(ExtensibleLoadManagerImpl.getAssignedBrokerLookupData( - brokerService.getPulsar(), topic).thenAccept(lookupData -> - producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData))) + brokerService.getPulsar(), topic).thenAccept(lookupData -> { + producers.values().forEach(producer -> futures.add(producer.disconnect(lookupData))); + // Topics unloaded due to the ExtensibleLoadManager undergo closing twice: first with + // disconnectClients = false, second with disconnectClients = true. The check below identifies the + // cases when Topic.close is called outside the scope of the ExtensibleLoadManager. In these + // situations, we must pursue the regular Subscription.close, as Topic.close is invoked just once. + if (isTransferring()) { + subscriptions.forEach((s, sub) -> futures.add(sub.disconnect(lookupData))); + } else { + subscriptions.forEach((s, sub) -> futures.add(sub.close(true, lookupData))); + } + } )); + } else { + subscriptions.forEach((s, sub) -> futures.add(sub.close(false, Optional.empty()))); } if (topicPublishRateLimiter != null) { topicPublishRateLimiter.close(); } - subscriptions.forEach((s, sub) -> futures.add(sub.disconnect())); if (this.resourceGroupPublishLimiter != null) { this.resourceGroupPublishLimiter.unregisterRateLimitFunction(this.getName()); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java index 146d2a9b87839..d207ecd56ee7b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java @@ -41,6 +41,7 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -53,19 +54,27 @@ import static org.testng.Assert.fail; import com.google.common.collect.Sets; import java.net.URL; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; +import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.mutable.MutableInt; import org.apache.commons.lang3.reflect.FieldUtils; @@ -98,13 +107,20 @@ import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.LookupService; import org.apache.pulsar.client.impl.TableViewImpl; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.ServiceUnitId; import org.apache.pulsar.common.naming.SystemTopicNames; +import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.naming.TopicVersion; import org.apache.pulsar.common.policies.data.BrokerAssignment; @@ -410,111 +426,240 @@ public boolean test(NamespaceBundle namespaceBundle) { assertTrue(ex.getMessage().contains("cannot be transfer to same broker")); } } - @DataProvider(name = "isPersistentTopicTest") - public Object[][] isPersistentTopicTest() { - return new Object[][] { { true }, { false }}; + + @DataProvider(name = "isPersistentTopicSubscriptionTypeTest") + public Object[][] isPersistentTopicSubscriptionTypeTest() { + return new Object[][]{ + {TopicDomain.persistent, SubscriptionType.Exclusive}, + {TopicDomain.persistent, SubscriptionType.Shared}, + {TopicDomain.persistent, SubscriptionType.Failover}, + {TopicDomain.persistent, SubscriptionType.Key_Shared}, + {TopicDomain.non_persistent, SubscriptionType.Exclusive}, + {TopicDomain.non_persistent, SubscriptionType.Shared}, + {TopicDomain.non_persistent, SubscriptionType.Failover}, + {TopicDomain.non_persistent, SubscriptionType.Key_Shared}, + }; } - @Test(timeOut = 30 * 1000, dataProvider = "isPersistentTopicTest") - public void testTransferClientReconnectionWithoutLookup(boolean isPersistentTopicTest) throws Exception { - String topicType = isPersistentTopicTest? "persistent" : "non-persistent"; - String topic = topicType + "://" + defaultTestNamespace + "/test-transfer-client-reconnect"; - TopicName topicName = TopicName.get(topic); - AtomicInteger lookupCount = new AtomicInteger(); - var lookup = spyLookupService(lookupCount, topicName); - var producer = pulsarClient.newProducer().topic(topic).create(); - int lookupCountBeforeUnload = lookupCount.get(); + @Test(timeOut = 30_000, dataProvider = "isPersistentTopicSubscriptionTypeTest") + public void testTransferClientReconnectionWithoutLookup(TopicDomain topicDomain, SubscriptionType subscriptionType) + throws Exception { + var id = String.format("test-tx-client-reconnect-%s-%s", subscriptionType, UUID.randomUUID()); + var topic = String.format("%s://%s/%s", topicDomain.toString(), defaultTestNamespace, id); + var topicName = TopicName.get(topic); + var timeoutMs = 30_000; - NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get(); - String broker = admin.lookups().lookupTopic(topic); - String dstBrokerUrl = pulsar1.getLookupServiceAddress(); - String dstBrokerServiceUrl; - if (broker.equals(pulsar1.getBrokerServiceUrl())) { - dstBrokerUrl = pulsar2.getLookupServiceAddress(); - dstBrokerServiceUrl = pulsar2.getBrokerServiceUrl(); - } else { - dstBrokerServiceUrl = pulsar1.getBrokerServiceUrl(); - } - checkOwnershipState(broker, bundle); + var clients = new ArrayList(); + var consumers = new ArrayList>(); + try { + var lookups = new ArrayList(); + + @Cleanup + var producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); + lookups.add(spyLookupService(pulsarClient)); + + var consumerCount = subscriptionType == SubscriptionType.Exclusive ? 1 : 3; + + for (int i = 0; i < consumerCount; i++) { + var client = newPulsarClient(lookupUrl.toString(), 0); + clients.add(client); + var consumer = client.newConsumer(Schema.STRING). + subscriptionName(id). + subscriptionType(subscriptionType). + subscriptionInitialPosition(SubscriptionInitialPosition.Earliest). + ackTimeout(1000, TimeUnit.MILLISECONDS). + topic(topic). + subscribe(); + consumers.add(consumer); + lookups.add(spyLookupService(client)); + } - final String finalDstBrokerUrl = dstBrokerUrl; - CompletableFuture.runAsync(() -> { + Awaitility.await() + .until(() -> producer.isConnected() && consumers.stream().allMatch(Consumer::isConnected)); + + NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get(); + String broker = admin.lookups().lookupTopic(topic); + final String dstBrokerUrl; + final String dstBrokerServiceUrl; + if (broker.equals(pulsar1.getBrokerServiceUrl())) { + dstBrokerUrl = pulsar2.getLookupServiceAddress(); + dstBrokerServiceUrl = pulsar2.getBrokerServiceUrl(); + } else { + dstBrokerUrl = pulsar1.getLookupServiceAddress(); + dstBrokerServiceUrl = pulsar1.getBrokerServiceUrl(); + } + checkOwnershipState(broker, bundle); + + var messageCountBeforeUnloading = 100; + var messageCountAfterUnloading = 100; + var messageCount = messageCountBeforeUnloading + messageCountAfterUnloading; + + var semMessagesReadyToSend = new Semaphore(0); + var cdlStart = new CountDownLatch(1); + + @Cleanup(value = "shutdown") + var executor = Executors.newFixedThreadPool(1 /* bundle unload */ + 1 /* producer */ + consumers.size()); + + var futures = new ArrayList>(); + futures.add(CompletableFuture.runAsync(() -> { + try { + cdlStart.await(); + semMessagesReadyToSend.release(messageCountBeforeUnloading); + admin.namespaces() + .unloadNamespaceBundle(defaultTestNamespace, bundle.getBundleRange(), dstBrokerUrl); + semMessagesReadyToSend.release(messageCountAfterUnloading); + } catch (InterruptedException | PulsarAdminException e) { + fail(); + } + }, executor)); + + var pendingMessages = Collections.synchronizedSet(new HashSet<>(messageCount)); + var producerFuture = CompletableFuture.runAsync(() -> { try { - admin.namespaces().unloadNamespaceBundle( - defaultTestNamespace, bundle.getBundleRange(), finalDstBrokerUrl); - } catch (PulsarAdminException e) { - throw new RuntimeException(e); + cdlStart.await(); + for (int i = 0; i < messageCount; i++) { + semMessagesReadyToSend.acquire(); + String message = String.format("message-%d", i); + if (topicDomain == TopicDomain.persistent) { + // Only verify receipt of persistent topic messages. + pendingMessages.add(message); + } + producer.send(message); + } + } catch (PulsarClientException | InterruptedException e) { + fail(); + } + }, executor); + futures.add(producerFuture); + + consumers.stream().map(consumer -> CompletableFuture.runAsync(() -> { + try { + cdlStart.await(); + } catch (InterruptedException e) { + fail(); } + while (!producerFuture.isDone() || !pendingMessages.isEmpty()) { + try { + var message = consumer.receive(1500, TimeUnit.MILLISECONDS); + if (message != null) { + consumer.acknowledge(message); + pendingMessages.remove(message.getValue()); + } + } catch (PulsarClientException e) { + // Retry read + } + } + }, executor)).forEach(futures::add); + + var asyncTasks = FutureUtil.waitForAllAndSupportCancel(futures).orTimeout(timeoutMs, TimeUnit.MILLISECONDS); + + cdlStart.countDown(); + Awaitility.await().atMost(timeoutMs, TimeUnit.MILLISECONDS).ignoreExceptions().until( + () -> dstBrokerServiceUrl.equals(admin.lookups().lookupTopic(topic))); + + asyncTasks.get(); + + assertTrue(futures.stream().allMatch(CompletableFuture::isDone)); + assertTrue(futures.stream().noneMatch(CompletableFuture::isCompletedExceptionally)); + assertTrue(pendingMessages.isEmpty()); + + assertTrue(producer.isConnected()); + assertTrue(consumers.stream().allMatch(Consumer::isConnected)); + + for (LookupService lookupService : lookups) { + verify(lookupService, never()).getBroker(topicName); } - ); + } finally { + for (var consumer: consumers) { + consumer.close(); + } + for (var client: clients) { + client.close(); + } + } + } - Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { - try { - producer.send("hi".getBytes()); - String newOwner = admin.lookups().lookupTopic(topic); - assertEquals(dstBrokerServiceUrl, newOwner); - } catch (PulsarClientException e) { - throw new RuntimeException(e); - } catch (PulsarAdminException e) { - throw new RuntimeException(e); + @Test(timeOut = 30 * 1000, dataProvider = "isPersistentTopicSubscriptionTypeTest") + public void testUnloadClientReconnectionWithLookup(TopicDomain topicDomain, + SubscriptionType subscriptionType) throws Exception { + var id = String.format("test-unload-%s-client-reconnect-%s-%s", + topicDomain, subscriptionType, UUID.randomUUID()); + var topic = String.format("%s://%s/%s", topicDomain, defaultTestNamespace, id); + var topicName = TopicName.get(topic); + + var consumers = new ArrayList>(); + try { + @Cleanup + var producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); + + var consumerCount = subscriptionType == SubscriptionType.Exclusive ? 1 : 3; + for (int i = 0; i < consumerCount; i++) { + consumers.add(pulsarClient.newConsumer(Schema.STRING). + subscriptionName(id).subscriptionType(subscriptionType).topic(topic).subscribe()); } - }); + Awaitility.await() + .until(() -> producer.isConnected() && consumers.stream().allMatch(Consumer::isConnected)); - Awaitility.await().atMost(5, TimeUnit.SECONDS).until(producer::isConnected); - verify(lookup, times(lookupCountBeforeUnload)).getBroker(topicName); - producer.close(); - } + var lookup = spyLookupService(pulsarClient); + final CountDownLatch cdl = new CountDownLatch(3); + NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get(); + CompletableFuture unloadNamespaceBundle = CompletableFuture.runAsync(() -> { + try { + cdl.await(); + admin.namespaces().unloadNamespaceBundle(defaultTestNamespace, bundle.getBundleRange()); + } catch (InterruptedException | PulsarAdminException e) { + fail(); + } + }); - @Test(timeOut = 30 * 1000, dataProvider = "isPersistentTopicTest") - public void testUnloadClientReconnectionWithLookup(boolean isPersistentTopicTest) throws Exception { - String topicType = isPersistentTopicTest? "persistent" : "non-persistent"; - String topic = topicType + "://" + defaultTestNamespace + "/test-unload-client-reconnect-" - + isPersistentTopicTest; - TopicName topicName = TopicName.get(topic); + MutableInt sendCount = new MutableInt(); + Awaitility.await().atMost(20, TimeUnit.SECONDS).ignoreExceptions().until(() -> { + var message = String.format("message-%d", sendCount.getValue()); - AtomicInteger lookupCount = new AtomicInteger(); - var lookup = spyLookupService(lookupCount, topicName); + boolean messageSent = false; + while (true) { + var recvFutures = consumers.stream(). + map(consumer -> consumer.receiveAsync().orTimeout(1000, TimeUnit.MILLISECONDS)). + collect(Collectors.toList()); - var producer = pulsarClient.newProducer().topic(topic).create(); - int lookupCountBeforeUnload = lookupCount.get(); + if (!messageSent) { + producer.send(message); + messageSent = true; + } - NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get(); - CompletableFuture.runAsync(() -> { - try { - admin.namespaces().unloadNamespaceBundle( - defaultTestNamespace, bundle.getBundleRange()); - } catch (PulsarAdminException e) { - throw new RuntimeException(e); + if (topicDomain == TopicDomain.non_persistent) { + // No need to wait for message receipt, we're only trying to stress the consumer lookup pathway. + break; + } + var msg = (Message) FutureUtil.waitForAny(recvFutures, __ -> true).get().get(); + if (Objects.equals(msg.getValue(), message)) { + break; } } - ); - MutableInt sendCount = new MutableInt(); - Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { - try { - producer.send("hi".getBytes()); - assertEquals(sendCount.incrementAndGet(), 10); - } catch (PulsarClientException e) { - throw new RuntimeException(e); + + cdl.countDown(); + return sendCount.incrementAndGet() == 10; + }); + + assertTrue(producer.isConnected()); + assertTrue(consumers.stream().allMatch(Consumer::isConnected)); + assertTrue(unloadNamespaceBundle.isDone()); + verify(lookup, times(1 + consumerCount)).getBroker(topicName); + } finally { + for (var consumer : consumers) { + consumer.close(); } - }); - assertTrue(producer.isConnected()); - verify(lookup, times(lookupCountBeforeUnload + 1)).getBroker(topicName); - producer.close(); + } } - private LookupService spyLookupService(AtomicInteger lookupCount, TopicName topicName) - throws IllegalAccessException { - var lookup = spy(lookupService); - FieldUtils.writeDeclaredField(pulsarClient, "lookup", lookup, true); - doAnswer(invocationOnMock -> { - lookupCount.incrementAndGet(); - return invocationOnMock.callRealMethod(); - }).when(lookup).getBroker(topicName); + private LookupService spyLookupService(PulsarClient client) throws IllegalAccessException { + LookupService svc = (LookupService) FieldUtils.readDeclaredField(client, "lookup", true); + var lookup = spy(svc); + FieldUtils.writeDeclaredField(client, "lookup", lookup, true); return lookup; } - private void checkOwnershipState(String broker, NamespaceBundle bundle) throws ExecutionException, InterruptedException { var targetLoadManager = secondaryLoadManager; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java index 332cccc2d2c6a..03aaf3c7bb275 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java @@ -37,6 +37,7 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; @@ -279,7 +280,8 @@ public boolean canUnsubscribe(Consumer consumer) { } @Override - public CompletableFuture close() { + public CompletableFuture close(boolean disconnectConsumers, + Optional assignedBrokerLookupData) { return null; } @@ -294,7 +296,8 @@ public CompletableFuture disconnectActiveConsumers(boolean isResetCursor) } @Override - public CompletableFuture disconnectAllConsumers(boolean isResetCursor) { + public CompletableFuture disconnectAllConsumers(boolean isResetCursor, + Optional assignedBrokerLookupData) { return null; } 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 0f0440d24dde7..b6dd42d702860 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 @@ -2413,7 +2413,8 @@ public void testSubscribeBookieTimeout() throws Exception { "test" /* consumer name */, 0 /* avoid reseting cursor */); channel.writeInbound(subscribe1); - ByteBuf closeConsumer = Commands.newCloseConsumer(1 /* consumer id */, 2 /* request id */); + ByteBuf closeConsumer = Commands.newCloseConsumer(1 /* consumer id */, 2 /* request id */, + null /* assignedBrokerServiceUrl */, null /* assignedBrokerServiceUrlTls */); channel.writeInbound(closeConsumer); ByteBuf subscribe2 = Commands.newSubscribe(successTopicName, // diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 6959d8dd04861..86b5883990357 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1378,7 +1378,7 @@ public void testGetConnectExceptionForAckMsgWhenCnxIsNull() throws Exception { producer.newMessage().value(Bytes.toBytes(i)).send(); } ClientCnx cnx = (ClientCnx) MethodUtils.invokeMethod(consumer, true, "cnx"); - MethodUtils.invokeMethod(consumer, true, "connectionClosed", cnx); + MethodUtils.invokeMethod(consumer, true, "connectionClosed", cnx, Optional.empty(), Optional.empty()); Message message = consumer.receive(); Transaction transaction = pulsarClient 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 61c7a98602b69..705b171929be6 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 @@ -270,7 +270,7 @@ private void consumerCreatedThenFailsRetryTimeout(String topic) throws Exception if (subscribeCount == 1) { ctx.writeAndFlush(Commands.newSuccess(subscribe.getRequestId())); // Trigger reconnect - ctx.writeAndFlush(Commands.newCloseConsumer(subscribe.getConsumerId(), -1)); + ctx.writeAndFlush(Commands.newCloseConsumer(subscribe.getConsumerId(), -1, null, null)); } else if (subscribeCount != 2) { // Respond to subsequent requests to prevent timeouts ctx.writeAndFlush(Commands.newSuccess(subscribe.getRequestId())); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java index 9036d82d84f01..6304ed82d4f87 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java @@ -21,6 +21,7 @@ import static org.awaitility.Awaitility.await; import com.google.common.collect.Sets; import java.time.Duration; +import java.util.Optional; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.broker.BrokerTestUtil; @@ -908,7 +909,7 @@ public void testClosingRateLimiter(SubscriptionType subscription) throws Excepti producer.close(); consumer.close(); - sub.disconnect().get(); + sub.close(true, Optional.empty()).get(); // Make sure that the rate limiter is closed Assert.assertEquals(dispatchRateLimiter.getDispatchRateOnMsg(), -1); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java index 28eef0326cce3..0395c59d58307 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java @@ -104,6 +104,7 @@ import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap; import org.awaitility.Awaitility; import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.Assert; @@ -173,26 +174,15 @@ public void testDisconnectClientWithoutClosingConnection() throws Exception { doAnswer(invocationOnMock -> cons1.getState()).when(consumer1).getState(); doAnswer(invocationOnMock -> cons1.getClientCnx()).when(consumer1).getClientCnx(); doAnswer(invocationOnMock -> cons1.cnx()).when(consumer1).cnx(); - doAnswer(invocationOnMock -> { - cons1.connectionClosed((ClientCnx) invocationOnMock.getArguments()[0]); - return null; - }).when(consumer1).connectionClosed(any()); + doAnswer(InvocationOnMock::callRealMethod).when(consumer1).connectionClosed(any(), any(), any()); ProducerImpl producer1 = spy(prod1); doAnswer(invocationOnMock -> prod1.getState()).when(producer1).getState(); doAnswer(invocationOnMock -> prod1.getClientCnx()).when(producer1).getClientCnx(); doAnswer(invocationOnMock -> prod1.cnx()).when(producer1).cnx(); - doAnswer(invocationOnMock -> { - prod1.connectionClosed((ClientCnx) invocationOnMock.getArguments()[0]); - return null; - }).when(producer1).connectionClosed(any()); ProducerImpl producer2 = spy(prod2); doAnswer(invocationOnMock -> prod2.getState()).when(producer2).getState(); doAnswer(invocationOnMock -> prod2.getClientCnx()).when(producer2).getClientCnx(); doAnswer(invocationOnMock -> prod2.cnx()).when(producer2).cnx(); - doAnswer(invocationOnMock -> { - prod2.connectionClosed((ClientCnx) invocationOnMock.getArguments()[0]); - return null; - }).when(producer2).connectionClosed(any()); ClientCnx clientCnx = producer1.getClientCnx(); @@ -223,11 +213,11 @@ public void testDisconnectClientWithoutClosingConnection() throws Exception { // let server send signal to close-connection and client close the connection Thread.sleep(1000); // [1] Verify: producer1 must get connectionClosed signal - verify(producer1, atLeastOnce()).connectionClosed(any()); + verify(producer1, atLeastOnce()).connectionClosed(any(), any(), any()); // [2] Verify: consumer1 must get connectionClosed signal - verify(consumer1, atLeastOnce()).connectionClosed(any()); + verify(consumer1, atLeastOnce()).connectionClosed(any(), any(), any()); // [3] Verify: producer2 should have not received connectionClosed signal - verify(producer2, never()).connectionClosed(any()); + verify(producer2, never()).connectionClosed(any(), any(), any()); // sleep for sometime to let other disconnected producer and consumer connect again: but they should not get // connected with same broker as that broker is already out from active-broker list @@ -247,7 +237,7 @@ public void testDisconnectClientWithoutClosingConnection() throws Exception { pulsar.getNamespaceService().unloadNamespaceBundle((NamespaceBundle) bundle2).join(); // let producer2 give some time to get disconnect signal and get disconnected Thread.sleep(200); - verify(producer2, atLeastOnce()).connectionClosed(any()); + verify(producer2, atLeastOnce()).connectionClosed(any(), any(), any()); // producer1 must not be able to connect again assertNull(prod1.getClientCnx()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java index d588ac8626f7b..8126ba1bba928 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java @@ -192,7 +192,7 @@ public void disconnectProducerAndRejectReconnecting(ProducerImpl producer) th // make the existing connection between the producer and broker to break by explicitly closing it ClientCnx cnx = producer.cnx(); - producer.connectionClosed(cnx); + producer.connectionClosed(cnx, Optional.empty(), Optional.empty()); cnx.close(); } 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 27ddd21249f86..75e84eeca3e6a 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 @@ -323,8 +323,8 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { waitingLookupRequests.forEach(pair -> pair.getRight().getRight().completeExceptionally(e)); // Notify all attached producers/consumers so they have a chance to reconnect - producers.forEach((id, producer) -> producer.connectionClosed(this)); - consumers.forEach((id, consumer) -> consumer.connectionClosed(this)); + producers.forEach((id, producer) -> producer.connectionClosed(this, Optional.empty(), Optional.empty())); + consumers.forEach((id, consumer) -> consumer.connectionClosed(this, Optional.empty(), Optional.empty())); transactionMetaStoreHandlers.forEach((id, handler) -> handler.connectionClosed(this)); topicListWatchers.forEach((__, watcher) -> watcher.connectionClosed(this)); @@ -803,46 +803,72 @@ protected void handleError(CommandError error) { @Override protected void handleCloseProducer(CommandCloseProducer closeProducer) { final long producerId = closeProducer.getProducerId(); + log.info("[{}] Broker notification of closed producer: {}, assignedBrokerUrl: {}, assignedBrokerUrlTls: {}", + remoteAddress, producerId, + closeProducer.hasAssignedBrokerServiceUrl() ? closeProducer.getAssignedBrokerServiceUrl() : null, + closeProducer.hasAssignedBrokerServiceUrlTls() ? closeProducer.getAssignedBrokerServiceUrlTls() : null); ProducerImpl producer = producers.remove(producerId); if (producer != null) { - if (closeProducer.hasAssignedBrokerServiceUrl() || closeProducer.hasAssignedBrokerServiceUrlTls()) { - try { - final URI uri = new URI(producer.client.conf.isUseTls() - ? closeProducer.getAssignedBrokerServiceUrlTls() - : closeProducer.getAssignedBrokerServiceUrl()); - log.info("[{}] Broker notification of Closed producer: {}. Redirecting to {}.", - remoteAddress, closeProducer.getProducerId(), uri); - producer.getConnectionHandler().connectionClosed( - this, Optional.of(0L), Optional.of(uri)); - } catch (Throwable e) { - log.error("[{}] Invalid redirect url {}/{} for {}", remoteAddress, - closeProducer.hasAssignedBrokerServiceUrl() - ? closeProducer.getAssignedBrokerServiceUrl() : "", - closeProducer.hasAssignedBrokerServiceUrlTls() - ? closeProducer.getAssignedBrokerServiceUrlTls() : "", - closeProducer.getRequestId(), e); - producer.connectionClosed(this); - } - } else { - log.info("[{}] Broker notification of Closed producer: {}.", - remoteAddress, closeProducer.getProducerId()); - producer.connectionClosed(this); - } + String brokerServiceUrl = getBrokerServiceUrl(closeProducer, producer); + Optional hostUri = parseUri(brokerServiceUrl, + closeProducer.hasRequestId() ? closeProducer.getRequestId() : null); + Optional initialConnectionDelayMs = hostUri.map(__ -> 0L); + producer.connectionClosed(this, initialConnectionDelayMs, hostUri); } else { - log.warn("Producer with id {} not found while closing producer ", producerId); + log.warn("[{}] Producer with id {} not found while closing producer", remoteAddress, producerId); + } + } + + private static String getBrokerServiceUrl(CommandCloseProducer closeProducer, ProducerImpl producer) { + if (producer.getClient().getConfiguration().isUseTls()) { + if (closeProducer.hasAssignedBrokerServiceUrlTls()) { + return closeProducer.getAssignedBrokerServiceUrlTls(); + } + } else if (closeProducer.hasAssignedBrokerServiceUrl()) { + return closeProducer.getAssignedBrokerServiceUrl(); } + return null; } @Override protected void handleCloseConsumer(CommandCloseConsumer closeConsumer) { - log.info("[{}] Broker notification of Closed consumer: {}", remoteAddress, closeConsumer.getConsumerId()); final long consumerId = closeConsumer.getConsumerId(); + log.info("[{}] Broker notification of closed consumer: {}, assignedBrokerUrl: {}, assignedBrokerUrlTls: {}", + remoteAddress, consumerId, + closeConsumer.hasAssignedBrokerServiceUrl() ? closeConsumer.getAssignedBrokerServiceUrl() : null, + closeConsumer.hasAssignedBrokerServiceUrlTls() ? closeConsumer.getAssignedBrokerServiceUrlTls() : null); ConsumerImpl consumer = consumers.remove(consumerId); if (consumer != null) { - consumer.connectionClosed(this); + String brokerServiceUrl = getBrokerServiceUrl(closeConsumer, consumer); + Optional hostUri = parseUri(brokerServiceUrl, + closeConsumer.hasRequestId() ? closeConsumer.getRequestId() : null); + Optional initialConnectionDelayMs = hostUri.map(__ -> 0L); + consumer.connectionClosed(this, initialConnectionDelayMs, hostUri); } else { - log.warn("Consumer with id {} not found while closing consumer ", consumerId); + log.warn("[{}] Consumer with id {} not found while closing consumer", remoteAddress, consumerId); + } + } + + private static String getBrokerServiceUrl(CommandCloseConsumer closeConsumer, ConsumerImpl consumer) { + if (consumer.getClient().getConfiguration().isUseTls()) { + if (closeConsumer.hasAssignedBrokerServiceUrlTls()) { + return closeConsumer.getAssignedBrokerServiceUrlTls(); + } + } else if (closeConsumer.hasAssignedBrokerServiceUrl()) { + return closeConsumer.getAssignedBrokerServiceUrl(); + } + return null; + } + + private Optional parseUri(String url, Long requestId) { + try { + if (url != null) { + return Optional.of(new URI(url)); + } + } catch (URISyntaxException e) { + log.warn("[{}] Invalid redirect URL {}, requestId {}: ", remoteAddress, url, requestId, e); } + return Optional.empty(); } @Override 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 600dc17a1b09a..178046864c987 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 @@ -182,11 +182,11 @@ public void connectionClosed(ClientCnx cnx, Optional initialConnectionDela } long delayMs = initialConnectionDelayMs.orElse(backoff.next()); state.setState(State.Connecting); - log.info("[{}] [{}] Closed connection {} -- Will try again in {} s", - state.topic, state.getHandlerName(), cnx.channel(), - delayMs / 1000.0); + log.info("[{}] [{}] Closed connection {} -- Will try again in {} s, hostUrl: {}", + state.topic, state.getHandlerName(), cnx.channel(), delayMs / 1000.0, hostUrl.orElse(null)); state.client.timer().newTimeout(timeout -> { - log.info("[{}] [{}] Reconnecting after timeout", state.topic, state.getHandlerName()); + log.info("[{}] [{}] Reconnecting after {} s timeout, hostUrl: {}", + state.topic, state.getHandlerName(), delayMs / 1000.0, hostUrl.orElse(null)); grabCnx(hostUrl); }, delayMs, TimeUnit.MILLISECONDS); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java index fbc2a8c285dd2..e7be0b2dbd473 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java @@ -35,6 +35,7 @@ import io.netty.util.Timeout; import io.netty.util.concurrent.FastThreadLocal; import java.io.IOException; +import java.net.URI; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; @@ -888,7 +889,7 @@ public CompletableFuture connectionOpened(final ClientCnx cnx) { // in case it was indeed created, otherwise it might prevent new create consumer operation, // since we are not necessarily closing the connection. long closeRequestId = client.newRequestId(); - ByteBuf cmd = Commands.newCloseConsumer(consumerId, closeRequestId); + ByteBuf cmd = Commands.newCloseConsumer(consumerId, closeRequestId, null, null); cnx.sendRequestWithId(cmd, closeRequestId); } @@ -1057,7 +1058,7 @@ public CompletableFuture closeAsync() { if (null == cnx) { cleanupAtClose(closeFuture, null); } else { - ByteBuf cmd = Commands.newCloseConsumer(consumerId, requestId); + ByteBuf cmd = Commands.newCloseConsumer(consumerId, requestId, null, null); cnx.sendRequestWithId(cmd, requestId).handle((v, exception) -> { final ChannelHandlerContext ctx = cnx.ctx(); boolean ignoreException = ctx == null || !ctx.channel().isActive(); @@ -2669,8 +2670,8 @@ void resetBackoff() { this.connectionHandler.resetBackoff(); } - void connectionClosed(ClientCnx cnx) { - this.connectionHandler.connectionClosed(cnx); + void connectionClosed(ClientCnx cnx, Optional initialConnectionDelayMs, Optional hostUrl) { + this.connectionHandler.connectionClosed(cnx, initialConnectionDelayMs, hostUrl); } public ClientCnx getClientCnx() { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index a17d4a06f02a6..2763da524cd58 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -41,6 +41,7 @@ import io.netty.util.TimerTask; import io.netty.util.concurrent.ScheduledFuture; import java.io.IOException; +import java.net.URI; import java.nio.ByteBuffer; import java.util.ArrayDeque; import java.util.ArrayList; @@ -2372,8 +2373,8 @@ void resetBackoff() { this.connectionHandler.resetBackoff(); } - void connectionClosed(ClientCnx cnx) { - this.connectionHandler.connectionClosed(cnx); + void connectionClosed(ClientCnx cnx, Optional initialConnectionDelayMs, Optional hostUrl) { + this.connectionHandler.connectionClosed(cnx, initialConnectionDelayMs, hostUrl); } public ClientCnx getClientCnx() { 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 22220805814f5..4f657da82b289 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 @@ -33,6 +33,7 @@ import io.netty.channel.EventLoopGroup; import io.netty.util.concurrent.DefaultThreadFactory; import java.lang.reflect.Field; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; @@ -278,13 +279,19 @@ public void testHandleCloseConsumer() { ClientCnx cnx = new ClientCnx(conf, eventLoop); long consumerId = 1; - cnx.registerConsumer(consumerId, mock(ConsumerImpl.class)); + PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class); + when(pulsarClient.getConfiguration()).thenReturn(conf); + ConsumerImpl consumer = mock(ConsumerImpl.class); + when(consumer.getClient()).thenReturn(pulsarClient); + cnx.registerConsumer(consumerId, consumer); assertEquals(cnx.consumers.size(), 1); - CommandCloseConsumer closeConsumer = new CommandCloseConsumer().setConsumerId(consumerId); + CommandCloseConsumer closeConsumer = new CommandCloseConsumer().setConsumerId(consumerId).setRequestId(1); cnx.handleCloseConsumer(closeConsumer); assertEquals(cnx.consumers.size(), 0); + verify(consumer).connectionClosed(cnx, Optional.empty(), Optional.empty()); + eventLoop.shutdownGracefully(); } @@ -296,13 +303,19 @@ public void testHandleCloseProducer() { ClientCnx cnx = new ClientCnx(conf, eventLoop); long producerId = 1; - cnx.registerProducer(producerId, mock(ProducerImpl.class)); + PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class); + when(pulsarClient.getConfiguration()).thenReturn(conf); + ProducerImpl producer = mock(ProducerImpl.class); + when(producer.getClient()).thenReturn(pulsarClient); + cnx.registerProducer(producerId, producer); assertEquals(cnx.producers.size(), 1); - CommandCloseProducer closeProducerCmd = new CommandCloseProducer().setProducerId(producerId); + CommandCloseProducer closeProducerCmd = new CommandCloseProducer().setProducerId(producerId).setRequestId(1); cnx.handleCloseProducer(closeProducerCmd); assertEquals(cnx.producers.size(), 0); + verify(producer).connectionClosed(cnx, Optional.empty(), Optional.empty()); + eventLoop.shutdownGracefully(); } 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 ff116d2406b40..e715173be5287 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 @@ -60,6 +60,7 @@ import org.apache.pulsar.common.api.proto.CommandAddSubscriptionToTxn; import org.apache.pulsar.common.api.proto.CommandAddSubscriptionToTxnResponse; import org.apache.pulsar.common.api.proto.CommandAuthChallenge; +import org.apache.pulsar.common.api.proto.CommandCloseConsumer; import org.apache.pulsar.common.api.proto.CommandCloseProducer; import org.apache.pulsar.common.api.proto.CommandConnect; import org.apache.pulsar.common.api.proto.CommandConnected; @@ -737,11 +738,21 @@ public static ByteBuf newSeek(long consumerId, long requestId, long timestamp) { return serializeWithSize(cmd); } - public static ByteBuf newCloseConsumer(long consumerId, long requestId) { + public static ByteBuf newCloseConsumer( + long consumerId, long requestId, String assignedBrokerUrl, String assignedBrokerUrlTls) { BaseCommand cmd = localCmd(Type.CLOSE_CONSUMER); - cmd.setCloseConsumer() + CommandCloseConsumer commandCloseConsumer = cmd.setCloseConsumer() .setConsumerId(consumerId) .setRequestId(requestId); + + if (assignedBrokerUrl != null) { + commandCloseConsumer.setAssignedBrokerServiceUrl(assignedBrokerUrl); + } + + if (assignedBrokerUrlTls != null) { + commandCloseConsumer.setAssignedBrokerServiceUrlTls(assignedBrokerUrlTls); + } + return serializeWithSize(cmd); } diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index 2c350aaf8a10e..819c6dfd59475 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -648,6 +648,8 @@ message CommandCloseProducer { message CommandCloseConsumer { required uint64 consumer_id = 1; required uint64 request_id = 2; + optional string assignedBrokerServiceUrl = 3; + optional string assignedBrokerServiceUrlTls = 4; } message CommandRedeliverUnacknowledgedMessages {