Skip to content

Commit 0e91bbd

Browse files
BewareMyPowerBewareMyPower
committed
### Motivation
After a seek operation is done, the `startMessageId` will be updated until the reconnection due to the seek is done in `connectionOpened`. So before it's updated, `hasMessageAvailable` could compare with an outdated `startMessageId` and return a wrong value. ### Modifications Replace `duringSeek` with a `SeekStatus` field: - `NOT_STARTED`: initial, or a seek operation is done. `seek` could only succeed in this status. - `IN_PROGRESS`: A seek operation has started but the client does not receive the response from broker. - `COMPLETED`: The client has received the seek response but the seek future is not done. After the status becomes `COMPLETED`, if the connection is not ready, next time the connection is established, the status will change from `COMPLETED` to `NOT_STARTED` and then seek future will be completed in the internal executor. Add `testHasMessageAvailableAfterSeekToEnd` and `testSeekInProgress`.
1 parent 747c186 commit 0e91bbd

4 files changed

Lines changed: 131 additions & 34 deletions

File tree

‎lib/ConsumerImpl.cc‎

Lines changed: 45 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -236,16 +236,15 @@ Future<Result, bool> ConsumerImpl::connectionOpened(const ClientConnectionPtr& c
236236
// sending the subscribe request.
237237
cnx->registerConsumer(consumerId_, get_shared_this_ptr());
238238

239-
if (duringSeek_) {
239+
if (duringSeek()) {
240240
ackGroupingTrackerPtr_->flushAndClean();
241241
}
242242

243243
Lock lockForMessageId(mutexForMessageId_);
244244
// Update startMessageId so that we can discard messages after delivery restarts
245-
const auto startMessageId = clearReceiveQueue();
245+
clearReceiveQueue();
246246
const auto subscribeMessageId =
247-
(subscriptionMode_ == Commands::SubscriptionModeNonDurable) ? startMessageId : boost::none;
248-
startMessageId_ = startMessageId;
247+
(subscriptionMode_ == Commands::SubscriptionModeNonDurable) ? startMessageId_.get() : boost::none;
249248
lockForMessageId.unlock();
250249

251250
unAckedMessageTrackerPtr_->clear();
@@ -1049,13 +1048,19 @@ void ConsumerImpl::messageProcessed(Message& msg, bool track) {
10491048
* was
10501049
* not seen by the application
10511050
*/
1052-
boost::optional<MessageId> ConsumerImpl::clearReceiveQueue() {
1053-
bool expectedDuringSeek = true;
1054-
if (duringSeek_.compare_exchange_strong(expectedDuringSeek, false)) {
1055-
return seekMessageId_.get();
1051+
void ConsumerImpl::clearReceiveQueue() {
1052+
if (duringSeek()) {
1053+
startMessageId_ = seekMessageId_.get();
1054+
SeekStatus expected = SeekStatus::COMPLETED;
1055+
if (seekStatus_.compare_exchange_strong(expected, SeekStatus::NOT_STARTED)) {
1056+
auto seekCallback = seekCallback_.release();
1057+
executor_->postWork([seekCallback] { seekCallback(ResultOk); });
1058+
}
1059+
return;
10561060
} else if (subscriptionMode_ == Commands::SubscriptionModeDurable) {
1057-
return startMessageId_.get();
1061+
return;
10581062
}
1063+
10591064
Message nextMessageInQueue;
10601065
if (incomingMessages_.peekAndClear(nextMessageInQueue)) {
10611066
// There was at least one message pending in the queue
@@ -1071,16 +1076,12 @@ boost::optional<MessageId> ConsumerImpl::clearReceiveQueue() {
10711076
.ledgerId(nextMessageId.ledgerId())
10721077
.entryId(nextMessageId.entryId() - 1)
10731078
.build();
1074-
return previousMessageId;
1079+
startMessageId_ = previousMessageId;
10751080
} else if (lastDequedMessageId_ != MessageId::earliest()) {
10761081
// If the queue was empty we need to restart from the message just after the last one that has been
10771082
// dequeued
10781083
// in the past
1079-
return lastDequedMessageId_;
1080-
} else {
1081-
// No message was received or dequeued by this consumer. Next message would still be the
1082-
// startMessageId
1083-
return startMessageId_.get();
1084+
startMessageId_ = lastDequedMessageId_;
10841085
}
10851086
}
10861087

@@ -1500,17 +1501,10 @@ void ConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
15001501

15011502
bool ConsumerImpl::isReadCompacted() { return readCompacted_; }
15021503

1503-
inline bool hasMoreMessages(const MessageId& lastMessageIdInBroker, const MessageId& messageId) {
1504-
return lastMessageIdInBroker > messageId && lastMessageIdInBroker.entryId() != -1;
1505-
}
1506-
15071504
void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback) {
1508-
const auto startMessageId = startMessageId_.get();
15091505
Lock lock(mutexForMessageId_);
1510-
const auto messageId =
1511-
(lastDequedMessageId_ == MessageId::earliest()) ? startMessageId.value() : lastDequedMessageId_;
1512-
1513-
if (messageId == MessageId::latest()) {
1506+
if (lastDequedMessageId_ == MessageId::earliest() &&
1507+
startMessageId_.get().value_or(MessageId::earliest()) == MessageId::latest()) {
15141508
lock.unlock();
15151509
auto self = get_shared_this_ptr();
15161510
getLastMessageIdAsync([self, callback](Result result, const GetLastMessageIdResponse& response) {
@@ -1543,16 +1537,19 @@ void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback
15431537
}
15441538
});
15451539
} else {
1546-
if (hasMoreMessages(lastMessageIdInBroker_, messageId)) {
1540+
if (hasMoreMessages()) {
15471541
lock.unlock();
15481542
callback(ResultOk, true);
15491543
return;
15501544
}
15511545
lock.unlock();
15521546

1553-
getLastMessageIdAsync([callback, messageId](Result result, const GetLastMessageIdResponse& response) {
1554-
callback(result, (result == ResultOk) && hasMoreMessages(response.getLastMessageId(), messageId));
1555-
});
1547+
auto self = get_shared_this_ptr();
1548+
getLastMessageIdAsync(
1549+
[this, self, callback](Result result, const GetLastMessageIdResponse& response) {
1550+
std::lock_guard<std::mutex> lock{mutexForMessageId_};
1551+
callback(result, (result == ResultOk) && hasMoreMessages());
1552+
});
15561553
}
15571554
}
15581555

@@ -1656,9 +1653,18 @@ void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const Me
16561653
return;
16571654
}
16581655

1656+
auto expected = SeekStatus::NOT_STARTED;
1657+
if (!seekStatus_.compare_exchange_strong(expected, SeekStatus::IN_PROGRESS)) {
1658+
LOG_ERROR(getName() << " attempted to seek (" << seekId << ", " << timestamp << " when the status is "
1659+
<< static_cast<int>(expected));
1660+
callback(ResultNotAllowedError);
1661+
return;
1662+
}
1663+
16591664
const auto originalSeekMessageId = seekMessageId_.get();
16601665
seekMessageId_ = seekId;
1661-
duringSeek_ = true;
1666+
seekStatus_ = SeekStatus::IN_PROGRESS;
1667+
seekCallback_ = std::move(callback);
16621668
if (timestamp > 0) {
16631669
LOG_INFO(getName() << " Seeking subscription to " << timestamp);
16641670
} else {
@@ -1679,15 +1685,22 @@ void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const Me
16791685
LOG_INFO(getName() << "Seek successfully");
16801686
ackGroupingTrackerPtr_->flushAndClean();
16811687
incomingMessages_.clear();
1682-
Lock lock(mutexForMessageId_);
1688+
Lock lock{mutexForMessageId_};
16831689
lastDequedMessageId_ = MessageId::earliest();
16841690
lock.unlock();
1691+
if (getCnx().expired()) {
1692+
// It's during reconnection, complete the seek future after connection is established
1693+
seekStatus_ = SeekStatus::COMPLETED;
1694+
} else {
1695+
startMessageId_ = seekMessageId_.get();
1696+
seekCallback_.release()(result);
1697+
}
16851698
} else {
16861699
LOG_ERROR(getName() << "Failed to seek: " << result);
16871700
seekMessageId_ = originalSeekMessageId;
1688-
duringSeek_ = false;
1701+
seekStatus_ = SeekStatus::NOT_STARTED;
1702+
seekCallback_.release()(result);
16891703
}
1690-
callback(result);
16911704
});
16921705
}
16931706

‎lib/ConsumerImpl.h‎

Lines changed: 29 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,13 @@ const static std::string SYSTEM_PROPERTY_REAL_TOPIC = "REAL_TOPIC";
7575
const static std::string PROPERTY_ORIGIN_MESSAGE_ID = "ORIGIN_MESSAGE_ID";
7676
const static std::string DLQ_GROUP_TOPIC_SUFFIX = "-DLQ";
7777

78+
enum class SeekStatus : std::uint8_t
79+
{
80+
NOT_STARTED,
81+
IN_PROGRESS,
82+
COMPLETED
83+
};
84+
7885
class ConsumerImpl : public ConsumerImplBase {
7986
public:
8087
ConsumerImpl(const ClientImplPtr client, const std::string& topic, const std::string& subscriptionName,
@@ -193,7 +200,7 @@ class ConsumerImpl : public ConsumerImplBase {
193200
const DeadlineTimerPtr& timer,
194201
BrokerGetLastMessageIdCallback callback);
195202

196-
boost::optional<MessageId> clearReceiveQueue();
203+
void clearReceiveQueue();
197204
void seekAsyncInternal(long requestId, SharedBuffer seek, const MessageId& seekId, long timestamp,
198205
ResultCallback callback);
199206
void processPossibleToDLQ(const MessageId& messageId, ProcessDLQCallBack cb);
@@ -239,10 +246,13 @@ class ConsumerImpl : public ConsumerImplBase {
239246
MessageId lastDequedMessageId_{MessageId::earliest()};
240247
MessageId lastMessageIdInBroker_{MessageId::earliest()};
241248

242-
std::atomic_bool duringSeek_{false};
249+
std::atomic<SeekStatus> seekStatus_{SeekStatus::NOT_STARTED};
250+
Synchronized<ResultCallback> seekCallback_{[](Result) {}};
243251
Synchronized<boost::optional<MessageId>> startMessageId_;
244252
Synchronized<MessageId> seekMessageId_{MessageId::earliest()};
245253

254+
bool duringSeek() const { return seekStatus_ != SeekStatus::NOT_STARTED; }
255+
246256
class ChunkedMessageCtx {
247257
public:
248258
ChunkedMessageCtx() : totalChunks_(0) {}
@@ -332,6 +342,23 @@ class ConsumerImpl : public ConsumerImplBase {
332342
const proto::MessageIdData& messageIdData,
333343
const ClientConnectionPtr& cnx, MessageId& messageId);
334344

345+
// It must be called when mutexForMessageId_ is held
346+
bool hasMoreMessages() {
347+
if (lastMessageIdInBroker_.entryId() == -1L) {
348+
return false;
349+
}
350+
351+
const auto inclusive = config_.isStartMessageIdInclusive();
352+
if (lastDequedMessageId_ == MessageId::earliest()) {
353+
// If startMessageId_ is none, use latest so that this method will return false
354+
const auto startMessageId = startMessageId_.get().value_or(MessageId::latest());
355+
return inclusive ? (lastMessageIdInBroker_ >= startMessageId)
356+
: (lastMessageIdInBroker_ > startMessageId);
357+
} else {
358+
return lastMessageIdInBroker_ > lastDequedMessageId_;
359+
}
360+
}
361+
335362
friend class PulsarFriend;
336363
friend class MultiTopicsConsumerImpl;
337364

‎lib/Synchronized.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,11 @@ class Synchronized {
3030
return value_;
3131
}
3232

33+
T&& release() {
34+
std::lock_guard<std::mutex> lock(mutex_);
35+
return std::move(value_);
36+
}
37+
3338
Synchronized& operator=(const T& value) {
3439
std::lock_guard<std::mutex> lock(mutex_);
3540
value_ = value;

‎tests/ReaderTest.cc‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -784,4 +784,56 @@ TEST(ReaderSeekTest, testStartAtLatestMessageId) {
784784
producer.close();
785785
}
786786

787+
TEST(ReaderTest, testSeekInProgress) {
788+
Client client(serviceUrl);
789+
const auto topic = "test-seek-in-progress-" + std::to_string(time(nullptr));
790+
Reader reader;
791+
ASSERT_EQ(ResultOk, client.createReader(topic, MessageId::earliest(), {}, reader));
792+
793+
reader.seekAsync(MessageId::earliest(), [](Result) {});
794+
Promise<Result, Result> promise;
795+
reader.seekAsync(MessageId::earliest(), [promise](Result result) { promise.setValue(result); });
796+
Result result;
797+
promise.getFuture().get(result);
798+
ASSERT_EQ(result, ResultNotAllowedError);
799+
client.close();
800+
}
801+
802+
TEST_P(ReaderTest, testHasMessageAvailableAfterSeekToEnd) {
803+
Client client(serviceUrl);
804+
const auto topic = "test-has-message-available-after-seek-to-end-" + std::to_string(time(nullptr));
805+
Producer producer;
806+
ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
807+
Reader reader;
808+
ASSERT_EQ(ResultOk, client.createReader(topic, MessageId::earliest(), {}, reader));
809+
810+
producer.send(MessageBuilder().setContent("msg-0").build());
811+
producer.send(MessageBuilder().setContent("msg-1").build());
812+
813+
bool hasMessageAvailable;
814+
if (GetParam()) {
815+
// Test the case when `ConsumerImpl.lastMessageIdInBroker_` has been initialized
816+
reader.hasMessageAvailable(hasMessageAvailable);
817+
}
818+
819+
ASSERT_EQ(ResultOk, reader.seek(MessageId::latest()));
820+
ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
821+
ASSERT_FALSE(hasMessageAvailable);
822+
823+
producer.send(MessageBuilder().setContent("msg-2").build());
824+
ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
825+
ASSERT_TRUE(hasMessageAvailable);
826+
827+
Message msg;
828+
ASSERT_EQ(ResultOk, reader.readNext(msg, 1000));
829+
ASSERT_EQ("msg-2", msg.getDataAsString());
830+
831+
// Test the 2nd seek
832+
ASSERT_EQ(ResultOk, reader.seek(MessageId::latest()));
833+
ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
834+
ASSERT_FALSE(hasMessageAvailable);
835+
836+
client.close();
837+
}
838+
787839
INSTANTIATE_TEST_SUITE_P(Pulsar, ReaderTest, ::testing::Values(true, false));

0 commit comments

Comments
 (0)