Do not close the socket when the broker failed due to MetadataStoreException

### Motivation

When the broker failed to acquire the ownership of a namespace bundle by
`LockBusyException`. It means there is another broker that has acquired
the metadata store path and didn't release that path. For example:

Broker 1:

```
2024-01-24T23:35:36,626+0000 [metadata-store-10-1] WARN  org.apache.pulsar.broker.lookup.TopicLookupBase - Failed to lookup <role> for topic persistent://<tenant>/<ns>/<topic> with error org.apache.pulsar.broker.PulsarServerException: Failed to acquire ownership for namespace bundle <tenant>/<ns>/0x50000000_0x51000000
   Caused by: java.util.concurrent.CompletionException: org.apache.pulsar.metadata.api.MetadataStoreException$LockBusyException: Resource at /namespace/<tenant>/<ns>/0x50000000_0x51000000 is already locked
```

Broker 2:

```
2024-01-24T23:35:36,650+0000 [broker-topic-workers-OrderedExecutor-3-0] INFO  org.apache.pulsar.broker.PulsarService - Loaded 1 topics on <tenant>/<ns>/0x50000000_0x51000000 -- time taken: 0.044 seconds
```

After broker 2 released the lock at 23:35:36,650, the lookup request to
broker 1 should tell the client that namespace bundle
0x50000000_0x51000000 is currently being unloaded and in the next retry
the client will connect to the new owner broker.

Here is another typical error:

```
2024-01-24T23:57:57,264+0000 [pulsar-io-4-5] INFO  org.apache.pulsar.broker.lookup.TopicLookupBase - Failed to lookup <role> for topic persistent://<tenant>/<ns>/<topic> with error Namespace bundle <tenant>/<ns>/0x0d000000_0x0e000000 is being unloaded
```

Though after apache/pulsar#21211, the server
error becomes `MetadataError` rather than `ServiceNotReady`.

However, since the `ServerError` is `ServiceNotReady`, the client will
close the connection. If there are many other producers or consumers on
the same connection, they will all reestablish connection to the broker,
which is unnecessary and brings much pressure to broker side.

### Modifications

In `checkServerError`, when the error code is `ServiceNotReady`, check
the error message as well, if it hit the case in `handleLookupError`, do
not close the connection.

Add `ConnectionTest` on a mocked `ClientConnection` object to verify
`close()` will not be called.

Author: BewareMyPower

lib/ClientConnection.cc

@@ -24,6 +24,7 @@
 #include <fstream>
 
 #include "AsioDefines.h"
+#include "ClientConnectionAdaptor.h"
 #include "ClientImpl.h"
 #include "Commands.h"
 #include "ConnectionPool.h"
