(pulsar) branch branch-3.1 updated: [fix] [broker] Replication stopped due to unload topic failed (#21947)

[te...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

technoboy pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-3.1 by this push:
     new 291bbb531c2 [fix] [broker] Replication stopped due to unload topic failed (#21947)
291bbb531c2 is described below

commit 291bbb531c24fd00d6850b0b5bd88f3438e1b775
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Jan 25 17:02:26 2024 +0800

    [fix] [broker] Replication stopped due to unload topic failed (#21947)
    
    ### Motivation
    
    **Steps to reproduce the issue**
    - Enable replication.
    - Send `10` messages to the local cluster then close the producer.
    - Call `pulsar-admin topics unload <topic>` and get an error due to the internal producer of the replicator close failing.
    - The topic closed failed, so we assumed the topic could work as expected, but the replication stopped.
    
    **Root cause**
    - `pulsar-admin topics unload <topic>`  will wait for the clients(including `consumers & producers & replicators`) to close successfully, and it will fail if clients can not be closed successfully.
    - `replicator.producer` close failed causing the Admin API to fail, but there is a scheduled task that will retry to close `replicator.producer` which causes replication to stop. see https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java#L209
    
    ### Modifications
    
    since the "replicator.producer.closeAsync()" will retry after it fails, the topic unload should be successful.
---
 .../pulsar/broker/service/AbstractReplicator.java  |  3 +-
 .../broker/service/OneWayReplicatorTest.java       | 70 +++++++++++++++++++---
 2 files changed, 63 insertions(+), 10 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
index 6dd296d16b5..1b5b2824257 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
@@ -194,7 +194,7 @@ public abstract class AbstractReplicator {
             return CompletableFuture.completedFuture(null);
         }
         CompletableFuture<Void> future = producer.closeAsync();
-        future.thenRun(() -> {
+        return future.thenRun(() -> {
             STATE_UPDATER.set(this, State.Stopped);
             this.producer = null;
             // deactivate further read
@@ -209,7 +209,6 @@ public abstract class AbstractReplicator {
             brokerService.executor().schedule(this::closeProducerAsync, waitTimeMs, TimeUnit.MILLISECONDS);
             return null;
         });
-        return future;
     }
 
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index e068cd55e90..5d748a9689e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -21,16 +21,22 @@ package org.apache.pulsar.broker.service;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
+import java.lang.reflect.Field;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.ProducerImpl;
 import org.apache.pulsar.common.policies.data.TopicStats;
 import org.awaitility.Awaitility;
+import org.mockito.Mockito;
+import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -50,6 +56,29 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
         super.cleanup();
     }
 
+    private void waitReplicatorStarted(String topicName) {
+        Awaitility.await().untilAsserted(() -> {
+            Optional<Topic> topicOptional2 = pulsar2.getBrokerService().getTopic(topicName, false).get();
+            assertTrue(topicOptional2.isPresent());
+            PersistentTopic persistentTopic2 = (PersistentTopic) topicOptional2.get();
+            assertFalse(persistentTopic2.getProducers().isEmpty());
+        });
+    }
+
+    /**
+     * Override "AbstractReplicator.producer" by {@param producer} and return the original value.
+     */
+    private ProducerImpl overrideProducerForReplicator(AbstractReplicator replicator, ProducerImpl newProducer)
+            throws Exception {
+        Field producerField = AbstractReplicator.class.getDeclaredField("producer");
+        producerField.setAccessible(true);
+        ProducerImpl originalValue = (ProducerImpl) producerField.get(replicator);
+        synchronized (replicator) {
+            producerField.set(replicator, newProducer);
+        }
+        return originalValue;
+    }
+
     @Test
     public void testReplicatorProducerStatInTopic() throws Exception {
         final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_");
@@ -85,18 +114,13 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
     public void testCreateRemoteConsumerFirst() throws Exception {
         final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_");
         Producer<String> producer1 = client1.newProducer(Schema.STRING).topic(topicName).create();
-        // Wait for replicator started.
-        Awaitility.await().untilAsserted(() -> {
-            Optional<Topic> topicOptional2 = pulsar2.getBrokerService().getTopic(topicName, false).get();
-            assertTrue(topicOptional2.isPresent());
-            PersistentTopic persistentTopic2 = (PersistentTopic) topicOptional2.get();
-            assertFalse(persistentTopic2.getProducers().isEmpty());
-        });
+
         // The topic in cluster2 has a replicator created producer(schema Auto_Produce), but does not have any schema。
         // Verify: the consumer of this cluster2 can create successfully.
         Consumer<String> consumer2 = client2.newConsumer(Schema.STRING).topic(topicName).subscriptionName("s1")
                 .subscribe();;
-
+        // Wait for replicator started.
+        waitReplicatorStarted(topicName);
         // cleanup.
         producer1.close();
         consumer2.close();
@@ -105,4 +129,34 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
             admin2.topics().delete(topicName);
         });
     }
+
+    @Test
+    public void testTopicCloseWhenInternalProducerCloseErrorOnce() throws Exception {
+        final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_");
+        admin1.topics().createNonPartitionedTopic(topicName);
+        // Wait for replicator started.
+        waitReplicatorStarted(topicName);
+        PersistentTopic persistentTopic =
+                (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+        PersistentReplicator replicator =
+                (PersistentReplicator) persistentTopic.getReplicators().values().iterator().next();
+        // Mock an error when calling "replicator.disconnect()"
+        ProducerImpl mockProducer = Mockito.mock(ProducerImpl.class);
+        Mockito.when(mockProducer.closeAsync()).thenReturn(CompletableFuture.failedFuture(new Exception("mocked ex")));
+        ProducerImpl originalProducer = overrideProducerForReplicator(replicator, mockProducer);
+        // Verify: since the "replicator.producer.closeAsync()" will retry after it failed, the topic unload should be
+        // successful.
+        admin1.topics().unload(topicName);
+        // Verify: After "replicator.producer.closeAsync()" retry again, the "replicator.producer" will be closed
+        // successful.
+        overrideProducerForReplicator(replicator, originalProducer);
+        Awaitility.await().untilAsserted(() -> {
+            Assert.assertFalse(replicator.isConnected());
+        });
+        // cleanup.
+        cleanupTopics(() -> {
+            admin1.topics().delete(topicName);
+            admin2.topics().delete(topicName);
+        });
+    }
 }