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 2025/11/05 13:04:50 UTC
(pulsar-client-cpp) branch branch-3.8 updated (bf7acb2 -> d8fb3b7)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a change to branch branch-3.8
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
from bf7acb2 Replace Roaring64Map with unordered_set to unblock fat static library build issue (#514)
new 105c025 Fix APK package installation workflow (#515)
new 1720e53 Fix a null ACK grouping tracker can be accessed after consumer is closed (#517)
new d8fb3b7 Fix possible zombie consumer when closing after reconnection (#518)
The 3 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/AckGroupingTracker.cc | 141 -----------------------
lib/AckGroupingTracker.h | 40 +++----
lib/AckGroupingTrackerDisabled.cc | 29 +++--
lib/AckGroupingTrackerEnabled.cc | 70 ++++++++----
lib/AckGroupingTrackerEnabled.h | 26 ++---
lib/ConsumerImpl.cc | 231 ++++++++++++++++++++++++++------------
lib/ConsumerImpl.h | 22 +++-
lib/HandlerBase.h | 5 +
pkg/apk/APKBUILD | 7 +-
pkg/apk/build-apk.sh | 9 ++
tests/AcknowledgeTest.cc | 28 +++++
tests/BasicEndToEndTest.cc | 45 +++-----
tests/ConsumerTest.cc | 47 +++++++-
13 files changed, 379 insertions(+), 321 deletions(-)
delete mode 100644 lib/AckGroupingTracker.cc
(pulsar-client-cpp) 02/03: Fix a null ACK grouping tracker can be accessed after consumer is closed (#517)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 1720e53075304f10bf465edd24e51615d8bc71b3
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Oct 27 15:20:28 2025 +0800
Fix a null ACK grouping tracker can be accessed after consumer is closed (#517)
Fixes https://github.com/apache/pulsar-client-cpp/issues/516
(cherry picked from commit 648b48be58f65a15f45d952986f3d9d5936f75a8)
---
lib/AckGroupingTracker.cc | 141 --------------------------
lib/AckGroupingTracker.h | 40 +++-----
lib/AckGroupingTrackerDisabled.cc | 29 ++++--
lib/AckGroupingTrackerEnabled.cc | 70 +++++++++----
lib/AckGroupingTrackerEnabled.h | 26 ++---
lib/ConsumerImpl.cc | 202 +++++++++++++++++++++++++-------------
lib/ConsumerImpl.h | 22 ++++-
tests/AcknowledgeTest.cc | 28 ++++++
tests/BasicEndToEndTest.cc | 45 +++------
tests/ConsumerTest.cc | 25 +++--
10 files changed, 311 insertions(+), 317 deletions(-)
diff --git a/lib/AckGroupingTracker.cc b/lib/AckGroupingTracker.cc
deleted file mode 100644
index 8a9ea0d..0000000
--- a/lib/AckGroupingTracker.cc
+++ /dev/null
@@ -1,141 +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.
- */
-
-#include "AckGroupingTracker.h"
-
-#include <atomic>
-#include <limits>
-#include <set>
-
-#include "BitSet.h"
-#include "ChunkMessageIdImpl.h"
-#include "ClientConnection.h"
-#include "Commands.h"
-#include "LogUtils.h"
-#include "MessageIdImpl.h"
-
-namespace pulsar {
-
-DECLARE_LOG_OBJECT();
-
-void AckGroupingTracker::doImmediateAck(const MessageId& msgId, const ResultCallback& callback,
- CommandAck_AckType ackType) const {
- const auto cnx = connectionSupplier_();
- if (!cnx) {
- LOG_DEBUG("Connection is not ready, ACK failed for " << msgId);
- if (callback) {
- callback(ResultAlreadyClosed);
- }
- return;
- }
- if (ackType == CommandAck_AckType_Individual) {
- // If it's individual ack, we need to acknowledge all message IDs in a chunked message Id
- // If it's cumulative ack, we only need to ack the last message ID of a chunked message.
- // ChunkedMessageId return last chunk message ID by default, so we don't need to handle it.
- if (auto chunkMessageId =
- std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(msgId))) {
- auto msgIdList = chunkMessageId->getChunkedMessageIds();
- doImmediateAck(std::set<MessageId>(msgIdList.begin(), msgIdList.end()), callback);
- return;
- }
- }
- const auto& ackSet = Commands::getMessageIdImpl(msgId)->getBitSet();
- if (waitResponse_) {
- const auto requestId = requestIdSupplier_();
- cnx->sendRequestWithId(
- Commands::newAck(consumerId_, msgId.ledgerId(), msgId.entryId(), ackSet, ackType, requestId),
- requestId)
- .addListener([callback](Result result, const ResponseData&) {
- if (callback) {
- callback(result);
- }
- });
- } else {
- cnx->sendCommand(Commands::newAck(consumerId_, msgId.ledgerId(), msgId.entryId(), ackSet, ackType));
- if (callback) {
- callback(ResultOk);
- }
- }
-}
-
-static std::ostream& operator<<(std::ostream& os, const std::set<MessageId>& msgIds) {
- bool first = true;
- for (auto&& msgId : msgIds) {
- if (first) {
- first = false;
- } else {
- os << ", ";
- }
- os << "[" << msgId << "]";
- }
- return os;
-}
-
-void AckGroupingTracker::doImmediateAck(const std::set<MessageId>& msgIds,
- const ResultCallback& callback) const {
- const auto cnx = connectionSupplier_();
- if (!cnx) {
- LOG_DEBUG("Connection is not ready, ACK failed for " << msgIds);
- if (callback) {
- callback(ResultAlreadyClosed);
- }
- return;
- }
-
- std::set<MessageId> ackMsgIds;
-
- for (const auto& msgId : msgIds) {
- if (auto chunkMessageId =
- std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(msgId))) {
- auto msgIdList = chunkMessageId->getChunkedMessageIds();
- ackMsgIds.insert(msgIdList.begin(), msgIdList.end());
- } else {
- ackMsgIds.insert(msgId);
- }
- }
-
- if (Commands::peerSupportsMultiMessageAcknowledgement(cnx->getServerProtocolVersion())) {
- if (waitResponse_) {
- const auto requestId = requestIdSupplier_();
- cnx->sendRequestWithId(Commands::newMultiMessageAck(consumerId_, ackMsgIds, requestId), requestId)
- .addListener([callback](Result result, const ResponseData&) {
- if (callback) {
- callback(result);
- }
- });
- } else {
- cnx->sendCommand(Commands::newMultiMessageAck(consumerId_, ackMsgIds));
- if (callback) {
- callback(ResultOk);
- }
- }
- } else {
- auto count = std::make_shared<std::atomic<size_t>>(ackMsgIds.size());
- auto wrappedCallback = [callback, count](Result result) {
- if (--*count == 0 && callback) {
- callback(result);
- }
- };
- for (auto&& msgId : ackMsgIds) {
- doImmediateAck(msgId, wrappedCallback, CommandAck_AckType_Individual);
- }
- }
-}
-
-} // namespace pulsar
diff --git a/lib/AckGroupingTracker.h b/lib/AckGroupingTracker.h
index f62492f..d00c3a2 100644
--- a/lib/AckGroupingTracker.h
+++ b/lib/AckGroupingTracker.h
@@ -22,11 +22,7 @@
#include <pulsar/MessageId.h>
#include <pulsar/Result.h>
-#include <cstdint>
#include <functional>
-#include <set>
-
-#include "ProtoApiEnums.h"
namespace pulsar {
@@ -34,6 +30,9 @@ class ClientConnection;
using ClientConnectionPtr = std::shared_ptr<ClientConnection>;
using ClientConnectionWeakPtr = std::weak_ptr<ClientConnection>;
using ResultCallback = std::function<void(Result)>;
+class ConsumerImpl;
+using ConsumerImplPtr = std::shared_ptr<ConsumerImpl>;
+using ConsumerImplWeakPtr = std::weak_ptr<ConsumerImpl>;
/**
* @class AckGroupingTracker
@@ -42,19 +41,12 @@ using ResultCallback = std::function<void(Result)>;
*/
class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracker> {
public:
- AckGroupingTracker(std::function<ClientConnectionPtr()> connectionSupplier,
- std::function<uint64_t()> requestIdSupplier, uint64_t consumerId, bool waitResponse)
- : connectionSupplier_(std::move(connectionSupplier)),
- requestIdSupplier_(std::move(requestIdSupplier)),
- consumerId_(consumerId),
- waitResponse_(waitResponse) {}
-
virtual ~AckGroupingTracker() = default;
/**
* Start tracking the ACK requests.
*/
- virtual void start() {}
+ virtual void start(const ConsumerImplPtr& consumer) { consumer_ = consumer; }
/**
* Since ACK requests are grouped and delayed, we need to do some best-effort duplicate check to
@@ -72,7 +64,9 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
* @param[in] callback the callback that is triggered when the message is acknowledged
*/
virtual void addAcknowledge(const MessageId& msgId, const ResultCallback& callback) {
- callback(ResultOk);
+ if (callback) {
+ callback(ResultOk);
+ }
}
/**
@@ -81,7 +75,9 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
* @param[in] callback the callback that is triggered when the messages are acknowledged
*/
virtual void addAcknowledgeList(const MessageIdList& msgIds, const ResultCallback& callback) {
- callback(ResultOk);
+ if (callback) {
+ callback(ResultOk);
+ }
}
/**
@@ -90,7 +86,9 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
* @param[in] callback the callback that is triggered when the message is acknowledged
*/
virtual void addAcknowledgeCumulative(const MessageId& msgId, const ResultCallback& callback) {
- callback(ResultOk);
+ if (callback) {
+ callback(ResultOk);
+ }
}
/**
@@ -99,18 +97,10 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
*/
virtual void flushAndClean() {}
- protected:
- void doImmediateAck(const MessageId& msgId, const ResultCallback& callback,
- CommandAck_AckType ackType) const;
- void doImmediateAck(const std::set<MessageId>& msgIds, const ResultCallback& callback) const;
-
- private:
- const std::function<ClientConnectionPtr()> connectionSupplier_;
- const std::function<uint64_t()> requestIdSupplier_;
- const uint64_t consumerId_;
+ virtual void close() {}
protected:
- const bool waitResponse_;
+ ConsumerImplWeakPtr consumer_;
}; // class AckGroupingTracker
diff --git a/lib/AckGroupingTrackerDisabled.cc b/lib/AckGroupingTrackerDisabled.cc
index d20de44..92cdfba 100644
--- a/lib/AckGroupingTrackerDisabled.cc
+++ b/lib/AckGroupingTrackerDisabled.cc
@@ -19,26 +19,41 @@
#include "AckGroupingTrackerDisabled.h"
-#include "ProtoApiEnums.h"
+#include "ConsumerImpl.h"
namespace pulsar {
void AckGroupingTrackerDisabled::addAcknowledge(const MessageId& msgId, const ResultCallback& callback) {
- doImmediateAck(msgId, callback, CommandAck_AckType_Individual);
+ auto consumer = consumer_.lock();
+ if (consumer && !consumer->isClosingOrClosed()) {
+ consumer->doImmediateAck(msgId, callback, CommandAck_AckType_Individual);
+ } else if (callback) {
+ callback(ResultAlreadyClosed);
+ }
}
void AckGroupingTrackerDisabled::addAcknowledgeList(const MessageIdList& msgIds,
const ResultCallback& callback) {
- std::set<MessageId> msgIdSet;
- for (auto&& msgId : msgIds) {
- msgIdSet.emplace(msgId);
+ auto consumer = consumer_.lock();
+ if (consumer && !consumer->isClosingOrClosed()) {
+ std::set<MessageId> uniqueMsgIds(msgIds.begin(), msgIds.end());
+ for (auto&& msgId : msgIds) {
+ uniqueMsgIds.insert(msgId);
+ }
+ consumer->doImmediateAck(uniqueMsgIds, callback);
+ } else if (callback) {
+ callback(ResultAlreadyClosed);
}
- doImmediateAck(msgIdSet, callback);
}
void AckGroupingTrackerDisabled::addAcknowledgeCumulative(const MessageId& msgId,
const ResultCallback& callback) {
- doImmediateAck(msgId, callback, CommandAck_AckType_Cumulative);
+ auto consumer = consumer_.lock();
+ if (consumer && !consumer->isClosingOrClosed()) {
+ consumer->doImmediateAck(msgId, callback, CommandAck_AckType_Cumulative);
+ } else if (callback) {
+ callback(ResultAlreadyClosed);
+ }
}
} // namespace pulsar
diff --git a/lib/AckGroupingTrackerEnabled.cc b/lib/AckGroupingTrackerEnabled.cc
index d88426e..3a2a35d 100644
--- a/lib/AckGroupingTrackerEnabled.cc
+++ b/lib/AckGroupingTrackerEnabled.cc
@@ -23,11 +23,8 @@
#include <memory>
#include <mutex>
-#include "ClientConnection.h"
-#include "ClientImpl.h"
-#include "Commands.h"
+#include "ConsumerImpl.h"
#include "ExecutorService.h"
-#include "HandlerBase.h"
#include "MessageIdUtil.h"
namespace pulsar {
@@ -45,7 +42,10 @@ static int compare(const MessageId& lhs, const MessageId& rhs) {
}
}
-void AckGroupingTrackerEnabled::start() { this->scheduleTimer(); }
+void AckGroupingTrackerEnabled::start(const ConsumerImplPtr& consumer) {
+ AckGroupingTracker::start(consumer);
+ this->scheduleTimer();
+}
bool AckGroupingTrackerEnabled::isDuplicate(const MessageId& msgId) {
{
@@ -62,6 +62,13 @@ bool AckGroupingTrackerEnabled::isDuplicate(const MessageId& msgId) {
}
void AckGroupingTrackerEnabled::addAcknowledge(const MessageId& msgId, const ResultCallback& callback) {
+ auto consumer = consumer_.lock();
+ if (!consumer || consumer->isClosingOrClosed()) {
+ if (callback) {
+ callback(ResultAlreadyClosed);
+ }
+ return;
+ }
std::lock_guard<std::recursive_mutex> lock(this->rmutexPendingIndAcks_);
this->pendingIndividualAcks_.insert(msgId);
if (waitResponse_) {
@@ -70,12 +77,19 @@ void AckGroupingTrackerEnabled::addAcknowledge(const MessageId& msgId, const Res
callback(ResultOk);
}
if (this->ackGroupingMaxSize_ > 0 && this->pendingIndividualAcks_.size() >= this->ackGroupingMaxSize_) {
- this->flush();
+ this->flush(consumer);
}
}
void AckGroupingTrackerEnabled::addAcknowledgeList(const MessageIdList& msgIds,
const ResultCallback& callback) {
+ auto consumer = consumer_.lock();
+ if (!consumer || consumer->isClosingOrClosed()) {
+ if (callback) {
+ callback(ResultAlreadyClosed);
+ }
+ return;
+ }
std::lock_guard<std::recursive_mutex> lock(this->rmutexPendingIndAcks_);
for (const auto& msgId : msgIds) {
this->pendingIndividualAcks_.emplace(msgId);
@@ -86,12 +100,19 @@ void AckGroupingTrackerEnabled::addAcknowledgeList(const MessageIdList& msgIds,
callback(ResultOk);
}
if (this->ackGroupingMaxSize_ > 0 && this->pendingIndividualAcks_.size() >= this->ackGroupingMaxSize_) {
- this->flush();
+ this->flush(consumer);
}
}
void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId,
const ResultCallback& callback) {
+ auto consumer = consumer_.lock();
+ if (!consumer || consumer->isClosingOrClosed()) {
+ if (callback) {
+ callback(ResultAlreadyClosed);
+ }
+ return;
+ }
std::unique_lock<std::mutex> lock(this->mutexCumulativeAckMsgId_);
bool completeCallback = true;
if (compare(msgId, this->nextCumulativeAckMsgId_) > 0) {
@@ -115,23 +136,28 @@ void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId,
callback(ResultOk);
}
}
-
AckGroupingTrackerEnabled::~AckGroupingTrackerEnabled() {
- isClosed_ = true;
- this->flush();
std::lock_guard<std::mutex> lock(this->mutexTimer_);
if (this->timer_) {
cancelTimer(*this->timer_);
}
}
-void AckGroupingTrackerEnabled::flush() {
+void AckGroupingTrackerEnabled::close() {
+ flushAndClean();
+ std::lock_guard<std::mutex> lock(this->mutexTimer_);
+ if (this->timer_) {
+ cancelTimer(*this->timer_);
+ }
+}
+
+void AckGroupingTrackerEnabled::flush(const ConsumerImplPtr& consumer) {
// Send ACK for cumulative ACK requests.
{
std::lock_guard<std::mutex> lock(this->mutexCumulativeAckMsgId_);
if (this->requireCumulativeAck_) {
- this->doImmediateAck(this->nextCumulativeAckMsgId_, this->latestCumulativeCallback_,
- CommandAck_AckType_Cumulative);
+ consumer->doImmediateAck(this->nextCumulativeAckMsgId_, this->latestCumulativeCallback_,
+ CommandAck_AckType_Cumulative);
this->latestCumulativeCallback_ = nullptr;
this->requireCumulativeAck_ = false;
}
@@ -147,13 +173,17 @@ void AckGroupingTrackerEnabled::flush() {
callback(result);
}
};
- this->doImmediateAck(this->pendingIndividualAcks_, callback);
+ consumer->doImmediateAck(this->pendingIndividualAcks_, callback);
this->pendingIndividualAcks_.clear();
}
}
void AckGroupingTrackerEnabled::flushAndClean() {
- this->flush();
+ auto consumer = consumer_.lock();
+ if (!consumer) {
+ return;
+ }
+ this->flush(consumer);
{
std::lock_guard<std::mutex> lock(this->mutexCumulativeAckMsgId_);
this->nextCumulativeAckMsgId_ = MessageId::earliest();
@@ -165,10 +195,6 @@ void AckGroupingTrackerEnabled::flushAndClean() {
}
void AckGroupingTrackerEnabled::scheduleTimer() {
- if (isClosed_) {
- return;
- }
-
std::lock_guard<std::mutex> lock(this->mutexTimer_);
this->timer_ = this->executor_->createDeadlineTimer();
this->timer_->expires_after(std::chrono::milliseconds(std::max(1L, this->ackGroupingTimeMs_)));
@@ -176,7 +202,11 @@ void AckGroupingTrackerEnabled::scheduleTimer() {
this->timer_->async_wait([this, weakSelf](const ASIO_ERROR& ec) -> void {
auto self = weakSelf.lock();
if (self && !ec) {
- this->flush();
+ auto consumer = consumer_.lock();
+ if (!consumer || consumer->isClosingOrClosed()) {
+ return;
+ }
+ this->flush(consumer);
this->scheduleTimer();
}
});
diff --git a/lib/AckGroupingTrackerEnabled.h b/lib/AckGroupingTrackerEnabled.h
index 5eb04b9..eb2b449 100644
--- a/lib/AckGroupingTrackerEnabled.h
+++ b/lib/AckGroupingTrackerEnabled.h
@@ -21,8 +21,6 @@
#include <pulsar/MessageId.h>
-#include <atomic>
-#include <cstdint>
#include <mutex>
#include <set>
@@ -35,9 +33,6 @@ class ClientImpl;
using ClientImplPtr = std::shared_ptr<ClientImpl>;
class ExecutorService;
using ExecutorServicePtr = std::shared_ptr<ExecutorService>;
-class HandlerBase;
-using HandlerBasePtr = std::shared_ptr<HandlerBase>;
-using HandlerBaseWeakPtr = std::weak_ptr<HandlerBase>;
/**
* @class AckGroupingTrackerEnabled
@@ -45,34 +40,31 @@ using HandlerBaseWeakPtr = std::weak_ptr<HandlerBase>;
*/
class AckGroupingTrackerEnabled : public AckGroupingTracker {
public:
- AckGroupingTrackerEnabled(const std::function<ClientConnectionPtr()>& connectionSupplier,
- const std::function<uint64_t()>& requestIdSupplier, uint64_t consumerId,
- bool waitResponse, long ackGroupingTimeMs, long ackGroupingMaxSize,
+ AckGroupingTrackerEnabled(long ackGroupingTimeMs, long ackGroupingMaxSize, bool waitResponse,
const ExecutorServicePtr& executor)
- : AckGroupingTracker(connectionSupplier, requestIdSupplier, consumerId, waitResponse),
- ackGroupingTimeMs_(ackGroupingTimeMs),
+ : ackGroupingTimeMs_(ackGroupingTimeMs),
ackGroupingMaxSize_(ackGroupingMaxSize),
+ waitResponse_(waitResponse),
executor_(executor) {
pendingIndividualCallbacks_.reserve(ackGroupingMaxSize);
}
~AckGroupingTrackerEnabled();
- void start() override;
+ void start(const ConsumerImplPtr& consumer) override;
bool isDuplicate(const MessageId& msgId) override;
void addAcknowledge(const MessageId& msgId, const ResultCallback& callback) override;
void addAcknowledgeList(const MessageIdList& msgIds, const ResultCallback& callback) override;
void addAcknowledgeCumulative(const MessageId& msgId, const ResultCallback& callback) override;
- void flush();
void flushAndClean() override;
+ void close() override;
+
+ private:
+ void flush(const ConsumerImplPtr& consumer);
protected:
- //! Method for scheduling grouping timer.
void scheduleTimer();
- //! State
- std::atomic_bool isClosed_{false};
-
//! Next message ID to be cumulatively cumulatively.
MessageId nextCumulativeAckMsgId_{MessageId::earliest()};
bool requireCumulativeAck_{false};
@@ -90,6 +82,8 @@ class AckGroupingTrackerEnabled : public AckGroupingTracker {
//! Max number of ACK requests can be grouped.
const long ackGroupingMaxSize_;
+ const bool waitResponse_;
+
//! ACK request sender's scheduled executor.
const ExecutorServicePtr executor_;
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index 325adda..92d25cb 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -74,6 +74,22 @@ static boost::optional<MessageId> getStartMessageId(const boost::optional<Messag
return startMessageId;
}
+static std::shared_ptr<AckGroupingTracker> newAckGroupingTracker(const std::string& topic,
+ const ConsumerConfiguration& config,
+ const ClientImplPtr& client) {
+ if (TopicName::get(topic)->isPersistent()) {
+ if (config.getAckGroupingTimeMs() > 0) {
+ return std::make_shared<AckGroupingTrackerEnabled>(
+ config.getAckGroupingTimeMs(), config.getAckGroupingMaxSize(), config.isAckReceiptEnabled(),
+ client->getIOExecutorProvider()->get());
+ } else {
+ return std::make_shared<AckGroupingTrackerDisabled>();
+ }
+ } else {
+ return std::make_shared<AckGroupingTracker>();
+ }
+}
+
ConsumerImpl::ConsumerImpl(const ClientImplPtr& client, const std::string& topic,
const std::string& subscriptionName, const ConsumerConfiguration& conf,
bool isPersistent, const ConsumerInterceptorsPtr& interceptors,
@@ -105,12 +121,14 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr& client, const std::string& topic
consumerStr_("[" + topic + ", " + subscriptionName + ", " + std::to_string(consumerId_) + "] "),
messageListenerRunning_(!conf.isStartPaused()),
negativeAcksTracker_(std::make_shared<NegativeAcksTracker>(client, *this, conf)),
+ ackGroupingTrackerPtr_(newAckGroupingTracker(topic, conf, client)),
readCompacted_(conf.isReadCompacted()),
startMessageId_(getStartMessageId(startMessageId, conf.isStartMessageIdInclusive())),
maxPendingChunkedMessage_(conf.getMaxPendingChunkedMessage()),
autoAckOldestChunkedMessageOnQueueFull_(conf.isAutoAckOldestChunkedMessageOnQueueFull()),
expireTimeOfIncompleteChunkedMessageMs_(conf.getExpireTimeOfIncompleteChunkedMessageMs()),
- interceptors_(interceptors) {
+ interceptors_(interceptors),
+ requestIdGenerator_(client->getRequestIdGenerator()) {
// Initialize un-ACKed messages OT tracker.
if (conf.getUnAckedMessagesTimeoutMs() != 0) {
if (conf.getTickDurationInMs() > 0) {
@@ -169,9 +187,8 @@ ConsumerImpl::~ConsumerImpl() {
LOG_WARN(consumerStr_ << "Destroyed consumer which was not properly closed");
ClientConnectionPtr cnx = getCnx().lock();
- ClientImplPtr client = client_.lock();
- if (client && cnx) {
- int requestId = client->newRequestId();
+ if (cnx) {
+ auto requestId = newRequestId();
cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId);
cnx->removeConsumer(consumerId_);
LOG_INFO(consumerStr_ << "Closed consumer for race condition: " << consumerId_);
@@ -186,8 +203,6 @@ void ConsumerImpl::setPartitionIndex(int partitionIndex) { partitionIndex_ = par
int ConsumerImpl::getPartitionIndex() { return partitionIndex_; }
-uint64_t ConsumerImpl::getConsumerId() { return consumerId_; }
-
Future<Result, ConsumerImplBaseWeakPtr> ConsumerImpl::getConsumerCreatedFuture() {
return consumerCreatedPromise_.getFuture();
}
@@ -198,38 +213,7 @@ const std::string& ConsumerImpl::getTopic() const { return topic(); }
void ConsumerImpl::start() {
HandlerBase::start();
-
- std::weak_ptr<ConsumerImpl> weakSelf{get_shared_this_ptr()};
- auto connectionSupplier = [weakSelf]() -> ClientConnectionPtr {
- auto self = weakSelf.lock();
- if (!self) {
- return nullptr;
- }
- return self->getCnx().lock();
- };
-
- // NOTE: start() is always called in `ClientImpl`'s method, so lock() returns not null
- const auto requestIdGenerator = client_.lock()->getRequestIdGenerator();
- const auto requestIdSupplier = [requestIdGenerator] { return (*requestIdGenerator)++; };
-
- // Initialize ackGroupingTrackerPtr_ here because the get_shared_this_ptr() was not initialized until the
- // constructor completed.
- if (TopicName::get(topic())->isPersistent()) {
- if (config_.getAckGroupingTimeMs() > 0) {
- ackGroupingTrackerPtr_.reset(new AckGroupingTrackerEnabled(
- connectionSupplier, requestIdSupplier, consumerId_, config_.isAckReceiptEnabled(),
- config_.getAckGroupingTimeMs(), config_.getAckGroupingMaxSize(),
- client_.lock()->getIOExecutorProvider()->get()));
- } else {
- ackGroupingTrackerPtr_.reset(new AckGroupingTrackerDisabled(
- connectionSupplier, requestIdSupplier, consumerId_, config_.isAckReceiptEnabled()));
- }
- } else {
- LOG_INFO(getName() << "ACK will NOT be sent to broker for this non-persistent topic.");
- ackGroupingTrackerPtr_.reset(new AckGroupingTracker(connectionSupplier, requestIdSupplier,
- consumerId_, config_.isAckReceiptEnabled()));
- }
- ackGroupingTrackerPtr_->start();
+ ackGroupingTrackerPtr_->start(get_shared_this_ptr());
}
void ConsumerImpl::beforeConnectionChange(ClientConnection& cnx) { cnx.removeConsumer(consumerId_); }
@@ -265,7 +249,7 @@ Future<Result, bool> ConsumerImpl::connectionOpened(const ClientConnectionPtr& c
unAckedMessageTrackerPtr_->clear();
ClientImplPtr client = client_.lock();
- long requestId = client->newRequestId();
+ auto requestId = newRequestId();
SharedBuffer cmd = Commands::newSubscribe(
topic(), subscription_, consumerId_, requestId, getSubType(), getConsumerName(), subscriptionMode_,
subscribeMessageId, readCompacted_, config_.getProperties(), config_.getSubscriptionProperties(),
@@ -344,7 +328,7 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
// Creating the consumer has timed out. We need to ensure the broker closes the consumer
// in case it was indeed created, otherwise it might prevent new subscribe operation,
// since we are not closing the connection
- int requestId = client_.lock()->newRequestId();
+ auto requestId = newRequestId();
cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId);
}
@@ -396,7 +380,7 @@ void ConsumerImpl::unsubscribeAsync(const ResultCallback& originalCallback) {
LOG_DEBUG(getName() << "Unsubscribe request sent for consumer - " << consumerId_);
ClientImplPtr client = client_.lock();
lock.unlock();
- int requestId = client->newRequestId();
+ auto requestId = newRequestId();
SharedBuffer cmd = Commands::newUnsubscribe(consumerId_, requestId);
auto self = get_shared_this_ptr();
cnx->sendRequestWithId(cmd, requestId)
@@ -591,17 +575,16 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
LOG_DEBUG(getName() << " metadata.has_num_messages_in_batch() = "
<< metadata.has_num_messages_in_batch());
- uint32_t numOfMessageReceived = m.impl_->metadata.num_messages_in_batch();
- auto ackGroupingTrackerPtr = ackGroupingTrackerPtr_;
- if (ackGroupingTrackerPtr == nullptr) { // The consumer is closing
+ const auto state = state_.load(std::memory_order_relaxed);
+ if (state == Closing || state == Closed) {
return;
}
- if (ackGroupingTrackerPtr->isDuplicate(m.getMessageId())) {
+ uint32_t numOfMessageReceived = m.impl_->metadata.num_messages_in_batch();
+ if (ackGroupingTrackerPtr_->isDuplicate(m.getMessageId())) {
LOG_DEBUG(getName() << " Ignoring message as it was ACKed earlier by same consumer.");
increaseAvailablePermits(cnx, numOfMessageReceived);
return;
}
- ackGroupingTrackerPtr.reset();
if (metadata.has_num_messages_in_batch()) {
BitSet::Data words(msg.ack_set_size());
@@ -1340,12 +1323,7 @@ void ConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
incomingMessages_.close();
// Flush pending grouped ACK requests.
- if (ackGroupingTrackerPtr_.use_count() != 1) {
- LOG_ERROR("AckGroupingTracker is shared by other "
- << (ackGroupingTrackerPtr_.use_count() - 1)
- << " threads, which will prevent flushing the ACKs");
- }
- ackGroupingTrackerPtr_.reset();
+ ackGroupingTrackerPtr_->close();
negativeAcksTracker_->close();
ClientConnectionPtr cnx = getCnx().lock();
@@ -1364,7 +1342,7 @@ void ConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
cancelTimers();
- int requestId = client->newRequestId();
+ auto requestId = newRequestId();
auto self = get_shared_this_ptr();
cnx->sendRequestWithId(Commands::newCloseConsumer(consumerId_, requestId), requestId)
.addListener([self, callback](Result result, const ResponseData&) { callback(result); });
@@ -1375,7 +1353,7 @@ const std::string& ConsumerImpl::getName() const { return consumerStr_; }
void ConsumerImpl::shutdown() { internalShutdown(); }
void ConsumerImpl::internalShutdown() {
- ackGroupingTrackerPtr_.reset();
+ ackGroupingTrackerPtr_->close();
incomingMessages_.clear();
possibleSendToDeadLetterTopicMessages_.clear();
resetCnx();
@@ -1499,8 +1477,7 @@ void ConsumerImpl::getBrokerConsumerStatsAsync(const BrokerConsumerStatsCallback
ClientConnectionPtr cnx = getCnx().lock();
if (cnx) {
if (cnx->getServerProtocolVersion() >= proto::v8) {
- ClientImplPtr client = client_.lock();
- uint64_t requestId = client->newRequestId();
+ auto requestId = newRequestId();
LOG_DEBUG(getName() << " Sending ConsumerStats Command for Consumer - " << getConsumerId()
<< ", requestId - " << requestId);
@@ -1542,12 +1519,7 @@ void ConsumerImpl::seekAsync(const MessageId& msgId, const ResultCallback& callb
return;
}
- ClientImplPtr client = client_.lock();
- if (!client) {
- LOG_ERROR(getName() << "Client is expired when seekAsync " << msgId);
- return;
- }
- const auto requestId = client->newRequestId();
+ const auto requestId = newRequestId();
seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, msgId), SeekArg{msgId}, callback);
}
@@ -1561,12 +1533,7 @@ void ConsumerImpl::seekAsync(uint64_t timestamp, const ResultCallback& callback)
return;
}
- ClientImplPtr client = client_.lock();
- if (!client) {
- LOG_ERROR(getName() << "Client is expired when seekAsync " << timestamp);
- return;
- }
- const auto requestId = client->newRequestId();
+ const auto requestId = newRequestId();
seekAsyncInternal(requestId, Commands::newSeek(consumerId_, requestId, timestamp), SeekArg{timestamp},
callback);
}
@@ -1658,8 +1625,7 @@ void ConsumerImpl::internalGetLastMessageIdAsync(const BackoffPtr& backoff, Time
ClientConnectionPtr cnx = getCnx().lock();
if (cnx) {
if (cnx->getServerProtocolVersion() >= proto::v12) {
- ClientImplPtr client = client_.lock();
- uint64_t requestId = client->newRequestId();
+ auto requestId = newRequestId();
LOG_DEBUG(getName() << " Sending getLastMessageId Command for Consumer - " << getConsumerId()
<< ", requestId - " << requestId);
@@ -1926,4 +1892,100 @@ void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, const Proces
}
}
+void ConsumerImpl::doImmediateAck(const ClientConnectionPtr& cnx, const MessageId& msgId,
+ CommandAck_AckType ackType, const ResultCallback& callback) {
+ const auto& ackSet = Commands::getMessageIdImpl(msgId)->getBitSet();
+ if (config_.isAckReceiptEnabled()) {
+ auto requestId = newRequestId();
+ cnx->sendRequestWithId(
+ Commands::newAck(consumerId_, msgId.ledgerId(), msgId.entryId(), ackSet, ackType, requestId),
+ requestId)
+ .addListener([callback](Result result, const ResponseData&) {
+ if (callback) {
+ callback(result);
+ }
+ });
+ } else {
+ cnx->sendCommand(Commands::newAck(consumerId_, msgId.ledgerId(), msgId.entryId(), ackSet, ackType));
+ if (callback) {
+ callback(ResultOk);
+ }
+ }
+}
+
+void ConsumerImpl::doImmediateAck(const ClientConnectionPtr& cnx, const std::set<MessageId>& msgIds,
+ const ResultCallback& callback) {
+ std::set<MessageId> ackMsgIds;
+
+ for (const auto& msgId : msgIds) {
+ if (auto chunkMessageId =
+ std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(msgId))) {
+ auto msgIdList = chunkMessageId->getChunkedMessageIds();
+ ackMsgIds.insert(msgIdList.begin(), msgIdList.end());
+ } else {
+ ackMsgIds.insert(msgId);
+ }
+ }
+ if (Commands::peerSupportsMultiMessageAcknowledgement(cnx->getServerProtocolVersion())) {
+ if (config_.isAckReceiptEnabled()) {
+ auto requestId = newRequestId();
+ cnx->sendRequestWithId(Commands::newMultiMessageAck(consumerId_, ackMsgIds, requestId), requestId)
+ .addListener([callback](Result result, const ResponseData&) {
+ if (callback) {
+ callback(result);
+ }
+ });
+ } else {
+ cnx->sendCommand(Commands::newMultiMessageAck(consumerId_, ackMsgIds));
+ if (callback) {
+ callback(ResultOk);
+ }
+ }
+ } else {
+ auto count = std::make_shared<std::atomic<size_t>>(ackMsgIds.size());
+ auto wrappedCallback = [callback, count](Result result) {
+ if (--*count == 0 && callback) {
+ callback(result);
+ }
+ };
+ for (auto&& msgId : ackMsgIds) {
+ doImmediateAck(msgId, wrappedCallback, CommandAck_AckType_Individual);
+ }
+ }
+}
+
+void ConsumerImpl::doImmediateAck(const MessageId& msgId, const ResultCallback& callback,
+ CommandAck_AckType ackType) {
+ const auto cnx = getCnx().lock();
+ if (!cnx) {
+ if (callback) {
+ callback(ResultAlreadyClosed);
+ }
+ return;
+ }
+ if (ackType == CommandAck_AckType_Individual) {
+ // If it's individual ack, we need to acknowledge all message IDs in a chunked message Id
+ // If it's cumulative ack, we only need to ack the last message ID of a chunked message.
+ // ChunkedMessageId return last chunk message ID by default, so we don't need to handle it.
+ if (auto chunkMessageId =
+ std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(msgId))) {
+ auto msgIdList = chunkMessageId->getChunkedMessageIds();
+ doImmediateAck(cnx, std::set<MessageId>(msgIdList.begin(), msgIdList.end()), callback);
+ return;
+ }
+ }
+ doImmediateAck(cnx, msgId, ackType, callback);
+}
+
+void ConsumerImpl::doImmediateAck(const std::set<MessageId>& msgIds, const ResultCallback& callback) {
+ const auto cnx = getCnx().lock();
+ if (!cnx) {
+ if (callback) {
+ callback(ResultAlreadyClosed);
+ }
+ return;
+ }
+ doImmediateAck(cnx, msgIds, callback);
+}
+
} /* namespace pulsar */
diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h
index 055b487..5e06723 100644
--- a/lib/ConsumerImpl.h
+++ b/lib/ConsumerImpl.h
@@ -27,6 +27,7 @@
#include <functional>
#include <list>
#include <memory>
+#include <set>
#include <utility>
#include "BrokerConsumerStatsImpl.h"
@@ -96,7 +97,7 @@ class ConsumerImpl : public ConsumerImplBase {
void setPartitionIndex(int partitionIndex);
int getPartitionIndex();
void sendFlowPermitsToBroker(const ClientConnectionPtr& cnx, int numMessages);
- uint64_t getConsumerId();
+ uint64_t getConsumerId() const noexcept { return consumerId_; }
void messageReceived(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
bool& isChecksumValid, proto::BrokerEntryMetadata& brokerEntryMetadata,
proto::MessageMetadata& msgMetadata, SharedBuffer& payload);
@@ -124,6 +125,10 @@ class ConsumerImpl : public ConsumerImplBase {
void shutdown() override;
void internalShutdown();
bool isClosed() override;
+ bool isClosingOrClosed() const noexcept {
+ const auto state = state_.load(std::memory_order_relaxed);
+ return state == Closing || state == Closed;
+ }
bool isOpen() override;
Result pauseMessageListener() override;
Result resumeMessageListener() override;
@@ -152,6 +157,9 @@ class ConsumerImpl : public ConsumerImplBase {
void beforeConnectionChange(ClientConnection& cnx) override;
void onNegativeAcksSend(const std::set<MessageId>& messageIds);
+ void doImmediateAck(const MessageId& msgId, const ResultCallback& callback, CommandAck_AckType ackType);
+ void doImmediateAck(const std::set<MessageId>& msgIds, const ResultCallback& callback);
+
protected:
// overrided methods from HandlerBase
Future<Result, bool> connectionOpened(const ClientConnectionPtr& cnx) override;
@@ -237,7 +245,7 @@ class ConsumerImpl : public ConsumerImplBase {
std::queue<ReceiveCallback> pendingReceives_;
std::atomic_int availablePermits_;
const int receiverQueueRefillThreshold_;
- uint64_t consumerId_;
+ const uint64_t consumerId_;
const std::string consumerStr_;
int32_t partitionIndex_ = -1;
Promise<Result, ConsumerImplBaseWeakPtr> consumerCreatedPromise_;
@@ -246,7 +254,7 @@ class ConsumerImpl : public ConsumerImplBase {
UnAckedMessageTrackerPtr unAckedMessageTrackerPtr_;
BrokerConsumerStatsImpl brokerConsumerStats_;
std::shared_ptr<NegativeAcksTracker> negativeAcksTracker_;
- AckGroupingTrackerPtr ackGroupingTrackerPtr_;
+ const AckGroupingTrackerPtr ackGroupingTrackerPtr_;
MessageCryptoPtr msgCrypto_;
const bool readCompacted_;
@@ -340,6 +348,9 @@ class ConsumerImpl : public ConsumerImplBase {
std::atomic_bool expireChunkMessageTaskScheduled_{false};
ConsumerInterceptorsPtr interceptors_;
+ const std::shared_ptr<std::atomic<uint64_t>> requestIdGenerator_;
+
+ uint64_t newRequestId() const { return (*requestIdGenerator_)++; }
void triggerCheckExpiredChunkedTimer();
void discardChunkMessages(const std::string& uuid, const MessageId& messageId, bool autoAck);
@@ -379,6 +390,11 @@ class ConsumerImpl : public ConsumerImplBase {
}
}
+ 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);
+
friend class PulsarFriend;
friend class MultiTopicsConsumerImpl;
diff --git a/tests/AcknowledgeTest.cc b/tests/AcknowledgeTest.cc
index 0e2183e..464d5d2 100644
--- a/tests/AcknowledgeTest.cc
+++ b/tests/AcknowledgeTest.cc
@@ -375,6 +375,34 @@ TEST_F(AcknowledgeTest, testAckReceiptEnabled) {
client.close();
}
+TEST_F(AcknowledgeTest, testCloseConsumer) {
+ Client client(lookupUrl);
+ const auto topic = "test-close-consumer" + unique_str();
+ Producer producer;
+ ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
+ ConsumerConfiguration consumerConfig;
+ consumerConfig.setAckGroupingTimeMs(60000);
+ consumerConfig.setAckGroupingMaxSize(100);
+ Consumer consumer;
+ ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", consumerConfig, consumer));
+
+ producer.send(MessageBuilder().setContent("msg-0").build());
+ Message msg;
+ ASSERT_EQ(ResultOk, consumer.receive(msg, 3000));
+ consumer.acknowledgeAsync(
+ msg, nullptr); // it just adds the msg id to the pending ack list due to the ack grouping configs
+ consumer.close(); // it will flush the pending ACK and prevent any further ack
+ ASSERT_EQ(ResultAlreadyClosed, consumer.acknowledge(msg));
+ ASSERT_EQ(ResultAlreadyClosed, consumer.acknowledgeCumulative(msg));
+ ASSERT_EQ(ResultAlreadyClosed, consumer.acknowledge(std::vector<MessageId>{msg.getMessageId()}));
+
+ producer.send(MessageBuilder().setContent("msg-1").build());
+ // Recreate the consumer to verify the first message is acknowledged
+ ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", consumerConfig, consumer));
+ ASSERT_EQ(ResultOk, consumer.receive(msg, 3000));
+ ASSERT_EQ("msg-1", msg.getDataAsString());
+}
+
INSTANTIATE_TEST_SUITE_P(BasicEndToEndTest, AcknowledgeTest,
testing::Combine(testing::Values(100, 0), testing::Values(true, false)),
[](const testing::TestParamInfo<std::tuple<int, bool>>& info) {
diff --git a/tests/BasicEndToEndTest.cc b/tests/BasicEndToEndTest.cc
index 4330609..5cf478b 100644
--- a/tests/BasicEndToEndTest.cc
+++ b/tests/BasicEndToEndTest.cc
@@ -38,7 +38,6 @@
#include "lib/AckGroupingTrackerEnabled.h"
#include "lib/ClientConnection.h"
#include "lib/ClientImpl.h"
-#include "lib/Commands.h"
#include "lib/ConsumerImpl.h"
#include "lib/Future.h"
#include "lib/Latch.h"
@@ -3633,7 +3632,7 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerDefaultBehavior) {
ASSERT_EQ(configConsumer.getAckGroupingTimeMs(), 100);
ASSERT_EQ(configConsumer.getAckGroupingMaxSize(), 1000);
- AckGroupingTracker tracker{nullptr, nullptr, 0, false};
+ AckGroupingTracker tracker;
Message msg;
ASSERT_FALSE(tracker.isDuplicate(msg.getMessageId()));
}
@@ -3672,10 +3671,8 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerSingleAckBehavior) {
// Send ACK.
auto clientImplPtr = PulsarFriend::getClientImplPtr(client);
- AckGroupingTrackerDisabled tracker([&consumerImpl]() { return consumerImpl.getCnx().lock(); },
- [&clientImplPtr] { return clientImplPtr->newRequestId(); },
- consumerImpl.getConsumerId(), false);
- tracker.start();
+ AckGroupingTrackerDisabled tracker;
+ tracker.start(PulsarFriend::getConsumerImplPtr(consumer));
for (auto msgIdx = 0; msgIdx < numMsg; ++msgIdx) {
auto connPtr = connWeakPtr.lock();
ASSERT_NE(connPtr, nullptr);
@@ -3707,8 +3704,6 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerMultiAckBehavior) {
Consumer consumer;
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
- auto &consumerImpl = PulsarFriend::getConsumerImpl(consumer);
-
// Sending and receiving messages.
for (auto count = 0; count < numMsg; ++count) {
Message msg = MessageBuilder().setContent(std::string("MSG-") + std::to_string(count)).build();
@@ -3724,10 +3719,8 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerMultiAckBehavior) {
// Send ACK.
auto clientImplPtr = PulsarFriend::getClientImplPtr(client);
- AckGroupingTrackerDisabled tracker([&consumerImpl]() { return consumerImpl.getCnx().lock(); },
- [&clientImplPtr] { return clientImplPtr->newRequestId(); },
- consumerImpl.getConsumerId(), false);
- tracker.start();
+ AckGroupingTrackerDisabled tracker;
+ tracker.start(PulsarFriend::getConsumerImplPtr(consumer));
tracker.addAcknowledgeList(recvMsgId, nullptr);
consumer.close();
@@ -3755,7 +3748,6 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerDisabledIndividualAck) {
Consumer consumer;
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
- auto &consumerImpl = PulsarFriend::getConsumerImpl(consumer);
// Sending and receiving messages.
for (auto count = 0; count < numMsg; ++count) {
@@ -3771,8 +3763,8 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerDisabledIndividualAck) {
}
// Send ACK.
- AckGroupingTrackerDisabled tracker([&consumerImpl] { return consumerImpl.getCnx().lock(); }, nullptr,
- consumerImpl.getConsumerId(), false);
+ AckGroupingTrackerDisabled tracker;
+ tracker.start(PulsarFriend::getConsumerImplPtr(consumer));
for (auto &msgId : recvMsgId) {
tracker.addAcknowledge(msgId, nullptr);
}
@@ -3802,7 +3794,6 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerDisabledCumulativeAck) {
Consumer consumer;
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
- auto &consumerImpl = PulsarFriend::getConsumerImpl(consumer);
// Sending and receiving messages.
for (auto count = 0; count < numMsg; ++count) {
@@ -3818,8 +3809,8 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerDisabledCumulativeAck) {
}
// Send ACK.
- AckGroupingTrackerDisabled tracker([&consumerImpl] { return consumerImpl.getCnx().lock(); }, nullptr,
- consumerImpl.getConsumerId(), false);
+ AckGroupingTrackerDisabled tracker;
+ tracker.start(PulsarFriend::getConsumerImplPtr(consumer));
auto &latestMsgId = *std::max_element(recvMsgId.begin(), recvMsgId.end());
tracker.addAcknowledgeCumulative(latestMsgId, nullptr);
consumer.close();
@@ -3861,7 +3852,6 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerEnabledIndividualAck) {
Consumer consumer;
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
- auto consumerImpl = PulsarFriend::getConsumerImplPtr(consumer);
// Sending and receiving messages.
for (auto count = 0; count < numMsg; ++count) {
@@ -3877,9 +3867,8 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerEnabledIndividualAck) {
}
auto tracker = std::make_shared<AckGroupingTrackerEnabledMock>(
- [&consumerImpl] { return consumerImpl->getCnx().lock(); }, nullptr, consumerImpl->getConsumerId(),
- false, ackGroupingTimeMs, ackGroupingMaxSize, clientImplPtr->getIOExecutorProvider()->get());
- tracker->start();
+ ackGroupingTimeMs, ackGroupingMaxSize, false, clientImplPtr->getIOExecutorProvider()->get());
+ tracker->start(PulsarFriend::getConsumerImplPtr(consumer));
ASSERT_EQ(tracker->getPendingIndividualAcks().size(), 0);
ASSERT_EQ(tracker->getAckGroupingTimeMs(), ackGroupingTimeMs);
ASSERT_EQ(tracker->getAckGroupingMaxSize(), ackGroupingMaxSize);
@@ -3939,9 +3928,8 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerEnabledCumulativeAck) {
std::sort(recvMsgId.begin(), recvMsgId.end());
auto tracker0 = std::make_shared<AckGroupingTrackerEnabledMock>(
- [&consumerImpl0] { return consumerImpl0->getCnx().lock(); }, nullptr, consumerImpl0->getConsumerId(),
- false, ackGroupingTimeMs, ackGroupingMaxSize, clientImplPtr->getIOExecutorProvider()->get());
- tracker0->start();
+ ackGroupingTimeMs, ackGroupingMaxSize, false, clientImplPtr->getIOExecutorProvider()->get());
+ tracker0->start(PulsarFriend::getConsumerImplPtr(consumer));
ASSERT_EQ(tracker0->getNextCumulativeAckMsgId(), MessageId::earliest());
ASSERT_FALSE(tracker0->requireCumulativeAck());
@@ -3976,11 +3964,10 @@ TEST(BasicEndToEndTest, testAckGroupingTrackerEnabledCumulativeAck) {
auto ret = consumer.receive(msg, 1000);
ASSERT_EQ(ResultTimeout, ret) << "Received redundant message: " << msg.getDataAsString();
auto tracker1 = std::make_shared<AckGroupingTrackerEnabledMock>(
- [&consumerImpl1] { return consumerImpl1->getCnx().lock(); }, nullptr, consumerImpl1->getConsumerId(),
- false, ackGroupingTimeMs, ackGroupingMaxSize, clientImplPtr->getIOExecutorProvider()->get());
- tracker1->start();
+ ackGroupingTimeMs, ackGroupingMaxSize, false, clientImplPtr->getIOExecutorProvider()->get());
+ tracker1->start(PulsarFriend::getConsumerImplPtr(consumer));
tracker1->addAcknowledgeCumulative(recvMsgId[numMsg - 1], nullptr);
- tracker1.reset();
+ tracker1->close();
consumer.close();
ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumer));
diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc
index dfbc276..3aa1dd3 100644
--- a/tests/ConsumerTest.cc
+++ b/tests/ConsumerTest.cc
@@ -25,6 +25,7 @@
#include <atomic>
#include <chrono>
#include <ctime>
+#include <future>
#include <map>
#include <mutex>
#include <set>
@@ -1542,12 +1543,19 @@ TEST(ConsumerTest, testConsumerListenerShouldNotSegfaultAfterClose) {
consumerConfig.setSubscriptionInitialPosition(InitialPositionEarliest);
Latch latchFirstReceiveMsg(1);
Latch latchAfterClosed(1);
- consumerConfig.setMessageListener(
- [&latchFirstReceiveMsg, &latchAfterClosed](Consumer consumer, const Message& msg) {
- latchFirstReceiveMsg.countdown();
- LOG_INFO("Consume message: " << msg.getDataAsString());
- latchAfterClosed.wait();
- });
+
+ std::promise<std::vector<Result>> ackResultsPromise;
+ consumerConfig.setMessageListener([&latchFirstReceiveMsg, &latchAfterClosed, &ackResultsPromise](
+ Consumer consumer, const Message& msg) {
+ latchFirstReceiveMsg.countdown();
+ LOG_INFO("Consume message: " << msg.getDataAsString());
+ latchAfterClosed.wait();
+ std::vector<Result> results(3);
+ results[0] = consumer.acknowledge(msg);
+ results[1] = consumer.acknowledgeCumulative(msg);
+ results[2] = consumer.acknowledge(std::vector<MessageId>{msg.getMessageId()});
+ ackResultsPromise.set_value(results);
+ });
auto result = client.subscribe(topicName, "test-sub", consumerConfig, consumer);
ASSERT_EQ(ResultOk, result);
@@ -1555,6 +1563,11 @@ TEST(ConsumerTest, testConsumerListenerShouldNotSegfaultAfterClose) {
latchFirstReceiveMsg.wait();
ASSERT_EQ(ResultOk, consumer.close());
latchAfterClosed.countdown();
+ const auto ackResults = ackResultsPromise.get_future().get();
+ ASSERT_EQ(3, ackResults.size());
+ for (size_t i = 0; i < ackResults.size(); i++) {
+ ASSERT_EQ(ResultAlreadyClosed, ackResults[i]) << "ack result[" << i << "] " << ackResults[i];
+ }
ASSERT_EQ(ResultOk, producer.close());
ASSERT_EQ(ResultOk, client.close());
(pulsar-client-cpp) 03/03: Fix possible zombie consumer when closing after reconnection (#518)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit d8fb3b7f576b2cff39243274ac53a22b48e8154a
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Nov 5 18:58:52 2025 +0800
Fix possible zombie consumer when closing after reconnection (#518)
(cherry picked from commit 268aa4e2fdc621137a1a1fff9587ddd200bb0eda)
---
lib/ConsumerImpl.cc | 29 +++++++++++++++++++++++++++--
lib/HandlerBase.h | 5 +++++
tests/ConsumerTest.cc | 22 ++++++++++++++++++++++
3 files changed, 54 insertions(+), 2 deletions(-)
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index 92d25cb..3d5d294 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -293,13 +293,38 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
Result handleResult = ResultOk;
if (result == ResultOk) {
- LOG_INFO(getName() << "Created consumer on broker " << cnx->cnxString());
{
Lock mutexLock(mutex_);
+ if (!changeToReadyState()) {
+ auto client = client_.lock();
+ if (client) {
+ 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)
+ .addListener([name](Result result, const ResponseData&) {
+ if (result == ResultOk) {
+ LOG_INFO(name << "Closed consumer successfully after subscribe completed");
+ } else {
+ LOG_WARN(name << "Failed to close consumer: " << strResult(result));
+ }
+ });
+ } else {
+ // This should not happen normally because if client is destroyed, the connection pool
+ // should also be closed, which means all connections should be closed. Close the
+ // connection to let broker know this registered consumer is inactive.
+ LOG_WARN(getName()
+ << "Client already closed when subscribe completed, close the connection "
+ << cnx->cnxString());
+ cnx->close(ResultNotConnected);
+ }
+ return ResultAlreadyClosed;
+ }
+
+ LOG_INFO(getName() << "Created consumer on broker " << cnx->cnxString());
setCnx(cnx);
incomingMessages_.clear();
possibleSendToDeadLetterTopicMessages_.clear();
- state_ = Ready;
backoff_.reset();
if (!messageListener_ && config_.getReceiverQueueSize() == 0) {
// Complicated logic since we don't have a isLocked() function for mutex
diff --git a/lib/HandlerBase.h b/lib/HandlerBase.h
index df7dc24..967322f 100644
--- a/lib/HandlerBase.h
+++ b/lib/HandlerBase.h
@@ -148,6 +148,11 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
firstRequestIdAfterConnect_.store(requestId, std::memory_order_release);
}
+ bool changeToReadyState() noexcept {
+ State expected = Pending;
+ return state_ == Ready || state_.compare_exchange_strong(expected, Ready);
+ }
+
private:
DeadlineTimerPtr timer_;
DeadlineTimerPtr creationTimer_;
diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc
index 3aa1dd3..f1bca77 100644
--- a/tests/ConsumerTest.cc
+++ b/tests/ConsumerTest.cc
@@ -1573,4 +1573,26 @@ TEST(ConsumerTest, testConsumerListenerShouldNotSegfaultAfterClose) {
ASSERT_EQ(ResultOk, client.close());
}
+TEST(ConsumerTest, testCloseAfterSeek) {
+ const auto topic = "test-close-after-seek-" + std::to_string(time(nullptr));
+ const auto subscription = "sub";
+ Client client(lookupUrl);
+ Consumer consumer;
+ ASSERT_EQ(ResultOk, client.subscribe(topic, subscription, consumer));
+ ASSERT_EQ(ResultOk, consumer.seek(TimeUtils::currentTimeMillis()));
+ consumer.closeAsync(nullptr);
+
+ // Test the previous consumer will be closed even after seek is done, at the moment the connection might
+ // not be established.
+ ASSERT_EQ(ResultOk, client.subscribe(topic, subscription, consumer));
+
+ // Test creating a consumer from a different client should also work for this case
+ Client anotherClient(lookupUrl);
+ consumer.closeAsync(nullptr);
+ ASSERT_EQ(ResultOk, anotherClient.subscribe(topic, subscription, consumer));
+
+ client.close();
+ anotherClient.close();
+}
+
} // namespace pulsar
(pulsar-client-cpp) 01/03: Fix APK package installation workflow (#515)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 105c025f4f2bd9810237f5fd0337fdec60b4798a
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Oct 27 15:00:19 2025 +0800
Fix APK package installation workflow (#515)
* Fix wrong install directory for APK packages
* Fix DESTDIR
* Build libpulsarwithdeps.a
(cherry picked from commit 12471ec55099a51d0bcb9f551f83bb17a8358069)
---
pkg/apk/APKBUILD | 7 +++++--
pkg/apk/build-apk.sh | 9 +++++++++
2 files changed, 14 insertions(+), 2 deletions(-)
diff --git a/pkg/apk/APKBUILD b/pkg/apk/APKBUILD
index 1b476c2..85bb1ba 100644
--- a/pkg/apk/APKBUILD
+++ b/pkg/apk/APKBUILD
@@ -44,7 +44,7 @@ build() {
set +e
cmake -B build -DINTEGRATE_VCPKG=ON -DCMAKE_BUILD_TYPE=Release \
-DCMAKE_MAKE_PROGRAM=make \
- -DCMAKE_INSTALL_PREFIX="$pkgdir" \
+ -DCMAKE_INSTALL_PREFIX="/usr" \
-DBUILD_TESTS=OFF -DBUILD_DYNAMIC_LIB=ON -DBUILD_STATIC_LIB=ON \
$ROOT_DIR
if [[ $? -ne 0 ]]; then
@@ -53,8 +53,11 @@ build() {
fi
set -e
cmake --build build -j8
+ $ROOT_DIR/build-support/merge_archives_vcpkg.sh $PWD/build
}
package() {
- cmake --build build --target install
+ set -x -e
+ DESTDIR="$pkgdir" cmake --build build --target install
+ cp ./build/libpulsarwithdeps.a "$pkgdir/usr/lib/"
}
diff --git a/pkg/apk/build-apk.sh b/pkg/apk/build-apk.sh
index 5436048..68fdf89 100755
--- a/pkg/apk/build-apk.sh
+++ b/pkg/apk/build-apk.sh
@@ -40,3 +40,12 @@ sudo abuild-keygen -a -i -n
abuild -F -c -r
cp -r /root/packages/pkg ./build
+
+# Test installation
+apk add --allow-untrusted build/$PLATFORM/*.apk
+
+cd $ROOT_DIR/win-examples
+g++ -o dynamic.out -std=c++11 ./example.cc -Wl,-rpath=/usr/lib -lpulsar
+./dynamic.out
+g++ -o static.out -std=c++11 ./example.cc /usr/lib/libpulsarwithdeps.a -lpthread -ldl
+./static.out