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 2026/01/16 06:41:00 UTC
(pulsar-client-cpp) branch branch-4.0 updated (cada45d -> 4bd7e8c)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
from cada45d Release 4.0.0
new 1e5ef27 [improve][client] Add TLSv1.3 support (#529)
new 4bd7e8c Fix the seek method could be blocked forever when subscribe RPC is slower than seek RPC (#533)
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/ClientConnection.cc | 36 +++++--
lib/ClientConnection.h | 22 +++-
lib/ConsumerImpl.cc | 246 +++++++++++++++++++++++++-------------------
lib/ConsumerImpl.h | 63 ++++++------
lib/HandlerBase.cc | 2 +-
lib/HandlerBase.h | 2 +-
lib/MockServer.h | 139 +++++++++++++++++++++++++
lib/ProducerImpl.cc | 10 +-
lib/Synchronized.h | 47 ---------
tests/ConsumerSeekTest.cc | 58 +++++++++++
tests/TlsNegotiationTest.cc | 179 ++++++++++++++++++++++++++++++++
11 files changed, 603 insertions(+), 201 deletions(-)
create mode 100644 lib/MockServer.h
delete mode 100644 lib/Synchronized.h
create mode 100644 tests/TlsNegotiationTest.cc
(pulsar-client-cpp) 01/02: [improve][client] Add TLSv1.3 support (#529)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 1e5ef270a9aeaf026323a57cb0dd65f4e2ca05b4
Author: Ceyhun Can Ulker <ce...@gmail.com>
AuthorDate: Wed Dec 17 05:08:57 2025 +0300
[improve][client] Add TLSv1.3 support (#529)
(cherry picked from commit d040039e3a351056d0b3be31adbad4a0b72fee09)
---
lib/ClientConnection.cc | 5 +-
tests/TlsNegotiationTest.cc | 179 ++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 183 insertions(+), 1 deletion(-)
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index 0bd935d..86dfd9d 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -207,7 +207,10 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
}
if (clientConfiguration.isUseTls()) {
- ASIO::ssl::context ctx(ASIO::ssl::context::tlsv12_client);
+ ASIO::ssl::context ctx(ASIO::ssl::context::sslv23_client);
+ ctx.set_options(ASIO::ssl::context::default_workarounds | ASIO::ssl::context::no_sslv2 |
+ ASIO::ssl::context::no_sslv3 | ASIO::ssl::context::no_tlsv1 |
+ ASIO::ssl::context::no_tlsv1_1);
Url serviceUrl;
Url proxyUrl;
Url::parse(physicalAddress, serviceUrl);
diff --git a/tests/TlsNegotiationTest.cc b/tests/TlsNegotiationTest.cc
new file mode 100644
index 0000000..64f956d
--- /dev/null
+++ b/tests/TlsNegotiationTest.cc
@@ -0,0 +1,179 @@
+/**
+ * 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.
+ */
+#include <gtest/gtest.h>
+#include <openssl/ssl.h>
+#include <pulsar/Authentication.h>
+#include <pulsar/Client.h>
+
+#include <atomic>
+#include <future>
+#include <thread>
+
+#include "lib/AsioDefines.h"
+#include "lib/LogUtils.h"
+
+#ifdef USE_ASIO
+#include <asio.hpp>
+#include <asio/ssl.hpp>
+#else
+#include <boost/asio.hpp>
+#include <boost/asio/ssl.hpp>
+#endif
+
+DECLARE_LOG_OBJECT()
+
+#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";
+
+using namespace pulsar;
+
+class MockTlsServer {
+ public:
+ MockTlsServer()
+ : acceptor_(io_context_, ASIO::ip::tcp::endpoint(ASIO::ip::tcp::v4(), 0)),
+ ctx_(ASIO::ssl::context::sslv23) {
+ ctx_.set_options(ASIO::ssl::context::default_workarounds | ASIO::ssl::context::no_sslv2 |
+ ASIO::ssl::context::no_sslv3);
+
+ ctx_.use_certificate_chain_file(clientPublicKeyPath);
+ ctx_.use_private_key_file(clientPrivateKeyPath, ASIO::ssl::context::pem);
+ ctx_.set_verify_mode(ASIO::ssl::context::verify_none);
+ }
+
+ int getPort() const { return acceptor_.local_endpoint().port(); }
+
+ void setTls12Only() {
+ SSL_CTX* ssl_ctx = ctx_.native_handle();
+#if defined(TLS1_2_VERSION)
+ SSL_CTX_set_min_proto_version(ssl_ctx, TLS1_2_VERSION);
+ SSL_CTX_set_max_proto_version(ssl_ctx, TLS1_2_VERSION);
+#else
+ LOG_WARN("TLS 1.2 not supported by OpenSSL headers");
+#endif
+ }
+
+ void setTls13Only() {
+ SSL_CTX* ssl_ctx = ctx_.native_handle();
+#if defined(TLS1_3_VERSION)
+ SSL_CTX_set_min_proto_version(ssl_ctx, TLS1_3_VERSION);
+ SSL_CTX_set_max_proto_version(ssl_ctx, TLS1_3_VERSION);
+#else
+ LOG_WARN("TLS 1.3 not supported by OpenSSL headers");
+#endif
+ }
+
+ bool acceptAndHandshake() {
+ auto socket = std::make_shared<ASIO::ip::tcp::socket>(io_context_);
+ acceptor_.accept(*socket);
+
+ ASIO::ssl::stream<ASIO::ip::tcp::socket&> ssl_stream(*socket, ctx_);
+
+ ASIO_ERROR error;
+ ssl_stream.handshake(ASIO::ssl::stream_base::server, error);
+
+ if (error) {
+ LOG_ERROR("Handshake failed: " << error.message());
+ return false;
+ }
+ LOG_INFO("Handshake success!");
+ return true;
+ }
+
+ private:
+ ASIO::io_context io_context_;
+ ASIO::ip::tcp::acceptor acceptor_;
+ ASIO::ssl::context ctx_;
+};
+
+TEST(TlsNegotiationTest, testTls12) {
+#if !defined(TLS1_2_VERSION)
+ return; // Skip if TLS 1.2 is not available
+#endif
+
+ MockTlsServer server;
+ server.setTls12Only();
+ int port = server.getPort();
+
+ std::promise<bool> handshakePromise;
+ auto handshakeFuture = handshakePromise.get_future();
+
+ std::thread serverThread([&server, &handshakePromise]() {
+ bool result = server.acceptAndHandshake();
+ handshakePromise.set_value(result);
+ });
+
+ std::string serviceUrl = "pulsar+ssl://localhost:" + std::to_string(port);
+ ClientConfiguration config;
+ config.setTlsTrustCertsFilePath(caPath);
+ config.setTlsAllowInsecureConnection(true); // Self-signed certs match
+ config.setValidateHostName(false);
+
+ Client client(serviceUrl, config);
+
+ // Trigger connection by creating a producer.
+ // It will fail to create producer because mock server doesn't speak Pulsar,
+ // but we only care about the handshake.
+ Producer producer;
+ client.createProducerAsync("topic", [](Result, Producer) {});
+
+ // Wait for handshake
+ ASSERT_TRUE(handshakeFuture.get());
+
+ serverThread.join();
+ client.close();
+}
+
+TEST(TlsNegotiationTest, testTls13) {
+#if !defined(TLS1_3_VERSION)
+ LOG_INFO("Skipping TLS 1.3 test because OpenSSL does not support it");
+ return;
+#endif
+
+ MockTlsServer server;
+ server.setTls13Only();
+ int port = server.getPort();
+
+ std::promise<bool> handshakePromise;
+ auto handshakeFuture = handshakePromise.get_future();
+
+ std::thread serverThread([&server, &handshakePromise]() {
+ bool result = server.acceptAndHandshake();
+ handshakePromise.set_value(result);
+ });
+
+ std::string serviceUrl = "pulsar+ssl://localhost:" + std::to_string(port);
+ ClientConfiguration config;
+ config.setTlsTrustCertsFilePath(caPath);
+ config.setTlsAllowInsecureConnection(true);
+ config.setValidateHostName(false);
+
+ Client client(serviceUrl, config);
+
+ client.createProducerAsync("topic", [](Result, Producer) {});
+
+ ASSERT_TRUE(handshakeFuture.get());
+
+ serverThread.join();
+ client.close();
+}
(pulsar-client-cpp) 02/02: Fix the seek method could be blocked forever when subscribe RPC is slower than seek RPC (#533)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 4bd7e8c8543d1d32be40ce674e92c376042b4fe2
Author: Yunze Xu <xy...@163.com>
AuthorDate: Fri Jan 16 14:14:50 2026 +0800
Fix the seek method could be blocked forever when subscribe RPC is slower than seek RPC (#533)
(cherry picked from commit 0b893a1622aa7dd702d48afee33af321bb2c06dd)
---
lib/ClientConnection.cc | 31 ++++--
lib/ClientConnection.h | 22 ++++-
lib/ConsumerImpl.cc | 246 ++++++++++++++++++++++++++--------------------
lib/ConsumerImpl.h | 63 ++++++------
lib/HandlerBase.cc | 2 +-
lib/HandlerBase.h | 2 +-
lib/MockServer.h | 139 ++++++++++++++++++++++++++
lib/ProducerImpl.cc | 10 +-
lib/Synchronized.h | 47 ---------
tests/ConsumerSeekTest.cc | 58 +++++++++++
10 files changed, 420 insertions(+), 200 deletions(-)
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index 86dfd9d..4f7a1dd 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -21,6 +21,7 @@
#include <openssl/x509.h>
#include <pulsar/MessageIdBuilder.h>
+#include <chrono>
#include <fstream>
#include "AsioDefines.h"
@@ -31,6 +32,7 @@
#include "ConsumerImpl.h"
#include "ExecutorService.h"
#include "LogUtils.h"
+#include "MockServer.h"
#include "OpSendMsg.h"
#include "ProducerImpl.h"
#include "PulsarApi.pb.h"
@@ -1005,15 +1007,17 @@ Future<Result, BrokerConsumerStatsImpl> ClientConnection::newConsumerStats(uint6
void ClientConnection::newTopicLookup(const std::string& topicName, bool authoritative,
const std::string& listenerName, uint64_t requestId,
const LookupDataResultPromisePtr& promise) {
- newLookup(Commands::newLookup(topicName, authoritative, requestId, listenerName), requestId, promise);
+ newLookup(Commands::newLookup(topicName, authoritative, requestId, listenerName), requestId, "LOOKUP",
+ promise);
}
void ClientConnection::newPartitionedMetadataLookup(const std::string& topicName, uint64_t requestId,
const LookupDataResultPromisePtr& promise) {
- newLookup(Commands::newPartitionMetadataRequest(topicName, requestId), requestId, promise);
+ newLookup(Commands::newPartitionMetadataRequest(topicName, requestId), requestId, "PARTITIONED_METADATA",
+ promise);
}
-void ClientConnection::newLookup(const SharedBuffer& cmd, uint64_t requestId,
+void ClientConnection::newLookup(const SharedBuffer& cmd, uint64_t requestId, const char* requestType,
const LookupDataResultPromisePtr& promise) {
Lock lock(mutex_);
std::shared_ptr<LookupDataResultPtr> lookupDataResult;
@@ -1042,6 +1046,7 @@ void ClientConnection::newLookup(const SharedBuffer& cmd, uint64_t requestId,
pendingLookupRequests_.insert(std::make_pair(requestId, requestData));
numOfPendingLookupRequest_++;
lock.unlock();
+ LOG_DEBUG(cnxString_ << "Inserted lookup request " << requestType << " (req_id: " << requestId << ")");
sendCommand(cmd);
}
@@ -1158,12 +1163,15 @@ void ClientConnection::sendPendingCommands() {
}
}
-Future<Result, ResponseData> ClientConnection::sendRequestWithId(const SharedBuffer& cmd, int requestId) {
+Future<Result, ResponseData> ClientConnection::sendRequestWithId(const SharedBuffer& cmd, int requestId,
+ const char* requestType) {
Lock lock(mutex_);
if (isClosed()) {
lock.unlock();
Promise<Result, ResponseData> promise;
+ LOG_DEBUG(cnxString_ << "Fail " << requestType << "(req_id: " << requestId
+ << ") to a closed connection");
promise.setFailed(ResultNotConnected);
return promise.getFuture();
}
@@ -1182,7 +1190,17 @@ Future<Result, ResponseData> ClientConnection::sendRequestWithId(const SharedBuf
pendingRequests_.insert(std::make_pair(requestId, requestData));
lock.unlock();
- sendCommand(cmd);
+ LOG_DEBUG(cnxString_ << "Inserted request " << requestType << " (req_id: " << requestId << ")");
+ if (mockingRequests_.load(std::memory_order_acquire)) {
+ if (mockServer_ == nullptr) {
+ LOG_WARN(cnxString_ << "Mock server is unexpectedly null when processing " << requestType);
+ sendCommand(cmd);
+ } else if (!mockServer_->sendRequest(requestType, requestId)) {
+ sendCommand(cmd);
+ }
+ } else {
+ sendCommand(cmd);
+ }
return requestData.promise.getFuture();
}
@@ -1625,9 +1643,6 @@ void ClientConnection::handleConsumerStatsResponse(
void ClientConnection::handleLookupTopicRespose(
const proto::CommandLookupTopicResponse& lookupTopicResponse) {
- LOG_DEBUG(cnxString_ << "Received lookup response from server. req_id: "
- << lookupTopicResponse.request_id());
-
Lock lock(mutex_);
auto it = pendingLookupRequests_.find(lookupTopicResponse.request_id());
if (it != pendingLookupRequests_.end()) {
diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h
index 18a7d84..aae53d2 100644
--- a/lib/ClientConnection.h
+++ b/lib/ClientConnection.h
@@ -115,6 +115,7 @@ struct ResponseData {
typedef std::shared_ptr<std::vector<std::string>> NamespaceTopicsPtr;
+class MockServer;
class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<ClientConnection> {
enum State : uint8_t
{
@@ -123,6 +124,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
Ready,
Disconnected
};
+ using RequestDelayType =
+ std::unordered_map<std::string /* request type */, long /* delay in milliseconds */>;
public:
typedef std::shared_ptr<ASIO::ip::tcp::socket> SocketPtr;
@@ -185,7 +188,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
* Send a request with a specific Id over the connection. The future will be
* triggered when the response for this request is received
*/
- Future<Result, ResponseData> sendRequestWithId(const SharedBuffer& cmd, int requestId);
+ Future<Result, ResponseData> sendRequestWithId(const SharedBuffer& cmd, int requestId,
+ const char* requestType);
const std::string& brokerAddress() const;
@@ -208,6 +212,13 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
Future<Result, SchemaInfo> newGetSchema(const std::string& topicName, const std::string& version,
uint64_t requestId);
+ void attachMockServer(const std::shared_ptr<MockServer>& mockServer) {
+ mockServer_ = mockServer;
+ // Mark that requests will first go through the mock server, if the mock server cannot process it,
+ // fall back to the normal logic
+ mockingRequests_.store(true, std::memory_order_release);
+ }
+
private:
struct PendingRequestData {
Promise<Result, ResponseData> promise;
@@ -264,7 +275,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
void handleSend(const ASIO_ERROR& err, const SharedBuffer& cmd);
void handleSendPair(const ASIO_ERROR& err);
void sendPendingCommands();
- void newLookup(const SharedBuffer& cmd, uint64_t requestId, const LookupDataResultPromisePtr& promise);
+ void newLookup(const SharedBuffer& cmd, uint64_t requestId, const char* requestType,
+ const LookupDataResultPromisePtr& promise);
void handleRequestTimeout(const ASIO_ERROR& ec, const PendingRequestData& pendingRequestData);
@@ -308,6 +320,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
}
}
+ void mockSendCommand(const char* requestType, uint64_t requestId, const SharedBuffer& cmd);
+
std::atomic<State> state_{Pending};
TimeDuration operationsTimeout_;
AuthenticationPtr authentication_;
@@ -391,6 +405,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
DeadlineTimerPtr keepAliveTimer_;
DeadlineTimerPtr consumerStatsRequestTimer_;
+ std::atomic_bool mockingRequests_{false};
+ std::shared_ptr<MockServer> mockServer_;
+
void handleConsumerStatsTimeout(const ASIO_ERROR& ec, const std::vector<uint64_t>& consumerStatsRequests);
void startConsumerStatsTimer(std::vector<uint64_t> consumerStatsRequests);
@@ -405,6 +422,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
friend class PulsarFriend;
friend class ConsumerTest;
+ friend class MockServer;
void checkServerError(ServerError error, const std::string& message);
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index 430b851..a645f58 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -23,6 +23,8 @@
#include <pulsar/MessageIdBuilder.h>
#include <algorithm>
+#include <utility>
+#include <variant>
#include "AckGroupingTracker.h"
#include "AckGroupingTrackerDisabled.h"
@@ -123,7 +125,7 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr& client, const std::string& topic
negativeAcksTracker_(std::make_shared<NegativeAcksTracker>(client, *this, conf)),
ackGroupingTrackerPtr_(newAckGroupingTracker(topic, conf, client)),
readCompacted_(conf.isReadCompacted()),
- startMessageId_(getStartMessageId(startMessageId, conf.isStartMessageIdInclusive())),
+ startMessageId_(pulsar::getStartMessageId(startMessageId, conf.isStartMessageIdInclusive())),
maxPendingChunkedMessage_(conf.getMaxPendingChunkedMessage()),
autoAckOldestChunkedMessageOnQueueFull_(conf.isAutoAckOldestChunkedMessageOnQueueFull()),
expireTimeOfIncompleteChunkedMessageMs_(conf.getExpireTimeOfIncompleteChunkedMessageMs()),
@@ -189,7 +191,8 @@ ConsumerImpl::~ConsumerImpl() {
ClientConnectionPtr cnx = getCnx().lock();
if (cnx) {
auto requestId = newRequestId();
- cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId);
+ cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId,
+ "CLOSE_CONSUMER");
cnx->removeConsumer(consumerId_);
LOG_INFO(consumerStr_ << "Closed consumer for race condition: " << consumerId_);
} else {
@@ -232,20 +235,19 @@ Future<Result, bool> ConsumerImpl::connectionOpened(const ClientConnectionPtr& c
return promise.getFuture();
}
- // Register consumer so that we can handle other incomming commands (e.g. ACTIVE_CONSUMER_CHANGE) after
- // sending the subscribe request.
- cnx->registerConsumer(consumerId_, get_shared_this_ptr());
+ optional<MessageId> subscribeMessageId;
+ {
+ LockGuard lock{mutex_};
+ setCnx(cnx);
+ cnx->registerConsumer(consumerId_, get_shared_this_ptr());
+ LOG_DEBUG(cnx->cnxString() << "Registered consumer " << consumerId_);
- if (duringSeek()) {
- ackGroupingTrackerPtr_->flushAndClean();
+ clearReceiveQueue();
+ subscribeMessageId =
+ (subscriptionMode_ == Commands::SubscriptionModeNonDurable) ? startMessageId_ : std::nullopt;
+ lastDequedMessageId_ = MessageId::earliest();
}
- Lock lockForMessageId(mutexForMessageId_);
- clearReceiveQueue();
- const auto subscribeMessageId =
- (subscriptionMode_ == Commands::SubscriptionModeNonDurable) ? startMessageId_.get() : std::nullopt;
- lockForMessageId.unlock();
-
unAckedMessageTrackerPtr_->clear();
ClientImplPtr client = client_.lock();
@@ -259,13 +261,22 @@ Future<Result, bool> ConsumerImpl::connectionOpened(const ClientConnectionPtr& c
// Keep a reference to ensure object is kept alive.
auto self = get_shared_this_ptr();
setFirstRequestIdAfterConnect(requestId);
- cnx->sendRequestWithId(cmd, requestId)
+ cnx->sendRequestWithId(cmd, requestId, "SUBSCRIBE")
.addListener([this, self, cnx, promise](Result result, const ResponseData& responseData) {
Result handleResult = handleCreateConsumer(cnx, result);
- if (handleResult == ResultOk) {
- promise.setSuccess();
- } else {
+ if (handleResult != ResultOk) {
promise.setFailed(handleResult);
+ return;
+ }
+ promise.setSuccess();
+ // Complete the seek callback after completing `promise`, otherwise `reconnectionPending_` will
+ // still be true when the seek operation is done.
+ LockGuard lock{mutex_};
+ if (seekStatus_ == SeekStatus::COMPLETED) {
+ executor_->postWork([seekCallback{std::exchange(seekCallback_, std::nullopt).value()}]() {
+ seekCallback(ResultOk);
+ });
+ seekStatus_ = SeekStatus::NOT_STARTED;
}
});
@@ -301,7 +312,8 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
LOG_INFO(getName() << "Closing subscribed consumer since it was already closed");
int requestId = client->newRequestId();
auto name = getName();
- cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId)
+ cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId,
+ "CLOSE_CONSUMER")
.addListener([name](Result result, const ResponseData&) {
if (result == ResultOk) {
LOG_INFO(name << "Closed consumer successfully after subscribe completed");
@@ -321,8 +333,8 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
return ResultAlreadyClosed;
}
+ mutexLock.unlock();
LOG_INFO(getName() << "Created consumer on broker " << cnx->cnxString());
- setCnx(cnx);
incomingMessages_.clear();
possibleSendToDeadLetterTopicMessages_.clear();
backoff_.reset();
@@ -354,7 +366,8 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
// in case it was indeed created, otherwise it might prevent new subscribe operation,
// since we are not closing the connection
auto requestId = newRequestId();
- cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId);
+ cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId,
+ "CLOSE_CONSUMER");
}
if (consumerCreatedPromise_.isComplete()) {
@@ -408,7 +421,7 @@ void ConsumerImpl::unsubscribeAsync(const ResultCallback& originalCallback) {
auto requestId = newRequestId();
SharedBuffer cmd = Commands::newUnsubscribe(consumerId_, requestId);
auto self = get_shared_this_ptr();
- cnx->sendRequestWithId(cmd, requestId)
+ cnx->sendRequestWithId(cmd, requestId, "UNSUBSCRIBE")
.addListener([self, callback](Result result, const ResponseData&) { callback(result); });
} else {
Result result = ResultNotConnected;
@@ -502,9 +515,10 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
auto& chunkedMsgCtx = it->second;
if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
- auto startMessageId = startMessageId_.get().value_or(MessageId::earliest());
- if (!config_.isStartMessageIdInclusive() && startMessageId.ledgerId() == messageId.ledgerId() &&
- startMessageId.entryId() == messageId.entryId()) {
+ auto startMessageId = getStartMessageId();
+ if (!config_.isStartMessageIdInclusive() && startMessageId &&
+ startMessageId->ledgerId() == messageId.ledgerId() &&
+ startMessageId->entryId() == messageId.entryId()) {
// When the start message id is not inclusive, the last chunk of the previous chunked message will
// be delivered, which is expected and we only need to filter it out.
chunkedMessageCache_.remove(uuid);
@@ -621,17 +635,14 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
words[i] = msg.ack_set(i);
}
BitSet ackSet{std::move(words)};
- Lock lock(mutex_);
numOfMessageReceived = receiveIndividualMessagesFromBatch(cnx, m, ackSet, msg.redelivery_count());
} else {
// try convert key value data.
m.impl_->convertPayloadToKeyValue(config_.getSchema());
- const auto startMessageId = startMessageId_.get();
- if (isPersistent_ && startMessageId &&
- m.getMessageId().ledgerId() == startMessageId.value().ledgerId() &&
- m.getMessageId().entryId() == startMessageId.value().entryId() &&
- isPriorEntryIndex(m.getMessageId().entryId())) {
+ const auto startMessageId = getStartMessageId();
+ if (isPersistent_ && startMessageId && m.getMessageId().ledgerId() == startMessageId->ledgerId() &&
+ isPrior(m.getMessageId().entryId(), startMessageId->entryId())) {
LOG_DEBUG(getName() << " Ignoring message from before the startMessageId: "
<< startMessageId.value());
return;
@@ -753,7 +764,7 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection
auto batchSize = batchedMessage.impl_->metadata.num_messages_in_batch();
LOG_DEBUG("Received Batch messages of size - " << batchSize
<< " -- msgId: " << batchedMessage.getMessageId());
- const auto startMessageId = startMessageId_.get();
+ const auto startMessageId = getStartMessageId();
int skippedMessages = 0;
@@ -783,9 +794,9 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection
// If we are receiving a batch message, we need to discard messages that were prior
// to the startMessageId
- if (isPersistent_ && msgId.ledgerId() == startMessageId.value().ledgerId() &&
- msgId.entryId() == startMessageId.value().entryId() &&
- isPriorBatchIndex(msgId.batchIndex())) {
+ if (isPersistent_ && msgId.ledgerId() == startMessageId->ledgerId() &&
+ msgId.entryId() == startMessageId->entryId() &&
+ isPrior(msgId.batchIndex(), startMessageId->batchIndex())) {
LOG_DEBUG(getName() << "Ignoring message from before the startMessageId"
<< msg.getMessageId());
++skippedMessages;
@@ -925,7 +936,7 @@ void ConsumerImpl::internalListener() {
trackMessage(msg.getMessageId());
try {
consumerStatsBasePtr_->receivedMessage(msg, ResultOk);
- lastDequedMessageId_ = msg.getMessageId();
+ setLastDequedMessageId(msg.getMessageId());
Consumer consumer{get_shared_this_ptr()};
Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), msg);
messageListener_(consumer, interceptMsg);
@@ -1098,10 +1109,7 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) {
}
void ConsumerImpl::messageProcessed(Message& msg, bool track) {
- Lock lock(mutexForMessageId_);
- lastDequedMessageId_ = msg.getMessageId();
- lock.unlock();
-
+ setLastDequedMessageId(msg.getMessageId());
incomingMessagesSize_.fetch_sub(msg.getLength());
ClientConnectionPtr currentCnx = getCnx().lock();
@@ -1123,20 +1131,22 @@ void ConsumerImpl::messageProcessed(Message& msg, bool track) {
* was
* not seen by the application
* `startMessageId_` is updated so that we can discard messages after delivery restarts.
+ * NOTE: `mutex_` must be locked before calling this method.
*/
void ConsumerImpl::clearReceiveQueue() {
- if (duringSeek()) {
- if (hasSoughtByTimestamp()) {
- // Invalidate startMessageId_ so that isPriorBatchIndex and isPriorEntryIndex checks will be
- // skipped, and hasMessageAvailableAsync won't use startMessageId_ in compare.
- startMessageId_ = std::nullopt;
+ if (seekStatus_ != SeekStatus::NOT_STARTED) {
+ // Flush the pending ACKs in case newly arrived messages are filtered out by the previous pending ACKs
+ ackGroupingTrackerPtr_->flushAndClean();
+ if (lastSeekArg_.has_value()) {
+ if (std::holds_alternative<MessageId>(lastSeekArg_.value())) {
+ startMessageId_ = std::get<MessageId>(lastSeekArg_.value());
+ } else {
+ // Invalidate startMessageId_ so that `isPrior` checks will be skipped, and
+ // `hasMessageAvailableAsync` won't use `startMessageId_` in compare.
+ startMessageId_ = std::nullopt;
+ }
} else {
- startMessageId_ = seekMessageId_.get();
- }
- SeekStatus expected = SeekStatus::COMPLETED;
- if (seekStatus_.compare_exchange_strong(expected, SeekStatus::NOT_STARTED)) {
- auto seekCallback = seekCallback_.release();
- executor_->postWork([seekCallback] { seekCallback(ResultOk); });
+ LOG_ERROR(getName() << "SeekStatus is not NOT_STARTED but lastSeekArg_ is not set");
}
return;
} else if (subscriptionMode_ == Commands::SubscriptionModeDurable) {
@@ -1374,7 +1384,7 @@ void ConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
auto requestId = newRequestId();
auto self = get_shared_this_ptr();
- cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId)
+ cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId, "CLOSE_CONSUMER")
.addListener([self, callback](Result result, const ResponseData&) { callback(result); });
}
@@ -1550,10 +1560,12 @@ void ConsumerImpl::seekAsync(const MessageId& msgId, const ResultCallback& callb
}
const auto requestId = newRequestId();
- seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, msgId), SeekArg{msgId}, callback);
+ auto nonNullCallback = (callback != nullptr) ? callback : [](Result) {};
+ seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, msgId), SeekArg{msgId},
+ std::move(nonNullCallback));
}
-void ConsumerImpl::seekAsync(uint64_t timestamp, const ResultCallback& callback) {
+void ConsumerImpl::seekAsync(SeekTimestampType timestamp, const ResultCallback& callback) {
const auto state = state_.load();
if (state == Closed || state == Closing) {
LOG_ERROR(getName() << "Client connection already closed.");
@@ -1564,8 +1576,9 @@ void ConsumerImpl::seekAsync(uint64_t timestamp, const ResultCallback& callback)
}
const auto requestId = newRequestId();
+ auto nonNullCallback = (callback != nullptr) ? callback : [](Result) {};
seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, timestamp), SeekArg{timestamp},
- callback);
+ std::move(nonNullCallback));
}
bool ConsumerImpl::isReadCompacted() { return readCompacted_; }
@@ -1577,16 +1590,16 @@ void ConsumerImpl::hasMessageAvailableAsync(const HasMessageAvailableCallback& c
}
bool compareMarkDeletePosition;
{
- std::lock_guard<std::mutex> lock{mutexForMessageId_};
+ LockGuard lock{mutex_};
compareMarkDeletePosition =
// there is no message received by consumer, so we cannot compare the last position with the last
// received position
lastDequedMessageId_ == MessageId::earliest() &&
// If the start message id is latest, we should seek to the actual last message first.
- (startMessageId_.get().value_or(MessageId::earliest()) == MessageId::latest() ||
+ (startMessageId_.value_or(MessageId::earliest()) == MessageId::latest() ||
// If there is a previous seek operation by timestamp, the start message id will be incorrect, so
// we cannot compare the start position with the last position.
- hasSoughtByTimestamp());
+ (lastSeekArg_.has_value() && std::holds_alternative<SeekTimestampType>(lastSeekArg_.value())));
}
if (compareMarkDeletePosition) {
auto self = get_shared_this_ptr();
@@ -1607,7 +1620,15 @@ void ConsumerImpl::hasMessageAvailableAsync(const HasMessageAvailableCallback& c
callback(ResultOk, false);
}
};
- if (self->config_.isStartMessageIdInclusive() && !self->hasSoughtByTimestamp()) {
+ bool lastSeekIsByTimestamp = false;
+ {
+ LockGuard lock{self->mutex_};
+ if (self->lastSeekArg_.has_value() &&
+ std::holds_alternative<SeekTimestampType>(self->lastSeekArg_.value())) {
+ lastSeekIsByTimestamp = true;
+ }
+ }
+ if (self->config_.isStartMessageIdInclusive() && !lastSeekIsByTimestamp) {
self->seekAsync(response.getLastMessageId(), [callback, handleResponse](Result result) {
if (result != ResultOk) {
callback(result, {});
@@ -1664,9 +1685,10 @@ void ConsumerImpl::internalGetLastMessageIdAsync(const BackoffPtr& backoff, Time
.addListener([this, self, callback](Result result, const GetLastMessageIdResponse& response) {
if (result == ResultOk) {
LOG_DEBUG(getName() << "getLastMessageId: " << response);
- Lock lock(mutexForMessageId_);
- lastMessageIdInBroker_ = response.getLastMessageId();
- lock.unlock();
+ {
+ LockGuard lock{mutex_};
+ lastMessageIdInBroker_ = response.getLastMessageId();
+ }
} else {
LOG_ERROR(getName() << "Failed to getLastMessageId: " << result);
}
@@ -1723,78 +1745,89 @@ bool ConsumerImpl::isConnected() const { return !getCnx().expired() && state_ ==
uint64_t ConsumerImpl::getNumberOfConnectedConsumer() { return isConnected() ? 1 : 0; }
void ConsumerImpl::seekAsyncInternal(long requestId, const SharedBuffer& seek, const SeekArg& seekArg,
- const ResultCallback& callback) {
+ ResultCallback&& callback) {
ClientConnectionPtr cnx = getCnx().lock();
if (!cnx) {
LOG_ERROR(getName() << " Client Connection not ready for Consumer");
callback(ResultNotConnected);
return;
}
-
- auto expected = SeekStatus::NOT_STARTED;
- if (!seekStatus_.compare_exchange_strong(expected, SeekStatus::IN_PROGRESS)) {
- LOG_ERROR(getName() << " attempted to seek " << seekArg << " when the status is "
- << static_cast<int>(expected));
+ bool hasPendingSeek = false;
+ // Save the previous last seek arg in case seek failed
+ decltype(lastSeekArg_) previousLastSeekArg;
+ {
+ std::lock_guard<std::mutex> lock(mutex_);
+ if (seekStatus_ != SeekStatus::NOT_STARTED) {
+ hasPendingSeek = true;
+ } else {
+ seekStatus_ = SeekStatus::IN_PROGRESS;
+ if (seekCallback_.has_value()) {
+ // This should never happen
+ LOG_ERROR(getName() << "Previous seek callback is not triggered unexpectedly");
+ executor_->postWork([callback{std::exchange(seekCallback_, std::nullopt).value()}] {
+ callback(ResultTimeout);
+ });
+ }
+ seekCallback_ = std::move(callback);
+ previousLastSeekArg = lastSeekArg_;
+ lastSeekArg_ = seekArg;
+ }
+ }
+ if (hasPendingSeek) {
+ std::visit(
+ [this](auto&& arg) {
+ LOG_ERROR(getName() << "Attempted to seek " << arg << " when there is a pending seek");
+ },
+ seekArg);
callback(ResultNotAllowedError);
return;
}
- const auto originalSeekMessageId = seekMessageId_.get();
- if (boost::get<uint64_t>(&seekArg)) {
- hasSoughtByTimestamp_.store(true, std::memory_order_release);
- } else {
- seekMessageId_ = *boost::get<MessageId>(&seekArg);
- hasSoughtByTimestamp_.store(false, std::memory_order_release);
- }
- seekStatus_ = SeekStatus::IN_PROGRESS;
- seekCallback_ = callback;
- LOG_INFO(getName() << " Seeking subscription to " << seekArg);
+ std::visit([this](auto&& arg) { LOG_INFO(getName() << "Seeking subscription to " << arg); }, seekArg);
auto weakSelf = weak_from_this();
- cnx->sendRequestWithId(seek, requestId)
- .addListener([this, weakSelf, callback, originalSeekMessageId](Result result,
- const ResponseData& responseData) {
+ cnx->sendRequestWithId(seek, requestId, "SEEK")
+ .addListener([this, weakSelf, previousLastSeekArg](Result result, const ResponseData& responseData) {
auto self = weakSelf.lock();
if (!self) {
- callback(result);
return;
}
if (result == ResultOk) {
- LOG_INFO(getName() << "Seek successfully");
- ackGroupingTrackerPtr_->flushAndClean();
- incomingMessages_.clear();
- Lock lock(mutexForMessageId_);
- lastDequedMessageId_ = MessageId::earliest();
- lock.unlock();
- if (getCnx().expired()) {
+ LockGuard lock(mutex_);
+ if (getCnx().expired() || reconnectionPending_) {
// It's during reconnection, complete the seek future after connection is established
seekStatus_ = SeekStatus::COMPLETED;
+ LOG_INFO(getName() << "Delay the seek future until the reconnection is done");
} else {
- if (!hasSoughtByTimestamp()) {
- startMessageId_ = seekMessageId_.get();
+ LOG_INFO(getName() << "Seek successfully");
+ ackGroupingTrackerPtr_->flushAndClean();
+ incomingMessages_.clear();
+ if (lastSeekArg_.has_value() && std::holds_alternative<MessageId>(lastSeekArg_.value())) {
+ startMessageId_ = std::get<MessageId>(lastSeekArg_.value());
}
- seekCallback_.release()(result);
- }
+ if (!seekCallback_.has_value()) {
+ LOG_ERROR(getName() << "Seek callback is not set");
+ return;
+ }
+ executor_->postWork(
+ [self, callback{std::exchange(seekCallback_, std::nullopt).value()}]() {
+ callback(ResultOk);
+ });
+ seekStatus_ = SeekStatus::NOT_STARTED;
+ } // else: complete the seek future after connection is established
} else {
LOG_ERROR(getName() << "Failed to seek: " << result);
- seekMessageId_ = originalSeekMessageId;
+ LockGuard lock{mutex_};
seekStatus_ = SeekStatus::NOT_STARTED;
- seekCallback_.release()(result);
+ lastSeekArg_ = previousLastSeekArg;
+ executor_->postWork([self, callback{std::exchange(seekCallback_, std::nullopt).value()}]() {
+ callback(ResultOk);
+ });
}
});
}
-bool ConsumerImpl::isPriorBatchIndex(int32_t idx) {
- return config_.isStartMessageIdInclusive() ? idx < startMessageId_.get().value().batchIndex()
- : idx <= startMessageId_.get().value().batchIndex();
-}
-
-bool ConsumerImpl::isPriorEntryIndex(int64_t idx) {
- return config_.isStartMessageIdInclusive() ? idx < startMessageId_.get().value().entryId()
- : idx <= startMessageId_.get().value().entryId();
-}
-
bool ConsumerImpl::hasEnoughMessagesForBatchReceive() const {
if (batchReceivePolicy_.getMaxNumMessages() <= 0 && batchReceivePolicy_.getMaxNumBytes() <= 0) {
return false;
@@ -1928,7 +1961,7 @@ void ConsumerImpl::doImmediateAck(const ClientConnectionPtr& cnx, const MessageI
auto requestId = newRequestId();
cnx->sendRequestWithId(
Commands::newAck(consumerId_, msgId.ledgerId(), msgId.entryId(), ackSet, ackType, requestId),
- requestId)
+ requestId, "ACK")
.addListener([callback](Result result, const ResponseData&) {
if (callback) {
callback(result);
@@ -1958,7 +1991,8 @@ void ConsumerImpl::doImmediateAck(const ClientConnectionPtr& cnx, const std::set
if (Commands::peerSupportsMultiMessageAcknowledgement(cnx->getServerProtocolVersion())) {
if (config_.isAckReceiptEnabled()) {
auto requestId = newRequestId();
- cnx->sendRequestWithId(Commands::newMultiMessageAck(consumerId_, ackMsgIds, requestId), requestId)
+ cnx->sendRequestWithId(Commands::newMultiMessageAck(consumerId_, ackMsgIds, requestId), requestId,
+ "ACK")
.addListener([callback](Result result, const ResponseData&) {
if (callback) {
callback(result);
diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h
index 63eb51d..0da82a2 100644
--- a/lib/ConsumerImpl.h
+++ b/lib/ConsumerImpl.h
@@ -21,13 +21,15 @@
#include <pulsar/Reader.h>
-#include <boost/variant.hpp>
+#include <atomic>
#include <cstdint>
#include <functional>
#include <list>
#include <memory>
+#include <optional>
#include <set>
#include <utility>
+#include <variant>
#include "BrokerConsumerStatsImpl.h"
#include "Commands.h"
@@ -37,7 +39,6 @@
#include "MapCache.h"
#include "MessageIdImpl.h"
#include "NegativeAcksTracker.h"
-#include "Synchronized.h"
#include "TestUtil.h"
#include "TimeUtils.h"
#include "UnboundedBlockingQueue.h"
@@ -79,9 +80,9 @@ const static std::string DLQ_GROUP_TOPIC_SUFFIX = "-DLQ";
enum class SeekStatus : std::uint8_t
{
- NOT_STARTED,
- IN_PROGRESS,
- COMPLETED
+ NOT_STARTED, // there is no pending seek RPC so that it's allowed to seek
+ IN_PROGRESS, // the seek RPC is in progress
+ COMPLETED // the seek RPC is done but the connection is not established yet
};
class ConsumerImpl : public ConsumerImplBase {
@@ -138,7 +139,9 @@ class ConsumerImpl : public ConsumerImplBase {
void getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback& callback) override;
void getLastMessageIdAsync(const BrokerGetLastMessageIdCallback& callback) override;
void seekAsync(const MessageId& msgId, const ResultCallback& callback) override;
- void seekAsync(uint64_t timestamp, const ResultCallback& callback) override;
+ using SeekTimestampType = uint64_t;
+ using SeekArg = std::variant<SeekTimestampType, MessageId>;
+ void seekAsync(SeekTimestampType timestamp, const ResultCallback& callback) override;
void negativeAcknowledge(const MessageId& msgId) override;
bool isConnected() const override;
uint64_t getNumberOfConnectedConsumer() override;
@@ -191,8 +194,10 @@ class ConsumerImpl : public ConsumerImplBase {
void drainIncomingMessageQueue(size_t count);
uint32_t receiveIndividualMessagesFromBatch(const ClientConnectionPtr& cnx, Message& batchedMessage,
const BitSet& ackSet, int redeliveryCount);
- bool isPriorBatchIndex(int32_t idx);
- bool isPriorEntryIndex(int64_t idx);
+ template <typename T>
+ bool isPrior(T index, T startIndex) const noexcept {
+ return config_.isStartMessageIdInclusive() ? (index < startIndex) : (index <= startIndex);
+ }
void brokerConsumerStatsListener(Result, BrokerConsumerStatsImpl, const BrokerConsumerStatsCallback&);
enum class DecryptionResult : uint8_t
@@ -218,19 +223,9 @@ class ConsumerImpl : public ConsumerImplBase {
const BrokerGetLastMessageIdCallback& callback);
void clearReceiveQueue();
- using SeekArg = boost::variant<uint64_t, MessageId>;
- friend std::ostream& operator<<(std::ostream& os, const SeekArg& seekArg) {
- auto ptr = boost::get<uint64_t>(&seekArg);
- if (ptr) {
- os << *ptr;
- } else {
- os << *boost::get<MessageId>(&seekArg);
- }
- return os;
- }
void seekAsyncInternal(long requestId, const SharedBuffer& seek, const SeekArg& seekArg,
- const ResultCallback& callback);
+ ResultCallback&& callback);
void processPossibleToDLQ(const MessageId& messageId, const ProcessDLQCallBack& cb);
std::mutex mutexForReceiveWithZeroQueueSize;
@@ -269,19 +264,13 @@ class ConsumerImpl : public ConsumerImplBase {
std::shared_ptr<Promise<Result, Producer>> deadLetterProducer_;
std::mutex createProducerLock_;
- // Make the access to `lastDequedMessageId_` and `lastMessageIdInBroker_` thread safe
- mutable std::mutex mutexForMessageId_;
MessageId lastDequedMessageId_{MessageId::earliest()};
MessageId lastMessageIdInBroker_{MessageId::earliest()};
+ optional<MessageId> startMessageId_;
- std::atomic<SeekStatus> seekStatus_{SeekStatus::NOT_STARTED};
- Synchronized<ResultCallback> seekCallback_{[](Result) {}};
- Synchronized<optional<MessageId>> startMessageId_;
- Synchronized<MessageId> seekMessageId_{MessageId::earliest()};
- std::atomic<bool> hasSoughtByTimestamp_{false};
-
- bool hasSoughtByTimestamp() const { return hasSoughtByTimestamp_.load(std::memory_order_acquire); }
- bool duringSeek() const { return seekStatus_ != SeekStatus::NOT_STARTED; }
+ SeekStatus seekStatus_{SeekStatus::NOT_STARTED};
+ optional<ResultCallback> seekCallback_;
+ optional<SeekArg> lastSeekArg_;
class ChunkedMessageCtx {
public:
@@ -380,7 +369,8 @@ class ConsumerImpl : public ConsumerImplBase {
const ClientConnectionPtr& cnx, MessageId& messageId);
bool hasMoreMessages() const {
- std::lock_guard<std::mutex> lock{mutexForMessageId_};
+ LockGuard lock{mutex_};
+
if (lastMessageIdInBroker_.entryId() == -1L) {
return false;
}
@@ -388,7 +378,7 @@ class ConsumerImpl : public ConsumerImplBase {
const auto inclusive = config_.isStartMessageIdInclusive();
if (lastDequedMessageId_ == MessageId::earliest()) {
// If startMessageId_ is none, use latest so that this method will return false
- const auto startMessageId = startMessageId_.get().value_or(MessageId::latest());
+ const auto startMessageId = startMessageId_.value_or(MessageId::latest());
return inclusive ? (lastMessageIdInBroker_ >= startMessageId)
: (lastMessageIdInBroker_ > startMessageId);
} else {
@@ -396,11 +386,22 @@ class ConsumerImpl : public ConsumerImplBase {
}
}
+ auto getStartMessageId() const {
+ LockGuard lock{mutex_};
+ return startMessageId_;
+ }
+ auto setLastDequedMessageId(const MessageId& messageId) {
+ LockGuard lock{mutex_};
+ lastDequedMessageId_ = messageId;
+ }
+
void doImmediateAck(const ClientConnectionPtr& cnx, const MessageId& msgId, CommandAck_AckType ackType,
const ResultCallback& callback);
void doImmediateAck(const ClientConnectionPtr& cnx, const std::set<MessageId>& msgIds,
const ResultCallback& callback);
+ using LockGuard = std::lock_guard<std::mutex>;
+
friend class PulsarFriend;
friend class MultiTopicsConsumerImpl;
diff --git a/lib/HandlerBase.cc b/lib/HandlerBase.cc
index 37c6e2d..1a4f573 100644
--- a/lib/HandlerBase.cc
+++ b/lib/HandlerBase.cc
@@ -44,9 +44,9 @@ HandlerBase::HandlerBase(const ClientImplPtr& client, const std::string& topic,
state_(NotStarted),
backoff_(backoff),
epoch_(0),
+ reconnectionPending_(false),
timer_(executor_->createDeadlineTimer()),
creationTimer_(executor_->createDeadlineTimer()),
- reconnectionPending_(false),
redirectedClusterURI_("") {}
HandlerBase::~HandlerBase() {
diff --git a/lib/HandlerBase.h b/lib/HandlerBase.h
index acce15d..0e733f0 100644
--- a/lib/HandlerBase.h
+++ b/lib/HandlerBase.h
@@ -143,6 +143,7 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
std::atomic<State> state_;
Backoff backoff_;
uint64_t epoch_;
+ std::atomic<bool> reconnectionPending_;
Result convertToTimeoutIfNecessary(Result result, ptime startTimestamp) const;
@@ -160,7 +161,6 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
DeadlineTimerPtr creationTimer_;
mutable std::mutex connectionMutex_;
- std::atomic<bool> reconnectionPending_;
ClientConnectionWeakPtr connection_;
std::string redirectedClusterURI_;
std::atomic<long> firstRequestIdAfterConnect_{-1L};
diff --git a/lib/MockServer.h b/lib/MockServer.h
new file mode 100644
index 0000000..8e4a213
--- /dev/null
+++ b/lib/MockServer.h
@@ -0,0 +1,139 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include <initializer_list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "ClientConnection.h"
+#include "ConsumerImpl.h"
+#include "ExecutorService.h"
+#include "LogUtils.h"
+#include "PulsarApi.pb.h"
+
+namespace pulsar {
+
+class MockServer : public std::enable_shared_from_this<MockServer> {
+ public:
+ using RequestDelayType = std::unordered_map<std::string, long /* delay in milliseconds */>;
+
+ MockServer(const ClientConnectionPtr& connection) : connection_(connection) {
+ requestDelays_["CLOSE_CONSUMER"] = 1;
+ }
+
+ void setRequestDelay(std::initializer_list<typename RequestDelayType::value_type> delays) {
+ std::lock_guard<std::mutex> lock(mutex_);
+ for (auto&& delay : delays) {
+ requestDelays_[delay.first] = delay.second;
+ }
+ }
+
+ bool sendRequest(const std::string& request, uint64_t requestId) {
+ auto connection = connection_.lock();
+ if (!connection) {
+ return false;
+ }
+ std::lock_guard<std::mutex> lock(mutex_);
+ if (auto iter = requestDelays_.find(request); iter != requestDelays_.end()) {
+ // Mock the `CLOSE_CONSUMER` command sent by broker, for simplicity, disconnect all consumers
+ if (request == "SEEK") {
+ schedule(connection, "CLOSE_CONSUMER" + std::to_string(requestId),
+ requestDelays_["CLOSE_CONSUMER"], [connection] {
+ std::vector<uint64_t> consumerIds;
+ {
+ std::lock_guard<std::mutex> lock{connection->mutex_};
+ for (auto&& kv : connection->consumers_) {
+ if (auto consumer = kv.second.lock()) {
+ consumerIds.push_back(consumer->getConsumerId());
+ }
+ }
+ }
+ for (auto consumerId : consumerIds) {
+ proto::CommandCloseConsumer closeConsumerCmd;
+ closeConsumerCmd.set_consumer_id(consumerId);
+ connection->handleCloseConsumer(closeConsumerCmd);
+ }
+ });
+ }
+ schedule(connection, request + std::to_string(requestId), iter->second, [connection, requestId] {
+ proto::CommandSuccess success;
+ success.set_request_id(requestId);
+ connection->handleSuccess(success);
+ });
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ // Return the number of pending timers cancelled
+ auto close() {
+ std::lock_guard<std::mutex> lock(mutex_);
+ auto result = pendingTimers_.size();
+ for (auto&& kv : pendingTimers_) {
+ try {
+ LOG_INFO("Cancelling timer for " << kv.first);
+ kv.second->cancel();
+ } catch (...) {
+ LOG_WARN("Failed to cancel timer for " << kv.first);
+ }
+ }
+ pendingTimers_.clear();
+ return result;
+ }
+
+ private:
+ mutable std::mutex mutex_;
+ std::unordered_map<std::string, long> requestDelays_;
+ std::unordered_map<std::string, DeadlineTimerPtr> pendingTimers_;
+ ClientConnectionWeakPtr connection_;
+
+ void schedule(ClientConnectionPtr& connection, const std::string& key, long delayMs,
+ std::function<void()>&& task) {
+ auto timer = connection->executor_->createDeadlineTimer();
+ pendingTimers_[key] = timer;
+ timer->expires_from_now(std::chrono::milliseconds(delayMs));
+ LOG_INFO("Mock scheduling " << key << " with delay " << delayMs << " ms");
+ auto self = shared_from_this();
+ timer->async_wait([this, self, key, connection, task{std::move(task)}](const auto& ec) {
+ {
+ std::lock_guard<std::mutex> lock(mutex_);
+ pendingTimers_.erase(key);
+ }
+ if (ec) {
+ LOG_INFO("Timer cancelled for " << key);
+ return;
+ }
+ if (connection->isClosed()) {
+ LOG_INFO("Connection is closed, not completing request " << key);
+ return;
+ }
+ LOG_INFO("Completing delayed request " << key);
+ task();
+ });
+ }
+
+ DECLARE_LOG_OBJECT()
+};
+
+} // namespace pulsar
diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
index 9d6a9a0..7fd14c7 100644
--- a/lib/ProducerImpl.cc
+++ b/lib/ProducerImpl.cc
@@ -160,7 +160,7 @@ Future<Result, bool> ProducerImpl::connectionOpened(const ClientConnectionPtr& c
// Keep a reference to ensure object is kept alive.
auto self = shared_from_this();
setFirstRequestIdAfterConnect(requestId);
- cnx->sendRequestWithId(cmd, requestId)
+ cnx->sendRequestWithId(cmd, requestId, "PRODUCER")
.addListener([this, self, cnx, promise](Result result, const ResponseData& responseData) {
Result handleResult = handleCreateProducer(cnx, result, responseData);
if (handleResult == ResultOk) {
@@ -204,7 +204,8 @@ Result ProducerImpl::handleCreateProducer(const ClientConnectionPtr& cnx, Result
auto client = client_.lock();
if (client) {
int requestId = client->newRequestId();
- cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId);
+ cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId,
+ "CLOSE_PRODUCER");
}
}
if (!producerCreatedPromise_.isComplete()) {
@@ -266,7 +267,8 @@ Result ProducerImpl::handleCreateProducer(const ClientConnectionPtr& cnx, Result
auto client = client_.lock();
if (client) {
int requestId = client->newRequestId();
- cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId);
+ cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId,
+ "CLOSE_PRODUCER");
}
}
@@ -818,7 +820,7 @@ void ProducerImpl::closeAsync(CloseCallback originalCallback) {
int requestId = client->newRequestId();
auto self = shared_from_this();
- cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId)
+ cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId, "CLOSE_PRODUCER")
.addListener([self, callback](Result result, const ResponseData&) { callback(result); });
}
diff --git a/lib/Synchronized.h b/lib/Synchronized.h
deleted file mode 100644
index 5449a9f..0000000
--- a/lib/Synchronized.h
+++ /dev/null
@@ -1,47 +0,0 @@
-/**
- * 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.
- */
-#pragma once
-
-#include <mutex>
-
-template <typename T>
-class Synchronized {
- public:
- explicit Synchronized(const T& value) : value_(value) {}
-
- T get() const {
- std::lock_guard<std::mutex> lock(mutex_);
- return value_;
- }
-
- T&& release() {
- std::lock_guard<std::mutex> lock(mutex_);
- return std::move(value_);
- }
-
- Synchronized& operator=(const T& value) {
- std::lock_guard<std::mutex> lock(mutex_);
- value_ = value;
- return *this;
- }
-
- private:
- T value_;
- mutable std::mutex mutex_;
-};
diff --git a/tests/ConsumerSeekTest.cc b/tests/ConsumerSeekTest.cc
index f03ea5e..f66c27d 100644
--- a/tests/ConsumerSeekTest.cc
+++ b/tests/ConsumerSeekTest.cc
@@ -19,12 +19,18 @@
#include <gtest/gtest.h>
#include <pulsar/Client.h>
+#include <chrono>
+#include <future>
+#include <memory>
#include <set>
#include <stdexcept>
#include <string>
#include "HttpHelper.h"
+#include "lib/ClientConnection.h"
#include "lib/LogUtils.h"
+#include "lib/MockServer.h"
+#include "tests/PulsarFriend.h"
DECLARE_LOG_OBJECT()
@@ -200,6 +206,58 @@ TEST_F(ConsumerSeekTest, testNoInternalConsumer) {
ASSERT_EQ(ResultOk, consumer.seek(MessageId::earliest()));
}
+static void assertSeekWithTimeout(Consumer& consumer) {
+ using namespace std::chrono_literals;
+ auto promise = std::make_shared<std::promise<Result>>();
+ std::weak_ptr<std::promise<Result>> weakPromise = promise;
+ consumer.seekAsync(0L, [weakPromise](Result result) {
+ if (auto promise = weakPromise.lock()) {
+ promise->set_value(result);
+ }
+ });
+ auto future = promise->get_future();
+ ASSERT_EQ(future.wait_for(5s), std::future_status::ready);
+ ASSERT_EQ(future.get(), ResultOk);
+}
+
+// Verify the `seek` method won't be blocked forever in any order of the Subscribe response and Seek response
+TEST_F(ConsumerSeekTest, testSubscribeSeekRaces) {
+ Client client(lookupUrl);
+ Consumer consumer;
+ ASSERT_EQ(ResultOk, client.subscribe("testSubscribeSeekRaces", "sub", consumer));
+
+ auto connection = *PulsarFriend::getConnections(client).begin();
+ auto mockServer = std::make_shared<MockServer>(connection);
+ connection->attachMockServer(mockServer);
+
+ mockServer->setRequestDelay({{"SUBSCRIBE", 1000}, {"SEEK", 500}});
+ assertSeekWithTimeout(consumer);
+
+ mockServer->setRequestDelay({{"SUBSCRIBE", 500}, {"SEEK", 1000}});
+ assertSeekWithTimeout(consumer);
+
+ ASSERT_EQ(mockServer->close(), 0);
+ client.close();
+}
+
+TEST_F(ConsumerSeekTest, testReconnectionSlow) {
+ Client client(lookupUrl, ClientConfiguration().setInitialBackoffIntervalMs(500));
+ Consumer consumer;
+ ASSERT_EQ(ResultOk, client.subscribe("testReconnectionSlow", "sub", consumer));
+
+ auto connection = *PulsarFriend::getConnections(client).begin();
+ auto mockServer = std::make_shared<MockServer>(connection);
+ connection->attachMockServer(mockServer);
+
+ // Make seek response received before `connectionOpened` is called
+ mockServer->setRequestDelay({{"SEEK", 500}, {"CLOSE_CONSUMER", 1000}});
+ assertSeekWithTimeout(consumer);
+
+ // The CLOSE_CONSUMER request is in still flight
+ ASSERT_EQ(mockServer->close(), 1);
+ client.close();
+}
+
INSTANTIATE_TEST_SUITE_P(Pulsar, ConsumerSeekTest, ::testing::Values(true, false));
} // namespace pulsar