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 06:28:02 UTC
(pulsar-client-cpp) branch branch-3.4 updated (f337eff -> aee12e2)
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 f337eff Fix ProducerBusy or ConsumerBusy error when configuring multiple brokers per connection (#337)
new 39d39a9 Use absolute path to find credential files in unit tests (#340)
new 1674a08 Fix close() returns ResultAlreadyClosed after unsubscribe or close (#338)
new 8408e65 Fix lazy partitioned producer might send duplicated messages (#342)
new aee12e2 Fix crash when removing connection from the pool (#347)
The 4 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:
CMakeLists.txt | 2 ++
lib/ClientConnection.cc | 9 +++++----
lib/ClientConnection.h | 4 +++-
lib/ConnectionPool.cc | 7 ++++---
lib/ConsumerImpl.cc | 11 ++++++----
lib/PartitionedProducerImpl.cc | 22 +++++++++++++++++---
lib/ProducerImpl.cc | 8 ++------
lib/ProducerImpl.h | 7 +++++--
pulsar-test-service-start.sh | 4 ++--
pulsar-test-service-stop.sh | 4 ++--
tests/AuthBasicTest.cc | 10 +++++++---
tests/AuthPluginTest.cc | 20 +++++++++++++------
tests/AuthTokenTest.cc | 6 +++++-
tests/BasicEndToEndTest.cc | 30 ++++++++++++++++------------
tests/CMakeLists.txt | 6 +++---
tests/ClientTest.cc | 41 ++++++++++++++++++++++++++++++++++++++
tests/ConsumerConfigurationTest.cc | 2 +-
tests/ConsumerTest.cc | 33 ++++++++++++++++++++++++++++--
tests/PulsarFriend.h | 2 ++
tests/c/c_BasicEndToEndTest.cc | 2 +-
tests/chunkdedup/ChunkDedupTest.cc | 26 ++++++++++++++++++++++++
tests/oauth2/Oauth2Test.cc | 6 +++---
22 files changed, 202 insertions(+), 60 deletions(-)
(pulsar-client-cpp) 02/04: Fix close() returns ResultAlreadyClosed after unsubscribe or close (#338)
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 1674a081587f11ce4b3ffc20401e3d411573d201
Author: Yunze Xu <xy...@163.com>
AuthorDate: Fri Nov 10 10:54:58 2023 +0800
Fix close() returns ResultAlreadyClosed after unsubscribe or close (#338)
Fixes https://github.com/apache/pulsar-client-cpp/issues/88
### Motivation
When `close` is called if the consumer has already called `unsubscribe`
or `close`, it should not fail. See
https://github.com/apache/pulsar/blob/428c18c8d0c3d135189920740192982e11ffb2bf/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1034
### Modifications
Use the same close logic with Java client.
Add `testCloseAgainBeforeCloseDone` and `testCloseAfterUnsubscribe` to
verify the new behaviors of `Consumer::close`.
(cherry picked from commit a8402da957143b322364c048a8b19dc790891724)
---
lib/ConsumerImpl.cc | 11 +++++++----
tests/BasicEndToEndTest.cc | 14 +++++++-------
tests/ConsumerConfigurationTest.cc | 2 +-
tests/ConsumerTest.cc | 25 +++++++++++++++++++++++++
tests/c/c_BasicEndToEndTest.cc | 2 +-
5 files changed, 41 insertions(+), 13 deletions(-)
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index 8dd334e..b466683 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -1238,10 +1238,12 @@ void ConsumerImpl::disconnectConsumer() {
}
void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
- auto callback = [this, originalCallback](Result result) {
+ auto callback = [this, originalCallback](Result result, bool alreadyClosed = false) {
shutdown();
if (result == ResultOk) {
- LOG_INFO(getName() << "Closed consumer " << consumerId_);
+ if (!alreadyClosed) {
+ LOG_INFO(getName() << "Closed consumer " << consumerId_);
+ }
} else {
LOG_WARN(getName() << "Failed to close consumer: " << result);
}
@@ -1250,8 +1252,9 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
}
};
- if (state_ != Ready) {
- callback(ResultAlreadyClosed);
+ auto state = state_.load();
+ if (state == Closing || state == Closed) {
+ callback(ResultOk, true);
return;
}
diff --git a/tests/BasicEndToEndTest.cc b/tests/BasicEndToEndTest.cc
index 0951dd4..e2c6697 100644
--- a/tests/BasicEndToEndTest.cc
+++ b/tests/BasicEndToEndTest.cc
@@ -244,7 +244,7 @@ TEST(BasicEndToEndTest, testProduceConsume) {
consumer.receive(receivedMsg);
ASSERT_EQ(content, receivedMsg.getDataAsString());
ASSERT_EQ(ResultOk, consumer.unsubscribe());
- ASSERT_EQ(ResultAlreadyClosed, consumer.close());
+ ASSERT_EQ(ResultOk, consumer.close());
ASSERT_EQ(ResultOk, producer.close());
ASSERT_EQ(ResultOk, client.close());
}
@@ -405,7 +405,7 @@ TEST(BasicEndToEndTest, testMultipleClientsMultipleSubscriptions) {
ASSERT_EQ(ResultOk, producer1.close());
ASSERT_EQ(ResultOk, consumer1.close());
- ASSERT_EQ(ResultAlreadyClosed, consumer1.close());
+ ASSERT_EQ(ResultOk, consumer1.close());
ASSERT_EQ(ResultConsumerNotInitialized, consumer2.close());
ASSERT_EQ(ResultOk, client1.close());
@@ -637,7 +637,7 @@ TEST(BasicEndToEndTest, testCompressionLZ4) {
ASSERT_EQ(content2, receivedMsg.getDataAsString());
ASSERT_EQ(ResultOk, consumer.unsubscribe());
- ASSERT_EQ(ResultAlreadyClosed, consumer.close());
+ ASSERT_EQ(ResultOk, consumer.close());
ASSERT_EQ(ResultOk, producer.close());
ASSERT_EQ(ResultOk, client.close());
}
@@ -675,7 +675,7 @@ TEST(BasicEndToEndTest, testCompressionZLib) {
ASSERT_EQ(content2, receivedMsg.getDataAsString());
ASSERT_EQ(ResultOk, consumer.unsubscribe());
- ASSERT_EQ(ResultAlreadyClosed, consumer.close());
+ ASSERT_EQ(ResultOk, consumer.close());
ASSERT_EQ(ResultOk, producer.close());
ASSERT_EQ(ResultOk, client.close());
}
@@ -750,7 +750,7 @@ TEST(BasicEndToEndTest, testConsumerClose) {
Consumer consumer;
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
ASSERT_EQ(consumer.close(), ResultOk);
- ASSERT_EQ(consumer.close(), ResultAlreadyClosed);
+ ASSERT_EQ(consumer.close(), ResultOk);
}
TEST(BasicEndToEndTest, testDuplicateConsumerCreationOnPartitionedTopic) {
@@ -1398,7 +1398,7 @@ TEST(BasicEndToEndTest, testRSAEncryption) {
}
ASSERT_EQ(ResultOk, consumer.unsubscribe());
- ASSERT_EQ(ResultAlreadyClosed, consumer.close());
+ ASSERT_EQ(ResultOk, consumer.close());
ASSERT_EQ(ResultOk, producer.close());
}
ASSERT_EQ(ResultOk, client.close());
@@ -1617,7 +1617,7 @@ TEST(BasicEndToEndTest, testSeek) {
ASSERT_EQ(expected.str(), msgReceived.getDataAsString());
ASSERT_EQ(ResultOk, consumer.acknowledge(msgReceived));
ASSERT_EQ(ResultOk, consumer.unsubscribe());
- ASSERT_EQ(ResultAlreadyClosed, consumer.close());
+ ASSERT_EQ(ResultOk, consumer.close());
ASSERT_EQ(ResultOk, producer.close());
ASSERT_EQ(ResultOk, client.close());
}
diff --git a/tests/ConsumerConfigurationTest.cc b/tests/ConsumerConfigurationTest.cc
index f378501..482f0fc 100644
--- a/tests/ConsumerConfigurationTest.cc
+++ b/tests/ConsumerConfigurationTest.cc
@@ -314,7 +314,7 @@ TEST(ConsumerConfigurationTest, testSubscriptionInitialPosition) {
ASSERT_EQ(content1, receivedMsg.getDataAsString());
ASSERT_EQ(ResultOk, consumer.unsubscribe());
- ASSERT_EQ(ResultAlreadyClosed, consumer.close());
+ ASSERT_EQ(ResultOk, consumer.close());
ASSERT_EQ(ResultOk, producer.close());
ASSERT_EQ(ResultOk, client.close());
}
diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc
index 80c0a71..0836fbf 100644
--- a/tests/ConsumerTest.cc
+++ b/tests/ConsumerTest.cc
@@ -1405,4 +1405,29 @@ TEST(ConsumerTest, testNoListenerThreadBlocking) {
client.close();
}
+TEST(ConsumerTest, testCloseAfterUnsubscribe) {
+ Client client{lookupUrl};
+ Consumer consumer;
+ ASSERT_EQ(ResultOk, client.subscribe("test-close-after-unsubscribe", "sub", consumer));
+ ASSERT_EQ(ResultOk, consumer.unsubscribe());
+ ASSERT_EQ(ResultOk, consumer.close());
+}
+
+TEST(ConsumerTest, testCloseAgainBeforeCloseDone) {
+ Client client{lookupUrl};
+ Consumer consumer;
+ ASSERT_EQ(ResultOk, client.subscribe("test-close-again-before-close-done", "sub", consumer));
+ auto done = std::make_shared<std::atomic_bool>(false);
+ auto result = std::make_shared<std::atomic<Result>>(ResultOk);
+ consumer.closeAsync([done, result](Result innerResult) {
+ result->store(innerResult);
+ done->store(true);
+ });
+ ASSERT_EQ(ResultOk, consumer.close());
+ ASSERT_FALSE(*done);
+ waitUntil(std::chrono::seconds(3), [done] { return done->load(); });
+ ASSERT_EQ(ResultOk, *result);
+ ASSERT_TRUE(*done);
+}
+
} // namespace pulsar
diff --git a/tests/c/c_BasicEndToEndTest.cc b/tests/c/c_BasicEndToEndTest.cc
index 04aa1dc..e3ff19a 100644
--- a/tests/c/c_BasicEndToEndTest.cc
+++ b/tests/c/c_BasicEndToEndTest.cc
@@ -109,7 +109,7 @@ TEST(c_BasicEndToEndTest, testAsyncProduceConsume) {
delete receive_ctx.data;
ASSERT_EQ(pulsar_result_Ok, pulsar_consumer_unsubscribe(consumer));
- ASSERT_EQ(pulsar_result_AlreadyClosed, pulsar_consumer_close(consumer));
+ ASSERT_EQ(pulsar_result_Ok, pulsar_consumer_close(consumer));
ASSERT_EQ(pulsar_result_Ok, pulsar_producer_close(producer));
ASSERT_EQ(pulsar_result_Ok, pulsar_client_close(client));
(pulsar-client-cpp) 03/04: Fix lazy partitioned producer might send duplicated messages (#342)
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 8408e657bf0ae1f34489da25e9aa3de4688bfd6f
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Nov 13 10:02:01 2023 +0800
Fix lazy partitioned producer might send duplicated messages (#342)
Fixes https://github.com/apache/pulsar-client-cpp/issues/341
### Motivation
When a lazy partitioned producer sends two messages, the flow is:
1. `start` is called to grab the connection via `grab()`.
2. Generate 0 as the sequence id of the 1st message.
3. Add the 1st message into the queuea.
4. The connection is established, `msgSequenceGenerator_` is reset from
1 to 0.
5. When sending the 2nd message, 0 is also generated as the sequence id.
Then two messages have the same sequence id.
### Modifications
For lazy partitioned producers, if the internal producer is not started,
sending the message in the callback of its future.
Add `ChunkDedupTest#testLazyPartitionedProducer` to verify it since only
the `tests/chunkdedup/docker-compose.yml` enables the deduplication.
(cherry picked from commit bb16f24bb68699c15eddba3163cfbef0b1282ebf)
---
lib/PartitionedProducerImpl.cc | 22 +++++++++++++++++++---
lib/ProducerImpl.cc | 8 ++------
lib/ProducerImpl.h | 7 +++++--
tests/CMakeLists.txt | 2 +-
tests/chunkdedup/ChunkDedupTest.cc | 26 ++++++++++++++++++++++++++
5 files changed, 53 insertions(+), 12 deletions(-)
diff --git a/lib/PartitionedProducerImpl.cc b/lib/PartitionedProducerImpl.cc
index f7957d6..a799d5c 100644
--- a/lib/PartitionedProducerImpl.cc
+++ b/lib/PartitionedProducerImpl.cc
@@ -198,7 +198,9 @@ void PartitionedProducerImpl::createLazyPartitionProducer(unsigned int partition
// override
void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callback) {
if (state_ != Ready) {
- callback(ResultAlreadyClosed, msg.getMessageId());
+ if (callback) {
+ callback(ResultAlreadyClosed, msg.getMessageId());
+ }
return;
}
@@ -209,7 +211,9 @@ void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callbac
LOG_ERROR("Got Invalid Partition for message from Router Policy, Partition - " << partition);
// change me: abort or notify failure in callback?
// change to appropriate error if callback
- callback(ResultUnknownError, msg.getMessageId());
+ if (callback) {
+ callback(ResultUnknownError, msg.getMessageId());
+ }
return;
}
// find a producer for that partition, index should start from 0
@@ -223,7 +227,19 @@ void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callbac
producersLock.unlock();
// send message on that partition
- producer->sendAsync(msg, callback);
+ if (!conf_.getLazyStartPartitionedProducers() || producer->ready()) {
+ producer->sendAsync(msg, std::move(callback));
+ } else {
+ // Wrapping the callback into a lambda has overhead, so we check if the producer is ready first
+ producer->getProducerCreatedFuture().addListener(
+ [msg, callback](Result result, ProducerImplBaseWeakPtr weakProducer) {
+ if (result == ResultOk) {
+ weakProducer.lock()->sendAsync(msg, std::move(callback));
+ } else if (callback) {
+ callback(result, {});
+ }
+ });
+ }
}
// override
diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
index 80ee735..76a999a 100644
--- a/lib/ProducerImpl.cc
+++ b/lib/ProducerImpl.cc
@@ -60,8 +60,9 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName,
userProvidedProducerName_(false),
producerStr_("[" + topic() + ", " + producerName_ + "] "),
producerId_(client->newProducerId()),
- msgSequenceGenerator_(0),
batchTimer_(executor_->createDeadlineTimer()),
+ lastSequenceIdPublished_(conf.getInitialSequenceId()),
+ msgSequenceGenerator_(lastSequenceIdPublished_ + 1),
sendTimer_(executor_->createDeadlineTimer()),
dataKeyRefreshTask_(*executor_, 4 * 60 * 60 * 1000),
memoryLimitController_(client->getMemoryLimitController()),
@@ -69,11 +70,6 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName,
interceptors_(interceptors) {
LOG_DEBUG("ProducerName - " << producerName_ << " Created producer on topic " << topic()
<< " id: " << producerId_);
-
- int64_t initialSequenceId = conf.getInitialSequenceId();
- lastSequenceIdPublished_ = initialSequenceId;
- msgSequenceGenerator_ = initialSequenceId + 1;
-
if (!producerName_.empty()) {
userProvidedProducerName_ = true;
}
diff --git a/lib/ProducerImpl.h b/lib/ProducerImpl.h
index 91b9544..2fb0b88 100644
--- a/lib/ProducerImpl.h
+++ b/lib/ProducerImpl.h
@@ -19,6 +19,7 @@
#ifndef LIB_PRODUCERIMPL_H_
#define LIB_PRODUCERIMPL_H_
+#include <atomic>
#include <boost/optional.hpp>
#include <list>
#include <memory>
@@ -103,6 +104,8 @@ class ProducerImpl : public HandlerBase, public ProducerImplBase {
ProducerImplWeakPtr weak_from_this() noexcept { return shared_from_this(); }
+ bool ready() const { return producerCreatedPromise_.isComplete(); }
+
protected:
ProducerStatsBasePtr producerStatsBasePtr_;
@@ -169,13 +172,13 @@ class ProducerImpl : public HandlerBase, public ProducerImplBase {
bool userProvidedProducerName_;
std::string producerStr_;
uint64_t producerId_;
- int64_t msgSequenceGenerator_;
std::unique_ptr<BatchMessageContainerBase> batchMessageContainer_;
DeadlineTimerPtr batchTimer_;
PendingFailures batchMessageAndSend(const FlushCallback& flushCallback = nullptr);
- volatile int64_t lastSequenceIdPublished_;
+ std::atomic<int64_t> lastSequenceIdPublished_;
+ std::atomic<int64_t> msgSequenceGenerator_;
std::string schemaVersion_;
DeadlineTimerPtr sendTimer_;
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 993c2fd..3bc5a15 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -72,5 +72,5 @@ add_executable(Oauth2Test oauth2/Oauth2Test.cc)
target_compile_options(Oauth2Test PRIVATE -DTEST_CONF_DIR="${TEST_CONF_DIR}")
target_link_libraries(Oauth2Test ${CLIENT_LIBS} pulsarStatic ${GTEST_LIBRARY_PATH})
-add_executable(ChunkDedupTest chunkdedup/ChunkDedupTest.cc)
+add_executable(ChunkDedupTest chunkdedup/ChunkDedupTest.cc HttpHelper.cc)
target_link_libraries(ChunkDedupTest ${CLIENT_LIBS} pulsarStatic ${GTEST_LIBRARY_PATH})
diff --git a/tests/chunkdedup/ChunkDedupTest.cc b/tests/chunkdedup/ChunkDedupTest.cc
index 609511f..4c9c28e 100644
--- a/tests/chunkdedup/ChunkDedupTest.cc
+++ b/tests/chunkdedup/ChunkDedupTest.cc
@@ -18,7 +18,9 @@
*/
#include <gtest/gtest.h>
#include <pulsar/Client.h>
+#include <time.h>
+#include "../HttpHelper.h"
#include "lib/Latch.h"
#include "lib/LogUtils.h"
@@ -47,6 +49,30 @@ TEST(ChunkDedupTest, testSendChunks) {
client.close();
}
+TEST(ChunkDedupTest, testLazyPartitionedProducer) {
+ std::string topic = "test-lazy-partitioned-producer-" + std::to_string(time(nullptr));
+ Client client{"pulsar://localhost:6650"};
+ ProducerConfiguration conf;
+ conf.setLazyStartPartitionedProducers(true);
+ Producer producer;
+ ASSERT_EQ(ResultOk, client.createProducer(topic, conf, producer));
+
+ constexpr int numPartitions = 3;
+ int res =
+ makePutRequest("http://localhost:8080/admin/v2/persistent/public/default/" + topic + "/partitions",
+ std::to_string(numPartitions));
+ ASSERT_TRUE(res == 204 || res == 409) << "res: " << res;
+
+ for (int i = 0; i < 10; i++) {
+ const auto key = std::to_string(i % numPartitions);
+ MessageId msgId;
+ producer.send(MessageBuilder().setPartitionKey(key).setContent("msg-" + std::to_string(i)).build(),
+ msgId);
+ ASSERT_TRUE(msgId.ledgerId() >= 0 && msgId.entryId() >= 0) << "i: " << i << ", msgId: " << msgId;
+ }
+ client.close();
+}
+
int main(int argc, char* argv[]) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
(pulsar-client-cpp) 04/04: Fix crash when removing connection from the pool (#347)
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 aee12e2a93cb74857888f746b787ea7b3e1a81ba
Author: Yunze Xu <xy...@163.com>
AuthorDate: Fri Nov 17 10:29:27 2023 +0800
Fix crash when removing connection from the pool (#347)
Fixes https://github.com/apache/pulsar-client-cpp/issues/346
### Motivation
https://github.com/apache/pulsar-client-cpp/pull/336 changes the key of
the `ClientConnection` in `ConnectionPool`, while in
`ClientConnection::close`, it still passes the old key (logical address)
to `ConnectionPool::remove`, which results in the connection could never
be removed and destroyed until being deleted as a stale connection.
What's worse, if the key does not exist, the iterator returned by
`std::map::find` will still be dereferenced, which might cause crash in
some platforms. See
https://github.com/apache/pulsar-client-cpp/blob/8d32fd254e294d1fabba73aed70115a434b341ef/lib/ConnectionPool.cc#L122-L123
### Modifications
- Avoid dereferencing the iterator if it's invalid in
`ConnectionPool::remove`.
- Store the key suffix in `ClientConnection` and pass the correct key to
`ConnectionPool::remove` in `ClientConnection::close`
- Add `ClientTest.testConnectionClose` to verify
`ClientConnection::close` can remove itself from the pool and the
connection will be destroyed eventually.
(cherry picked from commit 6d47e9495c071975d94fc284d6bfb6bbe9c7501a)
---
lib/ClientConnection.cc | 9 +++++----
lib/ClientConnection.h | 4 +++-
lib/ConnectionPool.cc | 7 ++++---
tests/ClientTest.cc | 41 +++++++++++++++++++++++++++++++++++++++++
tests/PulsarFriend.h | 2 ++
5 files changed, 55 insertions(+), 8 deletions(-)
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index 8f42535..fbcbe62 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -160,7 +160,7 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
ExecutorServicePtr executor,
const ClientConfiguration& clientConfiguration,
const AuthenticationPtr& authentication, const std::string& clientVersion,
- ConnectionPool& pool)
+ ConnectionPool& pool, size_t poolIndex)
: operationsTimeout_(seconds(clientConfiguration.getOperationTimeoutSeconds())),
authentication_(authentication),
serverProtocolVersion_(proto::ProtocolVersion_MIN),
@@ -184,7 +184,8 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
consumerStatsRequestTimer_(executor_->createDeadlineTimer()),
maxPendingLookupRequest_(clientConfiguration.getConcurrentLookupRequest()),
clientVersion_(clientVersion),
- pool_(pool) {
+ pool_(pool),
+ poolIndex_(poolIndex) {
LOG_INFO(cnxString_ << "Create ClientConnection, timeout=" << clientConfiguration.getConnectionTimeout());
if (clientConfiguration.isUseTls()) {
#if BOOST_VERSION >= 105400
@@ -265,7 +266,7 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
}
ClientConnection::~ClientConnection() {
- LOG_INFO(cnxString_ << "Destroyed connection to " << logicalAddress_);
+ LOG_INFO(cnxString_ << "Destroyed connection to " << logicalAddress_ << "-" << poolIndex_);
}
void ClientConnection::handlePulsarConnected(const proto::CommandConnected& cmdConnected) {
@@ -1320,7 +1321,7 @@ void ClientConnection::close(Result result, bool detach) {
}
// Remove the connection from the pool before completing any promise
if (detach) {
- pool_.remove(logicalAddress_, this); // trigger the destructor
+ pool_.remove(logicalAddress_ + "-" + std::to_string(poolIndex_), this);
}
auto self = shared_from_this();
diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h
index 30ea8d8..965c6ae 100644
--- a/lib/ClientConnection.h
+++ b/lib/ClientConnection.h
@@ -129,7 +129,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
ClientConnection(const std::string& logicalAddress, const std::string& physicalAddress,
ExecutorServicePtr executor, const ClientConfiguration& clientConfiguration,
const AuthenticationPtr& authentication, const std::string& clientVersion,
- ConnectionPool& pool);
+ ConnectionPool& pool, size_t poolIndex);
~ClientConnection();
#if __cplusplus < 201703L
@@ -400,6 +400,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
const std::string clientVersion_;
ConnectionPool& pool_;
+ const size_t poolIndex_;
+
friend class PulsarFriend;
void checkServerError(ServerError error);
diff --git a/lib/ConnectionPool.cc b/lib/ConnectionPool.cc
index 01285e9..b5b65e3 100644
--- a/lib/ConnectionPool.cc
+++ b/lib/ConnectionPool.cc
@@ -97,7 +97,8 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
ClientConnectionPtr cnx;
try {
cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(keySuffix),
- clientConfiguration_, authentication_, clientVersion_, *this));
+ clientConfiguration_, authentication_, clientVersion_, *this,
+ keySuffix));
} catch (const std::runtime_error& e) {
lock.unlock();
LOG_ERROR("Failed to create connection: " << e.what())
@@ -106,7 +107,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
return promise.getFuture();
}
- LOG_INFO("Created connection for " << logicalAddress);
+ LOG_INFO("Created connection for " << key);
Future<Result, ClientConnectionWeakPtr> future = cnx->getConnectFuture();
pool_.insert(std::make_pair(key, cnx));
@@ -120,7 +121,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
void ConnectionPool::remove(const std::string& key, ClientConnection* value) {
std::lock_guard<std::recursive_mutex> lock(mutex_);
auto it = pool_.find(key);
- if (it->second.get() == value) {
+ if (it != pool_.end() && it->second.get() == value) {
LOG_INFO("Remove connection for " << key);
pool_.erase(it);
}
diff --git a/tests/ClientTest.cc b/tests/ClientTest.cc
index 983e5d9..5b4dd2e 100644
--- a/tests/ClientTest.cc
+++ b/tests/ClientTest.cc
@@ -400,3 +400,44 @@ TEST(ClientTest, testClientVersion) {
client.close();
}
+
+TEST(ClientTest, testConnectionClose) {
+ std::vector<Client> clients;
+ clients.emplace_back(lookupUrl);
+ clients.emplace_back(lookupUrl, ClientConfiguration().setConnectionsPerBroker(5));
+
+ const auto topic = "client-test-connection-close";
+ for (auto &client : clients) {
+ auto testClose = [&client](ClientConnectionWeakPtr weakCnx) {
+ auto cnx = weakCnx.lock();
+ ASSERT_TRUE(cnx);
+
+ auto numConnections = PulsarFriend::getConnections(client).size();
+ LOG_INFO("Connection refcnt: " << cnx.use_count() << " before close");
+ auto executor = PulsarFriend::getExecutor(*cnx);
+ // Simulate the close() happens in the event loop
+ executor->postWork([cnx, &client, numConnections] {
+ cnx->close();
+ ASSERT_EQ(PulsarFriend::getConnections(client).size(), numConnections - 1);
+ LOG_INFO("Connection refcnt: " << cnx.use_count() << " after close");
+ });
+ cnx.reset();
+
+ // The ClientConnection could still be referred in a socket callback, wait until all these
+ // callbacks being cancelled due to the socket close.
+ ASSERT_TRUE(waitUntil(
+ std::chrono::seconds(1), [weakCnx] { return weakCnx.expired(); }, 1));
+ };
+ Producer producer;
+ ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
+ testClose(PulsarFriend::getProducerImpl(producer).getCnx());
+ producer.close();
+
+ Consumer consumer;
+ ASSERT_EQ(ResultOk, client.subscribe("client-test-connection-close", "sub", consumer));
+ testClose(PulsarFriend::getConsumerImpl(consumer).getCnx());
+ consumer.close();
+
+ client.close();
+ }
+}
diff --git a/tests/PulsarFriend.h b/tests/PulsarFriend.h
index de82ce4..c2863e8 100644
--- a/tests/PulsarFriend.h
+++ b/tests/PulsarFriend.h
@@ -140,6 +140,8 @@ class PulsarFriend {
return connections;
}
+ static ExecutorServicePtr getExecutor(const ClientConnection& cnx) { return cnx.executor_; }
+
static std::vector<ProducerImplPtr> getProducers(const ClientConnection& cnx) {
std::vector<ProducerImplPtr> producers;
std::lock_guard<std::mutex> lock(cnx.mutex_);
(pulsar-client-cpp) 01/04: Use absolute path to find credential files in unit tests (#340)
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 39d39a98dd8fd107d7f22c5f575a53ee94af75c8
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Nov 9 15:19:06 2023 +0800
Use absolute path to find credential files in unit tests (#340)
Fixes https://github.com/apache/pulsar-client-cpp/issues/339
### Motivation
Currently the relative path is used in unit tests to specify the token
file (`.test-token.txt`) and the credential directory (`./test-conf`).
If we don't run the `pulsar-tests` binary in a subdirectory, it won't
be able to read these files.
### Modifications
Add the macro `-DTOKEN_PATH="..." -DTEST_CONF_DIR="..."` to specify
absolute paths according to the `PROJECT_SOURCE_DIR`, which is defined
by CMake as the absolute path of the project directory.
### Verifications
```bash
cmake -B build
cmake --build build -j8
# Run the test in project directory
./build/tests/pulsar-tests --gtest_filter='AuthPlugin*:*testRSAEncryption:*testEncryptionFailure:*DecryptionFailedMessages'
# Run the test in a subdirectory
cd build
./tests/pulsar-tests --gtest_filter='AuthPlugin*:*testRSAEncryption:*testEncryptionFailure:*DecryptionFailedMessages'
```
(cherry picked from commit 47fb809a3f92fe928f134687d980ffbae9bcdf72)
---
CMakeLists.txt | 2 ++
pulsar-test-service-start.sh | 4 ++--
pulsar-test-service-stop.sh | 4 ++--
tests/AuthBasicTest.cc | 10 +++++++---
tests/AuthPluginTest.cc | 20 ++++++++++++++------
tests/AuthTokenTest.cc | 6 +++++-
tests/BasicEndToEndTest.cc | 16 ++++++++++------
tests/CMakeLists.txt | 4 ++--
tests/ConsumerTest.cc | 8 ++++++--
tests/oauth2/Oauth2Test.cc | 6 +++---
10 files changed, 53 insertions(+), 27 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index c1e7a79..c8d6a00 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -395,6 +395,8 @@ if (BUILD_DYNAMIC_LIB)
endif()
if (BUILD_TESTS)
+ set(TOKEN_PATH "${PROJECT_SOURCE_DIR}/.test-token.txt")
+ set(TEST_CONF_DIR "${PROJECT_SOURCE_DIR}/test-conf")
add_subdirectory(tests)
endif()
diff --git a/pulsar-test-service-start.sh b/pulsar-test-service-start.sh
index 2b1a1e3..23f7010 100755
--- a/pulsar-test-service-start.sh
+++ b/pulsar-test-service-start.sh
@@ -20,8 +20,8 @@
set -e
-SRC_DIR=$(git rev-parse --show-toplevel)
-cd $SRC_DIR
+cd `dirname $0`
+SRC_DIR=$PWD
./pulsar-test-service-stop.sh
diff --git a/pulsar-test-service-stop.sh b/pulsar-test-service-stop.sh
index 9d3bf4e..aad069e 100755
--- a/pulsar-test-service-stop.sh
+++ b/pulsar-test-service-stop.sh
@@ -20,8 +20,8 @@
set -e
-SRC_DIR=$(git rev-parse --show-toplevel)
-cd $SRC_DIR
+cd `dirname $0`
+SRC_DIR=$PWD
CONTAINER_ID_PATH=".tests-container-id.txt"
diff --git a/tests/AuthBasicTest.cc b/tests/AuthBasicTest.cc
index 0b8b7f0..0e18172 100644
--- a/tests/AuthBasicTest.cc
+++ b/tests/AuthBasicTest.cc
@@ -25,13 +25,17 @@
using namespace pulsar;
+#ifndef TEST_CONF_DIR
+#error "TEST_CONF_DIR is not specified"
+#endif
+
static const std::string serviceUrl = "pulsar://localhost:6650";
static const std::string serviceUrlHttp = "http://localhost:8080";
static const std::string serviceUrlTls = "pulsar+ssl://localhost:6651";
static const std::string serviceUrlHttps = "https://localhost:8443";
-static const std::string caPath = "../test-conf/cacert.pem";
-static const std::string clientCertificatePath = "../test-conf/client-cert.pem";
-static const std::string clientPrivateKeyPath = "../test-conf/client-key.pem";
+static const std::string caPath = TEST_CONF_DIR "/cacert.pem";
+static const std::string clientCertificatePath = TEST_CONF_DIR "/client-cert.pem";
+static const std::string clientPrivateKeyPath = TEST_CONF_DIR "/client-key.pem";
TEST(AuthPluginBasic, testBasic) {
ClientConfiguration config = ClientConfiguration();
diff --git a/tests/AuthPluginTest.cc b/tests/AuthPluginTest.cc
index 2d23bf9..1e16019 100644
--- a/tests/AuthPluginTest.cc
+++ b/tests/AuthPluginTest.cc
@@ -37,15 +37,19 @@ int globalTestTlsMessagesCounter = 0;
static const std::string serviceUrlTls = "pulsar+ssl://localhost:6651";
static const std::string serviceUrlHttps = "https://localhost:8443";
-static const std::string caPath = "../test-conf/cacert.pem";
-static const std::string clientPublicKeyPath = "../test-conf/client-cert.pem";
-static const std::string clientPrivateKeyPath = "../test-conf/client-key.pem";
+#ifndef TEST_CONF_DIR
+#error "TEST_CONF_DIR is not specified"
+#endif
+
+static const std::string caPath = TEST_CONF_DIR "/cacert.pem";
+static const std::string clientPublicKeyPath = TEST_CONF_DIR "/client-cert.pem";
+static const std::string clientPrivateKeyPath = TEST_CONF_DIR "/client-key.pem";
// Man in middle certificate which tries to act as a broker by sending its own valid certificate
static const std::string mimServiceUrlTls = "pulsar+ssl://localhost:6653";
static const std::string mimServiceUrlHttps = "https://localhost:8444";
-static const std::string mimCaPath = "../test-conf/hn-verification/cacert.pem";
+static const std::string mimCaPath = TEST_CONF_DIR "/hn-verification/cacert.pem";
static void sendCallBackTls(Result r, const MessageId& msgId) {
ASSERT_EQ(r, ResultOk);
@@ -468,12 +472,16 @@ TEST(AuthPluginTest, testOauth2WrongSecret) {
TEST(AuthPluginTest, testOauth2CredentialFile) {
// test success get token from oauth2 server.
pulsar::AuthenticationDataPtr data;
- std::string params = R"({
+ const char* paramsTemplate = R"({
"type": "client_credentials",
"issuer_url": "https://dev-kt-aa9ne.us.auth0.com",
- "private_key": "../test-conf/cpp_credentials_file.json",
+ "private_key": "%s/cpp_credentials_file.json",
"audience": "https://dev-kt-aa9ne.us.auth0.com/api/v2/"})";
+ char params[4096];
+ int numWritten = snprintf(params, sizeof(params), paramsTemplate, TEST_CONF_DIR);
+ ASSERT_TRUE(numWritten < sizeof(params));
+
int expectedTokenLength = 3379;
LOG_INFO("PARAMS: " << params);
pulsar::AuthenticationPtr auth = pulsar::AuthOauth2::create(params);
diff --git a/tests/AuthTokenTest.cc b/tests/AuthTokenTest.cc
index 8bdd268..a04da08 100644
--- a/tests/AuthTokenTest.cc
+++ b/tests/AuthTokenTest.cc
@@ -37,7 +37,11 @@ using namespace pulsar;
static const std::string serviceUrl = "pulsar://localhost:6650";
static const std::string serviceUrlHttp = "http://localhost:8080";
-static const std::string tokenPath = "../.test-token.txt";
+#ifndef TOKEN_PATH
+#error "TOKEN_PATH is not specified"
+#endif
+
+static const std::string tokenPath = TOKEN_PATH;
static std::string getToken() {
std::ifstream file(tokenPath);
diff --git a/tests/BasicEndToEndTest.cc b/tests/BasicEndToEndTest.cc
index 9ca2ab0..0951dd4 100644
--- a/tests/BasicEndToEndTest.cc
+++ b/tests/BasicEndToEndTest.cc
@@ -54,6 +54,10 @@ DECLARE_LOG_OBJECT()
using namespace pulsar;
+#ifndef TEST_CONF_DIR
+#error "TEST_CONF_DIR is not specified"
+#endif
+
std::mutex mutex_;
static int globalCount = 0;
static long globalResendMessageCount = 0;
@@ -1341,9 +1345,9 @@ TEST(BasicEndToEndTest, testRSAEncryption) {
std::string subName = "my-sub-name";
Producer producer;
- std::string PUBLIC_CERT_FILE_PATH = "../test-conf/public-key.client-rsa.pem";
+ std::string PUBLIC_CERT_FILE_PATH = TEST_CONF_DIR "/public-key.client-rsa.pem";
- std::string PRIVATE_CERT_FILE_PATH = "../test-conf//private-key.client-rsa.pem";
+ std::string PRIVATE_CERT_FILE_PATH = TEST_CONF_DIR "/private-key.client-rsa.pem";
std::shared_ptr<pulsar::DefaultCryptoKeyReader> keyReader =
std::make_shared<pulsar::DefaultCryptoKeyReader>(PUBLIC_CERT_FILE_PATH, PRIVATE_CERT_FILE_PATH);
@@ -1407,9 +1411,9 @@ TEST(BasicEndToEndTest, testEncryptionFailure) {
std::string subName = "my-sub-name";
Producer producer;
- std::string PUBLIC_CERT_FILE_PATH = "../test-conf/public-key.client-rsa-test.pem";
+ std::string PUBLIC_CERT_FILE_PATH = TEST_CONF_DIR "/public-key.client-rsa-test.pem";
- std::string PRIVATE_CERT_FILE_PATH = "../test-conf//private-key.client-rsa-test.pem";
+ std::string PRIVATE_CERT_FILE_PATH = TEST_CONF_DIR "/private-key.client-rsa-test.pem";
std::shared_ptr<pulsar::DefaultCryptoKeyReader> keyReader =
std::make_shared<pulsar::DefaultCryptoKeyReader>(PUBLIC_CERT_FILE_PATH, PRIVATE_CERT_FILE_PATH);
@@ -1449,9 +1453,9 @@ TEST(BasicEndToEndTest, testEncryptionFailure) {
// 2. Add valid key
{
- PUBLIC_CERT_FILE_PATH = "../test-conf/public-key.client-rsa.pem";
+ PUBLIC_CERT_FILE_PATH = TEST_CONF_DIR "/public-key.client-rsa.pem";
- PRIVATE_CERT_FILE_PATH = "../test-conf/private-key.client-rsa.pem";
+ PRIVATE_CERT_FILE_PATH = TEST_CONF_DIR "/private-key.client-rsa.pem";
keyReader =
std::make_shared<pulsar::DefaultCryptoKeyReader>(PUBLIC_CERT_FILE_PATH, PRIVATE_CERT_FILE_PATH);
ProducerConfiguration prodConf;
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index b373196..993c2fd 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -57,7 +57,7 @@ file(GLOB TEST_SOURCES *.cc c/*.cc)
add_executable(pulsar-tests ${TEST_SOURCES} ${PROTO_SOURCES})
target_include_directories(pulsar-tests PRIVATE ${AUTOGEN_DIR}/lib)
-
+target_compile_options(pulsar-tests PRIVATE -DTOKEN_PATH="${TOKEN_PATH}" -DTEST_CONF_DIR="${TEST_CONF_DIR}")
target_link_libraries(pulsar-tests ${CLIENT_LIBS} pulsarStatic $<$<CONFIG:Debug>:${GMOCKD_LIBRARY_PATH}> $<$<CONFIG:Debug>:${GTESTD_LIBRARY_PATH}> $<$<NOT:$<CONFIG:Debug>>:${GMOCK_LIBRARY_PATH}> $<$<NOT:$<CONFIG:Debug>>:${GTEST_LIBRARY_PATH}>)
if (UNIX)
@@ -69,7 +69,7 @@ add_executable(BrokerMetadataTest brokermetadata/BrokerMetadataTest.cc)
target_link_libraries(BrokerMetadataTest ${CLIENT_LIBS} pulsarStatic ${GTEST_LIBRARY_PATH})
add_executable(Oauth2Test oauth2/Oauth2Test.cc)
-target_compile_options(Oauth2Test PRIVATE "-DTEST_ROOT_PATH=\"${CMAKE_CURRENT_SOURCE_DIR}\"")
+target_compile_options(Oauth2Test PRIVATE -DTEST_CONF_DIR="${TEST_CONF_DIR}")
target_link_libraries(Oauth2Test ${CLIENT_LIBS} pulsarStatic ${GTEST_LIBRARY_PATH})
add_executable(ChunkDedupTest chunkdedup/ChunkDedupTest.cc)
diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc
index 7917508..80c0a71 100644
--- a/tests/ConsumerTest.cc
+++ b/tests/ConsumerTest.cc
@@ -50,6 +50,10 @@ static const std::string adminUrl = "http://localhost:8080/";
DECLARE_LOG_OBJECT()
+#ifndef TEST_CONF_DIR
+#error "TEST_CONF_DIR is not specified"
+#endif
+
namespace pulsar {
class ConsumerStateEventListener : public ConsumerEventListener {
@@ -960,8 +964,8 @@ TEST(ConsumerTest, testRedeliveryOfDecryptionFailedMessages) {
std::string topicName = "testRedeliveryOfDecryptionFailedMessages" + std::to_string(time(nullptr));
std::string subName = "sub-test";
- std::string PUBLIC_CERT_FILE_PATH = "../test-conf/public-key.client-rsa.pem";
- std::string PRIVATE_CERT_FILE_PATH = "../test-conf/private-key.client-rsa.pem";
+ std::string PUBLIC_CERT_FILE_PATH = TEST_CONF_DIR "/public-key.client-rsa.pem";
+ std::string PRIVATE_CERT_FILE_PATH = TEST_CONF_DIR "/private-key.client-rsa.pem";
std::shared_ptr<pulsar::DefaultCryptoKeyReader> keyReader =
std::make_shared<pulsar::DefaultCryptoKeyReader>(PUBLIC_CERT_FILE_PATH, PRIVATE_CERT_FILE_PATH);
diff --git a/tests/oauth2/Oauth2Test.cc b/tests/oauth2/Oauth2Test.cc
index 5f273d7..719ae69 100644
--- a/tests/oauth2/Oauth2Test.cc
+++ b/tests/oauth2/Oauth2Test.cc
@@ -27,11 +27,11 @@
using namespace pulsar;
-#ifndef TEST_ROOT_PATH
-#define TEST_ROOT_PATH "."
+#ifndef TEST_CONF_DIR
+#error "TEST_CONF_DIR is not specified"
#endif
-static const std::string gKeyPath = std::string(TEST_ROOT_PATH) + "/../test-conf/cpp_credentials_file.json";
+static const std::string gKeyPath = TEST_CONF_DIR "/cpp_credentials_file.json";
static std::string gClientId;
static std::string gClientSecret;
static ParamMap gCommonParams;