@@ -80,9 +81,7 @@ static Result getResult(ServerError serverError, const std::string& message) {
             return ResultConsumerBusy;
 
         case ServiceNotReady:
-            // If the error is not caused by a PulsarServerException, treat it as retryable.
-            return (message.find("PulsarServerException") == std::string::npos) ? ResultRetryable
-                                                                                : ResultServiceUnitNotReady;
+            return ResultServiceUnitNotReady;
 
         case ProducerBlockedQuotaExceededError:
             return ResultProducerBlockedQuotaExceededError;
@@ -1469,19 +1468,8 @@ Future<Result, SchemaInfo> ClientConnection::newGetSchema(const std::string& top
     return promise.getFuture();
 }
 
-void ClientConnection::checkServerError(ServerError error) {
-    switch (error) {
-        case proto::ServerError::ServiceNotReady:
-            close(ResultDisconnected);
-            break;
-        case proto::ServerError::TooManyRequests:
-            // TODO: Implement maxNumberOfRejectedRequestPerConnection like
-            // https://github.com/apache/pulsar/pull/274
-            close(ResultDisconnected);
-            break;
-        default:
-            break;
-    }
+void ClientConnection::checkServerError(ServerError error, const std::string& message) {
+    pulsar::adaptor::checkServerError(*this, error, message);
 }
 
 void ClientConnection::handleSendReceipt(const proto::CommandSendReceipt& sendReceipt) {
@@ -1573,7 +1561,7 @@ void ClientConnection::handlePartitionedMetadataResponse(
                                      << partitionMetadataResponse.request_id()
                                      << " error: " << partitionMetadataResponse.error()
                                      << " msg: " << partitionMetadataResponse.message());
-                checkServerError(partitionMetadataResponse.error());
+                checkServerError(partitionMetadataResponse.error(), partitionMetadataResponse.message());
                 lookupDataPromise->setFailed(
                     getResult(partitionMetadataResponse.error(), partitionMetadataResponse.message()));
             } else {
@@ -1650,7 +1638,7 @@ void ClientConnection::handleLookupTopicRespose(
                 LOG_ERROR(cnxString_ << "Failed lookup req_id: " << lookupTopicResponse.request_id()
                                      << " error: " << lookupTopicResponse.error()
                                      << " msg: " << lookupTopicResponse.message());
-                checkServerError(lookupTopicResponse.error());
+                checkServerError(lookupTopicResponse.error(), lookupTopicResponse.message());
                 lookupDataPromise->setFailed(
                     getResult(lookupTopicResponse.error(), lookupTopicResponse.message()));
             } else {

lib/ClientConnection.h

@@ -404,7 +404,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
 
     friend class PulsarFriend;
 
-    void checkServerError(ServerError error);
+    void checkServerError(ServerError error, const std::string& message);
 
     void handleSendReceipt(const proto::CommandSendReceipt&);
     void handleSendError(const proto::CommandSendError&);

lib/ClientConnectionAdaptor.h

@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+// This file is used to mock ClientConnection's methods
+#pragma once
+
+#include <pulsar/Result.h>
+
+#include "ProtoApiEnums.h"
+#include "PulsarApi.pb.h"
+
+namespace pulsar {
+
+namespace adaptor {
+
+template <typename Connection>
+inline void checkServerError(Connection& connection, ServerError error, const std::string& message) {
+    switch (error) {
+        case proto::ServerError::ServiceNotReady:
+            // There are some typical error messages that should not trigger the
+            // close() of the connection.
+            //   "Namespace bundle ... is being unloaded"
+            //   "KeeperException$..."
+            //   "Failed to acquire ownership for namespace bundle ..."
+            // Before https://github.com/apache/pulsar/pull/21211, the error of the 1st and 2nd messages
+            // is ServiceNotReady. Before https://github.com/apache/pulsar/pull/21993, the error of the 3rd
+            // message is ServiceNotReady.
+            if (message.find("Failed to acquire ownership") == std::string::npos &&
+                message.find("KeeperException") == std::string::npos &&
+                message.find("is being unloaded") == std::string::npos) {
+                connection.close(ResultDisconnected);
+            }
+            break;
+        case proto::ServerError::TooManyRequests:
+            // TODO: Implement maxNumberOfRejectedRequestPerConnection like
+            // https://github.com/apache/pulsar/pull/274
+            connection.close(ResultDisconnected);
+            break;
+        default:
+            break;
+    }
+}
+
+}  // namespace adaptor
+
+}  // namespace pulsar

tests/ConnectionTest.cc

@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ */
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include <vector>
+
+#include "lib/ClientConnection.h"
+#include "lib/ClientConnectionAdaptor.h"
+
+using namespace pulsar;
+
+class MockClientConnection {
+   public:
+    MOCK_METHOD(void, close, (Result));
+
+    void checkServerError(ServerError error, const std::string& message) {
+        ::pulsar::adaptor::checkServerError(*this, error, message);
+    }
+};
+
+// These error messages come from
+// https://github.com/apache/pulsar/blob/a702e5a582eaa8292720f9e25fc2dcf858078813/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java#L334-L351
+static const std::vector<std::string> retryableErrorMessages{
+    "Namespace bundle public/default/0x00000000_0xffffffff is being unloaded",
+    "org.apache.zookeeper.KeeperException$OperationTimeoutException: KeeperErrorCode = OperationTimeout for "
+    "/namespace/public/default/0x00000000_0xffffffff",
+    "Failed to acquire ownership for namespace bundle public/default/0x00000000_0xffffffff"};
+
+TEST(ConnectionTest, testCheckServerError) {
+    MockClientConnection conn;
+    EXPECT_CALL(conn, close(ResultDisconnected)).Times(0);
+    for (auto&& msg : retryableErrorMessages) {
+        conn.checkServerError(pulsar::proto::ServiceNotReady, msg);
+    }
+}