Add tests for reconnection ProducerBusy

Author: BewareMyPower

lib/ProducerImpl.cc

@@ -966,7 +966,7 @@ bool ProducerImpl::encryptMessage(proto::MessageMetadata& metadata, SharedBuffer
 }
 
 void ProducerImpl::disconnectProducer() {
-    LOG_DEBUG("Broker notification of Closed producer: " << producerId_);
+    LOG_INFO("Broker notification of Closed producer: " << producerId_);
     resetCnx();
     scheduleReconnection();
 }

tests/ProducerTest.cc

@@ -618,4 +618,19 @@ TEST(ProducerTest, testNoDeadlockWhenClosingPartitionedProducerAfterPartitionsUp
     client.close();
 }
 
+TEST(ProducerTest, testReconnectMultiConnectionsPerBroker) {
+    ClientConfiguration conf;
+    conf.setConnectionsPerBroker(10);
+
+    Client client(serviceUrl, conf);
+    Producer producer;
+    ASSERT_EQ(ResultOk, client.createProducer("producer-test-reconnect-twice", producer));
+
+    for (int i = 0; i < 5; i++) {
+        ASSERT_TRUE(PulsarFriend::reconnect(producer)) << "i: " << i;
+    }
+
+    client.close();
+}
+
 INSTANTIATE_TEST_CASE_P(Pulsar, ProducerTest, ::testing::Values(true, false));

tests/PulsarFriend.h

@@ -21,6 +21,7 @@
 
 #include <string>
 
+#include "WaitUtils.h"
 #include "lib/ClientConnection.h"
 #include "lib/ClientImpl.h"
 #include "lib/ConsumerConfigurationImpl.h"
@@ -197,6 +198,13 @@ class PulsarFriend {
         lookupData->setPartitions(newPartitions);
         partitionedProducer.handleGetPartitions(ResultOk, lookupData);
     }
+
+    static bool reconnect(Producer producer) {
+        auto producerImpl = std::dynamic_pointer_cast<ProducerImpl>(producer.impl_);
+        producerImpl->disconnectProducer();
+        return waitUntil(std::chrono::seconds(3),
+                         [producerImpl] { return !producerImpl->getCnx().expired(); });
+    }
 };
 }  // namespace pulsar