You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2023/02/16 05:09:50 UTC
[pulsar-client-cpp] branch main updated: [improve] Add configuration to limit times of client's lookup redirection. (#129)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 05807bd [improve] Add configuration to limit times of client's lookup redirection. (#129)
05807bd is described below
commit 05807bdaf3a4341b22efd7c71f7b00c47cc31413
Author: Tong <to...@users.noreply.github.com>
AuthorDate: Thu Feb 16 13:09:45 2023 +0800
[improve] Add configuration to limit times of client's lookup redirection. (#129)
---
include/pulsar/ClientConfiguration.h | 39 +++++++++++++++++++++++++
lib/BinaryProtoLookupService.cc | 25 ++++++++++------
lib/BinaryProtoLookupService.h | 17 +++++++----
lib/ClientConfiguration.cc | 21 ++++++++++++++
lib/ClientConfigurationImpl.h | 3 ++
lib/ClientImpl.cc | 5 ++--
lib/ConsumerImpl.cc | 7 +++--
lib/HTTPLookupService.cc | 4 +--
lib/HTTPLookupService.h | 1 +
lib/HandlerBase.h | 15 +++++-----
lib/ProducerImpl.cc | 8 +++---
tests/LookupServiceTest.cc | 55 ++++++++++++++++++++++++++++++++----
12 files changed, 163 insertions(+), 37 deletions(-)
diff --git a/include/pulsar/ClientConfiguration.h b/include/pulsar/ClientConfiguration.h
index 9df92ac..6ac04a0 100644
--- a/include/pulsar/ClientConfiguration.h
+++ b/include/pulsar/ClientConfiguration.h
@@ -116,6 +116,45 @@ class PULSAR_PUBLIC ClientConfiguration {
*/
int getConcurrentLookupRequest() const;
+ /**
+ * Max number of lookup redirection allowed on each look request to prevent overload on broker.
+ * <i>(default: 20)</i>
+ *
+ * @param maxLookupRedirects
+ */
+ ClientConfiguration& setMaxLookupRedirects(int maxLookupRedirects);
+
+ /**
+ * @return Get configured total allowed lookup redirecting.
+ */
+ int getMaxLookupRedirects() const;
+
+ /**
+ * Initial backoff interval in milliseconds.
+ * <i>(default: 100)</i>
+ *
+ * @param initialBackoffIntervalMs
+ */
+ ClientConfiguration& setInitialBackoffIntervalMs(int initialBackoffIntervalMs);
+
+ /**
+ * @return Get initial backoff interval in milliseconds.
+ */
+ int getInitialBackoffIntervalMs() const;
+
+ /**
+ * Max backoff interval in milliseconds.
+ * <i>(default: 60000)</i>
+ *
+ * @param maxBackoffIntervalMs
+ */
+ ClientConfiguration& setMaxBackoffIntervalMs(int maxBackoffIntervalMs);
+
+ /**
+ * @return Get max backoff interval in milliseconds.
+ */
+ int getMaxBackoffIntervalMs() const;
+
/**
* Initialize the log configuration
*
diff --git a/lib/BinaryProtoLookupService.cc b/lib/BinaryProtoLookupService.cc
index 9ce3369..502f7a8 100644
--- a/lib/BinaryProtoLookupService.cc
+++ b/lib/BinaryProtoLookupService.cc
@@ -30,17 +30,26 @@ DECLARE_LOG_OBJECT()
namespace pulsar {
auto BinaryProtoLookupService::getBroker(const TopicName& topicName) -> LookupResultFuture {
- return findBroker(serviceNameResolver_.resolveHost(), false, topicName.toString());
+ return findBroker(serviceNameResolver_.resolveHost(), false, topicName.toString(), 0);
}
auto BinaryProtoLookupService::findBroker(const std::string& address, bool authoritative,
- const std::string& topic) -> LookupResultFuture {
- LOG_DEBUG("find broker from " << address << ", authoritative: " << authoritative << ", topic: " << topic);
+ const std::string& topic, size_t redirectCount)
+ -> LookupResultFuture {
+ LOG_DEBUG("find broker from " << address << ", authoritative: " << authoritative << ", topic: " << topic
+ << ", redirect count: " << redirectCount);
auto promise = std::make_shared<Promise<Result, LookupResult>>();
+ if (maxLookupRedirects_ > 0 && redirectCount > maxLookupRedirects_) {
+ LOG_ERROR("Too many lookup request redirects on topic " << topic << ", configured limit is "
+ << maxLookupRedirects_);
+ promise->setFailed(ResultTooManyLookupRequestException);
+ return promise->getFuture();
+ }
+
// NOTE: we can use move capture for topic since C++14
- cnxPool_.getConnectionAsync(address).addListener([this, promise, topic, address, authoritative](
- Result result,
- const ClientConnectionWeakPtr& weakCnx) {
+ cnxPool_.getConnectionAsync(address).addListener([this, promise, topic, address, authoritative,
+ redirectCount](Result result,
+ const ClientConnectionWeakPtr& weakCnx) {
if (result != ResultOk) {
promise->setFailed(result);
return;
@@ -53,7 +62,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
}
auto lookupPromise = std::make_shared<LookupDataResultPromise>();
cnx->newTopicLookup(topic, authoritative, listenerName_, newRequestId(), lookupPromise);
- lookupPromise->getFuture().addListener([this, cnx, promise, topic, address](
+ lookupPromise->getFuture().addListener([this, cnx, promise, topic, address, redirectCount](
Result result, const LookupDataResultPtr& data) {
if (result != ResultOk || !data) {
LOG_ERROR("Lookup failed for " << topic << ", result " << result);
@@ -65,7 +74,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
(serviceNameResolver_.useTls() ? data->getBrokerUrlTls() : data->getBrokerUrl());
if (data->isRedirect()) {
LOG_DEBUG("Lookup request is for " << topic << " redirected to " << responseBrokerAddress);
- findBroker(responseBrokerAddress, data->isAuthoritative(), topic)
+ findBroker(responseBrokerAddress, data->isAuthoritative(), topic, redirectCount + 1)
.addListener([promise](Result result, const LookupResult& value) {
if (result == ResultOk) {
promise->setValue(value);
diff --git a/lib/BinaryProtoLookupService.h b/lib/BinaryProtoLookupService.h
index 9740fb1..cfac023 100644
--- a/lib/BinaryProtoLookupService.h
+++ b/lib/BinaryProtoLookupService.h
@@ -20,6 +20,7 @@
#define _PULSAR_BINARY_LOOKUP_SERVICE_HEADER_
#include <pulsar/Authentication.h>
+#include <pulsar/ClientConfiguration.h>
#include <pulsar/Schema.h>
#include <mutex>
@@ -38,8 +39,11 @@ using GetSchemaPromisePtr = std::shared_ptr<Promise<Result, boost::optional<Sche
class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
public:
BinaryProtoLookupService(ServiceNameResolver& serviceNameResolver, ConnectionPool& pool,
- const std::string& listenerName)
- : serviceNameResolver_(serviceNameResolver), cnxPool_(pool), listenerName_(listenerName) {}
+ const ClientConfiguration& clientConfiguration)
+ : serviceNameResolver_(serviceNameResolver),
+ cnxPool_(pool),
+ listenerName_(clientConfiguration.getListenerName()),
+ maxLookupRedirects_(clientConfiguration.getMaxLookupRedirects()) {}
LookupResultFuture getBroker(const TopicName& topicName) override;
@@ -49,6 +53,11 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
Future<Result, boost::optional<SchemaInfo>> getSchema(const TopicNamePtr& topicName) override;
+ protected:
+ // Mark findBroker as protected to make it accessible from test.
+ LookupResultFuture findBroker(const std::string& address, bool authoritative, const std::string& topic,
+ size_t redirectCount);
+
private:
std::mutex mutex_;
uint64_t requestIdGenerator_ = 0;
@@ -56,9 +65,7 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
ServiceNameResolver& serviceNameResolver_;
ConnectionPool& cnxPool_;
std::string listenerName_;
-
- // TODO: limit the redirect count, see https://github.com/apache/pulsar/pull/7096
- LookupResultFuture findBroker(const std::string& address, bool authoritative, const std::string& topic);
+ const int32_t maxLookupRedirects_;
void sendPartitionMetadataLookupRequest(const std::string& topicName, Result result,
const ClientConnectionWeakPtr& clientCnx,
diff --git a/lib/ClientConfiguration.cc b/lib/ClientConfiguration.cc
index 70d85cb..703f3aa 100644
--- a/lib/ClientConfiguration.cc
+++ b/lib/ClientConfiguration.cc
@@ -123,6 +123,27 @@ ClientConfiguration& ClientConfiguration::setConcurrentLookupRequest(int concurr
int ClientConfiguration::getConcurrentLookupRequest() const { return impl_->concurrentLookupRequest; }
+ClientConfiguration& ClientConfiguration::setMaxLookupRedirects(int maxLookupRedirects) {
+ impl_->maxLookupRedirects = maxLookupRedirects;
+ return *this;
+}
+
+int ClientConfiguration::getMaxLookupRedirects() const { return impl_->maxLookupRedirects; }
+
+ClientConfiguration& ClientConfiguration::setInitialBackoffIntervalMs(int initialBackoffIntervalMs) {
+ impl_->initialBackoffIntervalMs = initialBackoffIntervalMs;
+ return *this;
+}
+
+int ClientConfiguration::getInitialBackoffIntervalMs() const { return impl_->initialBackoffIntervalMs; }
+
+ClientConfiguration& ClientConfiguration::setMaxBackoffIntervalMs(int maxBackoffIntervalMs) {
+ impl_->maxBackoffIntervalMs = maxBackoffIntervalMs;
+ return *this;
+}
+
+int ClientConfiguration::getMaxBackoffIntervalMs() const { return impl_->maxBackoffIntervalMs; }
+
ClientConfiguration& ClientConfiguration::setLogConfFilePath(const std::string& logConfFilePath) {
impl_->logConfFilePath = logConfFilePath;
return *this;
diff --git a/lib/ClientConfigurationImpl.h b/lib/ClientConfigurationImpl.h
index f6f2a85..4785667 100644
--- a/lib/ClientConfigurationImpl.h
+++ b/lib/ClientConfigurationImpl.h
@@ -30,6 +30,9 @@ struct ClientConfigurationImpl {
int operationTimeoutSeconds{30};
int messageListenerThreads{1};
int concurrentLookupRequest{50000};
+ int maxLookupRedirects{20};
+ int initialBackoffIntervalMs{100};
+ int maxBackoffIntervalMs{60000};
std::string logConfFilePath;
bool useTls{false};
std::string tlsPrivateKeyFilePath;
diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc
index 14a4047..7575b28 100644
--- a/lib/ClientImpl.cc
+++ b/lib/ClientImpl.cc
@@ -116,9 +116,8 @@ ClientImpl::ClientImpl(const std::string& serviceUrl, const ClientConfiguration&
std::cref(clientConfiguration_.getAuthPtr()));
} else {
LOG_DEBUG("Using Binary Lookup");
- underlyingLookupServicePtr =
- std::make_shared<BinaryProtoLookupService>(std::ref(serviceNameResolver_), std::ref(pool_),
- std::cref(clientConfiguration_.getListenerName()));
+ underlyingLookupServicePtr = std::make_shared<BinaryProtoLookupService>(
+ std::ref(serviceNameResolver_), std::ref(pool_), std::cref(clientConfiguration_));
}
lookupServicePtr_ = RetryableLookupService::create(
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index 1703d2a..c63147f 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -62,8 +62,11 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
const ConsumerTopicType consumerTopicType /* = NonPartitioned by default */,
Commands::SubscriptionMode subscriptionMode,
boost::optional<MessageId> startMessageId)
- : ConsumerImplBase(client, topic, Backoff(milliseconds(100), seconds(60), milliseconds(0)), conf,
- listenerExecutor ? listenerExecutor : client->getListenerExecutorProvider()->get()),
+ : ConsumerImplBase(
+ client, topic,
+ Backoff(milliseconds(client->getClientConfig().getInitialBackoffIntervalMs()),
+ milliseconds(client->getClientConfig().getMaxBackoffIntervalMs()), milliseconds(0)),
+ conf, listenerExecutor ? listenerExecutor : client->getListenerExecutorProvider()->get()),
waitingForZeroQueueSizeMessage(false),
config_(conf),
subscription_(subscriptionName),
diff --git a/lib/HTTPLookupService.cc b/lib/HTTPLookupService.cc
index ae719a5..0ce6839 100644
--- a/lib/HTTPLookupService.cc
+++ b/lib/HTTPLookupService.cc
@@ -42,7 +42,6 @@ const static std::string V2_PATH = "/lookup/v2/topic/";
const static std::string ADMIN_PATH_V1 = "/admin/";
const static std::string ADMIN_PATH_V2 = "/admin/v2/";
-const static int MAX_HTTP_REDIRECTS = 20;
const static std::string PARTITION_METHOD_NAME = "partitions";
const static int NUMBER_OF_LOOKUP_THREADS = 1;
@@ -63,6 +62,7 @@ HTTPLookupService::HTTPLookupService(ServiceNameResolver &serviceNameResolver,
serviceNameResolver_(serviceNameResolver),
authenticationPtr_(authData),
lookupTimeoutInSeconds_(clientConfiguration.getOperationTimeoutSeconds()),
+ maxLookupRedirects_(clientConfiguration.getMaxLookupRedirects()),
tlsPrivateFilePath_(clientConfiguration.getTlsPrivateKeyFilePath()),
tlsCertificateFilePath_(clientConfiguration.getTlsCertificateFilePath()),
tlsTrustCertsFilePath_(clientConfiguration.getTlsTrustCertsFilePath()),
@@ -189,7 +189,7 @@ Result HTTPLookupService::sendHTTPRequest(std::string completeUrl, std::string &
long &responseCode) {
uint16_t reqCount = 0;
Result retResult = ResultOk;
- while (++reqCount <= MAX_HTTP_REDIRECTS) {
+ while (++reqCount <= maxLookupRedirects_) {
CURL *handle;
CURLcode res;
std::string version = std::string("Pulsar-CPP-v") + PULSAR_VERSION_STR;
diff --git a/lib/HTTPLookupService.h b/lib/HTTPLookupService.h
index f76d3e6..9135677 100644
--- a/lib/HTTPLookupService.h
+++ b/lib/HTTPLookupService.h
@@ -50,6 +50,7 @@ class HTTPLookupService : public LookupService, public std::enable_shared_from_t
ServiceNameResolver& serviceNameResolver_;
AuthenticationPtr authenticationPtr_;
int lookupTimeoutInSeconds_;
+ const int maxLookupRedirects_;
std::string tlsPrivateFilePath_;
std::string tlsCertificateFilePath_;
std::string tlsTrustCertsFilePath_;
diff --git a/lib/HandlerBase.h b/lib/HandlerBase.h
index 6faaebd..0cc8968 100644
--- a/lib/HandlerBase.h
+++ b/lib/HandlerBase.h
@@ -113,13 +113,14 @@ class HandlerBase {
enum State
{
- NotStarted,
- Pending,
- Ready,
- Closing,
- Closed,
- Failed,
- Producer_Fenced
+ NotStarted, // Not initialized, in Java client: HandlerState.State.Uninitialized
+ Pending, // Client connecting to broker, in Java client: HandlerState.State.Connecting
+ Ready, // Handler is being used, in Java client: HandlerState.State.Ready
+ Closing, // Close cmd has been sent to broker, in Java client: HandlerState.State.Closing
+ Closed, // Broker acked the close, in Java client: HandlerState.State.Closed
+ Failed, // Handler is failed, in Java client: HandlerState.State.Failed
+ Producer_Fenced, // The producer has been fenced by the broker
+ // in Java client: HandlerState.State.ProducerFenced
};
std::atomic<State> state_;
diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
index a319946..6514b3b 100644
--- a/lib/ProducerImpl.cc
+++ b/lib/ProducerImpl.cc
@@ -56,9 +56,10 @@ struct ProducerImpl::PendingCallbacks {
ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName,
const ProducerConfiguration& conf, int32_t partition)
- : HandlerBase(
- client, (partition < 0) ? topicName.toString() : topicName.getTopicPartitionName(partition),
- Backoff(milliseconds(100), seconds(60), milliseconds(std::max(100, conf.getSendTimeout() - 100)))),
+ : HandlerBase(client, (partition < 0) ? topicName.toString() : topicName.getTopicPartitionName(partition),
+ Backoff(milliseconds(client->getClientConfig().getInitialBackoffIntervalMs()),
+ milliseconds(client->getClientConfig().getMaxBackoffIntervalMs()),
+ milliseconds(std::max(100, conf.getSendTimeout() - 100)))),
conf_(conf),
semaphore_(),
pendingMessagesQueue_(),
@@ -978,4 +979,3 @@ void ProducerImpl::asyncWaitSendTimeout(DurationType expiryTime) {
ProducerImplWeakPtr ProducerImpl::weak_from_this() noexcept { return shared_from_this(); }
} // namespace pulsar
-/* namespace pulsar */
diff --git a/tests/LookupServiceTest.cc b/tests/LookupServiceTest.cc
index e31246f..9467727 100644
--- a/tests/LookupServiceTest.cc
+++ b/tests/LookupServiceTest.cc
@@ -50,7 +50,7 @@ TEST(LookupServiceTest, basicLookup) {
ExecutorServiceProviderPtr ioExecutorProvider_(std::make_shared<ExecutorServiceProvider>(1));
ConnectionPool pool_(conf, ioExecutorProvider_, authData, true);
ServiceNameResolver serviceNameResolver(url);
- BinaryProtoLookupService lookupService(serviceNameResolver, pool_, "");
+ BinaryProtoLookupService lookupService(serviceNameResolver, pool_, conf);
TopicNamePtr topicName = TopicName::get("topic");
@@ -115,7 +115,7 @@ TEST(LookupServiceTest, basicGetNamespaceTopics) {
ExecutorServiceProviderPtr ioExecutorProvider_(std::make_shared<ExecutorServiceProvider>(1));
ConnectionPool pool_(conf, ioExecutorProvider_, authData, true);
ServiceNameResolver serviceNameResolver(url);
- BinaryProtoLookupService lookupService(serviceNameResolver, pool_, "");
+ BinaryProtoLookupService lookupService(serviceNameResolver, pool_, conf);
TopicNamePtr topicName = TopicName::get(topicName1);
NamespaceNamePtr nsName = topicName->getNamespaceName();
@@ -177,7 +177,8 @@ static void testMultiAddresses(LookupService& lookupService) {
TEST(LookupServiceTest, testMultiAddresses) {
ConnectionPool pool({}, std::make_shared<ExecutorServiceProvider>(1), AuthFactory::Disabled(), true);
ServiceNameResolver serviceNameResolver("pulsar://localhost,localhost:9999");
- BinaryProtoLookupService binaryLookupService(serviceNameResolver, pool, "");
+ ClientConfiguration conf;
+ BinaryProtoLookupService binaryLookupService(serviceNameResolver, pool, conf);
testMultiAddresses(binaryLookupService);
// HTTPLookupService calls shared_from_this() internally, we must create a shared pointer to test
@@ -190,9 +191,10 @@ TEST(LookupServiceTest, testRetry) {
auto executorProvider = std::make_shared<ExecutorServiceProvider>(1);
ConnectionPool pool({}, executorProvider, AuthFactory::Disabled(), true);
ServiceNameResolver serviceNameResolver("pulsar://localhost:9999,localhost");
+ ClientConfiguration conf;
auto lookupService = RetryableLookupService::create(
- std::make_shared<BinaryProtoLookupService>(serviceNameResolver, pool, ""), 30 /* seconds */,
+ std::make_shared<BinaryProtoLookupService>(serviceNameResolver, pool, conf), 30 /* seconds */,
executorProvider);
PulsarFriend::setServiceUrlIndex(serviceNameResolver, 0);
@@ -240,10 +242,11 @@ TEST(LookupServiceTest, testTimeout) {
auto executorProvider = std::make_shared<ExecutorServiceProvider>(1);
ConnectionPool pool({}, executorProvider, AuthFactory::Disabled(), true);
ServiceNameResolver serviceNameResolver("pulsar://localhost:9990,localhost:9902,localhost:9904");
+ ClientConfiguration conf;
constexpr int timeoutInSeconds = 2;
auto lookupService = RetryableLookupService::create(
- std::make_shared<BinaryProtoLookupService>(serviceNameResolver, pool, ""), timeoutInSeconds,
+ std::make_shared<BinaryProtoLookupService>(serviceNameResolver, pool, conf), timeoutInSeconds,
executorProvider);
auto topicNamePtr = TopicName::get("lookup-service-test-retry");
@@ -355,4 +358,44 @@ TEST_P(LookupServiceTest, testGetKeyValueSchema) {
ASSERT_FALSE(schemaInfo->getProperties().empty());
}
-INSTANTIATE_TEST_CASE_P(Pulsar, LookupServiceTest, ::testing::Values(binaryLookupUrl, httpLookupUrl));
\ No newline at end of file
+INSTANTIATE_TEST_CASE_P(Pulsar, LookupServiceTest, ::testing::Values(binaryLookupUrl, httpLookupUrl));
+
+class BinaryProtoLookupServiceRedirectTestHelper : public BinaryProtoLookupService {
+ public:
+ BinaryProtoLookupServiceRedirectTestHelper(ServiceNameResolver& serviceNameResolver, ConnectionPool& pool,
+ const ClientConfiguration& clientConfiguration)
+ : BinaryProtoLookupService(serviceNameResolver, pool, clientConfiguration) {}
+
+ LookupResultFuture findBroker(const std::string& address, bool authoritative, const std::string& topic,
+ size_t redirectCount) {
+ return BinaryProtoLookupService::findBroker(address, authoritative, topic, redirectCount);
+ }
+}; // class BinaryProtoLookupServiceRedirectTestHelper
+
+TEST(LookupServiceTest, testRedirectionLimit) {
+ const auto redirect_limit = 5;
+ AuthenticationPtr authData = AuthFactory::Disabled();
+ ClientConfiguration conf;
+ conf.setMaxLookupRedirects(redirect_limit);
+ ExecutorServiceProviderPtr ioExecutorProvider_(std::make_shared<ExecutorServiceProvider>(1));
+ ConnectionPool pool_(conf, ioExecutorProvider_, authData, true);
+ std::string url = "pulsar://localhost:6650";
+ ServiceNameResolver serviceNameResolver(url);
+ BinaryProtoLookupServiceRedirectTestHelper lookupService(serviceNameResolver, pool_, conf);
+
+ const auto topicNamePtr = TopicName::get("topic");
+ for (auto idx = 0; idx < redirect_limit + 5; ++idx) {
+ auto future =
+ lookupService.findBroker(serviceNameResolver.resolveHost(), false, topicNamePtr->toString(), idx);
+ LookupService::LookupResult lookupResult;
+ auto result = future.get(lookupResult);
+
+ if (idx <= redirect_limit) {
+ ASSERT_EQ(ResultOk, result);
+ ASSERT_EQ(url, lookupResult.logicalAddress);
+ ASSERT_EQ(url, lookupResult.physicalAddress);
+ } else {
+ ASSERT_EQ(ResultTooManyLookupRequestException, result);
+ }
+ }
+}