forked from apache/pulsar-client-cpp
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Do not close the socket when the broker failed due to MetadataStoreEx…
…ception ### 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.
- Loading branch information
1 parent
d1dd08b
commit b6329e8
Showing
4 changed files
with
115 additions
and
16 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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); | ||
| } | ||
| } |