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/11/17 11:47:43 UTC
(pulsar-client-cpp) branch branch-3.4 updated (aee12e2 -> fa8decd)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a change to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
from aee12e2 Fix crash when removing connection from the pool (#347)
new 806698e Log topic lookup result (#351)
new fa8decd Fix bad_weak_ptr when close a ClientConnection during construction (#350)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
lib/BinaryProtoLookupService.cc | 3 ++-
lib/ClientConnection.cc | 21 +++++++++------------
lib/ClientConnection.h | 4 +---
lib/ConnectionPool.cc | 4 ++++
tests/AuthPluginTest.cc | 16 ++++++++++++++++
5 files changed, 32 insertions(+), 16 deletions(-)
(pulsar-client-cpp) 02/02: Fix bad_weak_ptr when close a ClientConnection during construction (#350)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit fa8decd2ec10e01d5c99b83ac5dad14f3a945720
Author: Yunze Xu <xy...@163.com>
AuthorDate: Fri Nov 17 17:51:50 2023 +0800
Fix bad_weak_ptr when close a ClientConnection during construction (#350)
Fixes https://github.com/apache/pulsar-client-cpp/issues/348
Fixes https://github.com/apache/pulsar-client-cpp/issues/349
### Motivation
When `close` is called in `ClientConnection`'s constructor,
`shared_from_this()` will be called, which results in a
`std::bad_weak_ptr` error. This error does not happen before
https://github.com/apache/pulsar-client-cpp/pull/317 because
`shared_from_this()` could only be called when the `producers` or
`consumers` field is not empty.
### Modifications
Throw `ResultAuthenticationError` when there is anything wrong with
authentication in `ClientConnection`'s constructor. Then catch the
result in `ConnectionPool::getConnectionAsync` and returned the failed
future.
In addition, check `authentication_` even for non-TLS URLs. Otherwise,
the null authentication will be used to construct `CommandConnect`.
Add `testInvalidPlugin` and `testTlsConfigError` to verify the changes.
(cherry picked from commit 7bb94f45b917ed30b5302ac93ffa1f1942fc6313)
---
lib/ClientConnection.cc | 21 +++++++++------------
lib/ClientConnection.h | 4 +---
lib/ConnectionPool.cc | 4 ++++
tests/AuthPluginTest.cc | 16 ++++++++++++++++
4 files changed, 30 insertions(+), 15 deletions(-)
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index fbcbe62..4086bdd 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -187,6 +187,12 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
pool_(pool),
poolIndex_(poolIndex) {
LOG_INFO(cnxString_ << "Create ClientConnection, timeout=" << clientConfiguration.getConnectionTimeout());
+ if (!authentication_) {
+ LOG_ERROR("Invalid authentication plugin");
+ throw ResultAuthenticationError;
+ return;
+ }
+
if (clientConfiguration.isUseTls()) {
#if BOOST_VERSION >= 105400
boost::asio::ssl::context ctx(boost::asio::ssl::context::tlsv12_client);
@@ -207,20 +213,13 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
ctx.load_verify_file(trustCertFilePath);
} else {
LOG_ERROR(trustCertFilePath << ": No such trustCertFile");
- close(ResultAuthenticationError, false);
- return;
+ throw ResultAuthenticationError;
}
} else {
ctx.set_default_verify_paths();
}
}
- if (!authentication_) {
- LOG_ERROR("Invalid authentication plugin");
- close(ResultAuthenticationError, false);
- return;
- }
-
std::string tlsCertificates = clientConfiguration.getTlsCertificateFilePath();
std::string tlsPrivateKey = clientConfiguration.getTlsPrivateKeyFilePath();
@@ -231,13 +230,11 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
tlsPrivateKey = authData->getTlsPrivateKey();
if (!file_exists(tlsCertificates)) {
LOG_ERROR(tlsCertificates << ": No such tlsCertificates");
- close(ResultAuthenticationError, false);
- return;
+ throw ResultAuthenticationError;
}
if (!file_exists(tlsCertificates)) {
LOG_ERROR(tlsCertificates << ": No such tlsCertificates");
- close(ResultAuthenticationError, false);
- return;
+ throw ResultAuthenticationError;
}
ctx.use_private_key_file(tlsPrivateKey, boost::asio::ssl::context::pem);
ctx.use_certificate_file(tlsCertificates, boost::asio::ssl::context::pem);
diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h
index 965c6ae..ce2bd88 100644
--- a/lib/ClientConnection.h
+++ b/lib/ClientConnection.h
@@ -148,9 +148,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
* @param result all pending futures will complete with this result
* @param detach remove it from the pool if it's true
*
- * `detach` should only be false when:
- * 1. Before the connection is put into the pool, i.e. during the construction.
- * 2. When the connection pool is closed
+ * `detach` should only be false when the connection pool is closed.
*/
void close(Result result = ResultConnectError, bool detach = true);
diff --git a/lib/ConnectionPool.cc b/lib/ConnectionPool.cc
index b5b65e3..95170a9 100644
--- a/lib/ConnectionPool.cc
+++ b/lib/ConnectionPool.cc
@@ -99,6 +99,10 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(keySuffix),
clientConfiguration_, authentication_, clientVersion_, *this,
keySuffix));
+ } catch (Result result) {
+ Promise<Result, ClientConnectionWeakPtr> promise;
+ promise.setFailed(result);
+ return promise.getFuture();
} catch (const std::runtime_error& e) {
lock.unlock();
LOG_ERROR("Failed to create connection: " << e.what())
diff --git a/tests/AuthPluginTest.cc b/tests/AuthPluginTest.cc
index 1e16019..9fd048b 100644
--- a/tests/AuthPluginTest.cc
+++ b/tests/AuthPluginTest.cc
@@ -581,3 +581,19 @@ TEST(AuthPluginTest, testOauth2Failure) {
ASSERT_EQ(client5.createProducer(topic, producer), ResultAuthenticationError);
client5.close();
}
+
+TEST(AuthPluginTest, testInvalidPlugin) {
+ Client client("pulsar://localhost:6650", ClientConfiguration{}.setAuth(AuthFactory::create("invalid")));
+ Producer producer;
+ ASSERT_EQ(ResultAuthenticationError, client.createProducer("my-topic", producer));
+ client.close();
+}
+
+TEST(AuthPluginTest, testTlsConfigError) {
+ Client client(serviceUrlTls, ClientConfiguration{}
+ .setAuth(AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath))
+ .setTlsTrustCertsFilePath("invalid"));
+ Producer producer;
+ ASSERT_EQ(ResultAuthenticationError, client.createProducer("my-topic", producer));
+ client.close();
+}
(pulsar-client-cpp) 01/02: Log topic lookup result (#351)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 806698ee50431d46a35d12c4ae7fbb913b419c81
Author: erobot <er...@users.noreply.github.com>
AuthorDate: Fri Nov 17 16:03:37 2023 +0800
Log topic lookup result (#351)
(cherry picked from commit c771b12fd564c9d7c351e61fa3ced56a8f15ec1f)
---
lib/BinaryProtoLookupService.cc | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/lib/BinaryProtoLookupService.cc b/lib/BinaryProtoLookupService.cc
index c8d6b76..87f02ea 100644
--- a/lib/BinaryProtoLookupService.cc
+++ b/lib/BinaryProtoLookupService.cc
@@ -83,7 +83,8 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
}
});
} else {
- LOG_DEBUG("Lookup response for " << topic << ", lookup-broker-url " << data->getBrokerUrl());
+ LOG_INFO("Lookup response for " << topic << ", lookup-broker-url " << data->getBrokerUrl()
+ << ", from " << cnx->cnxString());
if (data->shouldProxyThroughServiceUrl()) {
// logicalAddress is the proxy's address, we should still connect through proxy
promise->setValue({responseBrokerAddress, address});