[improve][broker] PIP-307: Implement broker and client consumer changes when topic is unloading

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<Void> close() {
+    @Override
+    public CompletableFuture<Void> close(boolean disconnectConsumers,
+                                         Optional<BrokerLookupData> 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<Void> disconnectAllConsumers(boolean isResetCursor) {
+    @Override
+    public synchronized CompletableFuture<Void> disconnectAllConsumers(
+            boolean isResetCursor, Optional<BrokerLookupData> 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

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<BrokerLookupData> assignedBrokerLookupData) {
         log.info("Disconnecting consumer: {}", this);
-        cnx.closeConsumer(this);
+        cnx.closeConsumer(this, assignedBrokerLookupData);
         try {
             close(isResetCursor);
         } catch (BrokerServiceException e) {

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<Void> close();
+    default CompletableFuture<Void> close() {
+        return close(true, Optional.empty());
+    }
+
+    CompletableFuture<Void> close(boolean disconnectClients, Optional<BrokerLookupData> assignedBrokerLookupData);
 
     boolean isClosed();
 
@@ -63,12 +68,17 @@ public interface Dispatcher {
      *
      * @return
      */
-    CompletableFuture<Void> disconnectAllConsumers(boolean isResetCursor);
+    default CompletableFuture<Void> disconnectAllConsumers(boolean isResetCursor) {
+        return disconnectAllConsumers(isResetCursor, Optional.empty());
+    }
 
     default CompletableFuture<Void> disconnectAllConsumers() {
         return disconnectAllConsumers(false);
     }
 
+    CompletableFuture<Void> disconnectAllConsumers(boolean isResetCursor,
+                                                   Optional<BrokerLookupData> assignedBrokerLookupData);
+
     void resetCloseFuture();
 
     /**

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<Void> disconnect() {
      */
     public CompletableFuture<Void> disconnect(Optional<BrokerLookupData> 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);

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<BrokerLookupDat
     }
 
     @Override
-    public void closeConsumer(Consumer consumer) {
+    public void closeConsumer(Consumer consumer, Optional<BrokerLookupData> 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<BrokerLookupData> 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();
         }

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<Consumer> getConsumers();
 
-    CompletableFuture<Void> close();
-
     CompletableFuture<Void> delete();
 
     CompletableFuture<Void> deleteForcefully();
 
-    CompletableFuture<Void> disconnect();
+    CompletableFuture<Void> disconnect(Optional<BrokerLookupData> assignedBrokerLookupData);
+
+    CompletableFuture<Void> close(boolean disconnectConsumers, Optional<BrokerLookupData> assignedBrokerLookupData);
 
     CompletableFuture<Void> doUnsubscribe(Consumer consumer);
 

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<BrokerLookupData> assignedBrokerLookupData);
 
     boolean isPreciseDispatcherFlowControl();
 

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<Void> close() {
+    public CompletableFuture<Void> close(boolean disconnectConsumers,
+                                         Optional<BrokerLookupData> 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<Void> disconnectAllConsumers(boolean isResetCursor) {
+    public synchronized CompletableFuture<Void> disconnectAllConsumers(
+            boolean isResetCursor, Optional<BrokerLookupData> assignedBrokerLookupData) {
         closeFuture = new CompletableFuture<>();
         if (consumerList.isEmpty()) {
             closeFuture.complete(null);
         } else {
-            consumerList.forEach(Consumer::disconnect);
+            consumerList.forEach(consumer -> consumer.disconnect(isResetCursor, assignedBrokerLookupData));
         }
         return closeFuture;
     }

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