You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "shibd (via GitHub)" <gi...@apache.org> on 2023/03/23 14:39:26 UTC
[GitHub] [pulsar-client-cpp] shibd opened a new pull request, #228: [fix] Consumer batch receive will cause data loss.
shibd opened a new pull request, #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228
### Motivation
Currently, the batch receive method may cause data loss.
The root cause is, when `message` can't add to `messages`(The batch has satisfied the batch receive policy), it has actually been popped.
https://github.com/apache/pulsar-client-cpp/blob/ba7f59d2bb28b72fa6d1d01abada71090537b45e/lib/ConsumerImpl.cc#L652-L657
### Modifications
- Peek first, pop only when it can be added.
### Verifying this change
- Change `testBatchReceive` test to cover this case:
Send 100 messages first, batch receive 10 messages at a time, and assert that all messages can be received.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc-required`
(Your PR needs to update docs and you will update later)
- [x] `doc-not-needed`
(Please explain why)
- [ ] `doc`
(Your PR contains doc changes)
- [ ] `doc-complete`
(Docs have been already added)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1150009935
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
Don't assume any possible result caused by a race condition.
For example, what if `incomingMessages_` is empty when `pop` is called?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd commented on pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#issuecomment-1488356086
> Should we also check `isClosedNoMutex` for both `BlockingQueue` and `UnboundedBlockingQueue`?
oh, I missed it. added check, PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1147117852
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
The combination of `peek` and `pop` is not atomic. So you have to use a lock to synchronize these two operations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1147190108
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
Correctness is more important than performance. The race condition problem is hard to analyse if it really happened. We don't have to copy everything from the Java client, including the bugs.
BTW, the `peek` and `pop` methods in C++ client already acquire the lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1151311569
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
Make sense to me.
I added the `popIf` method on the queue. This avoids adding locks in multiple places in the consumer(all pop, clear, peek call places). PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd merged pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd merged PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1147174340
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
I noticed that Java client has the same problem.
https://github.com/apache/pulsar/blob/9ab9d0170a0a955875656d683839dd516c335bde/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L981-L999
> So you have to use a lock to synchronize these two operations.
But the lock feels a bit heavy because we also need to synchronize the `take` operation of `receive`.
https://github.com/apache/pulsar/blob/e0b50c9ec5f12d0fb8275f235d8ac00e87a9099e/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L445-L446
The current implementation does not lose or duplicate data, that just may return batches that exceed `batchReceivePolicy.MaxNumBytes`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1148279075
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
> The current implementation does not lose or duplicate data, that just may return batches that exceed batchReceivePolicy.MaxNumBytes.
I mean, if we don't think it's a bug, there's no need to lock it here.
Because we can't accurately control the batch returned, it must be equal to `batchReceivePolicy.MaxNumBytes`
- If we lock here, it is possible to return batches less than `batchReceivePolicy.MaxNumBytes`.
- If we don't lock here, it is possible to return batches greater than `batchReceivePolicy.MaxNumBytes`.
We can discuss it first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1148279075
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
> The current implementation does not lose or duplicate data, that just may return batches that exceed batchReceivePolicy.MaxNumBytes.
I mean, if we don't think it's a bug, there's no need to lock it here.
Because we can't accurately control the batch returned be equal to `batchReceivePolicy.MaxNumBytes`
- If we lock here, it is possible to return batches less than `batchReceivePolicy.MaxNumBytes`.
- If we don't lock here, it is possible to return batches greater than `batchReceivePolicy.MaxNumBytes`.
We can discuss it first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1147174340
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
I noticed that Java client has the same problem.
https://github.com/apache/pulsar/blob/9ab9d0170a0a955875656d683839dd516c335bde/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L981-L999
> So you have to use a lock to synchronize these two operations.
But the lock feels a bit heavy because we also need to synchronize the take operation of `receive`.
https://github.com/apache/pulsar/blob/e0b50c9ec5f12d0fb8275f235d8ac00e87a9099e/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L445-L446
The current implementation does not lose or duplicate data, that just may return batches that exceed `batchReceivePolicy.MaxNumBytes`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #228: [fix] Consumer batch receive will cause data loss.
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-cpp/pull/228#discussion_r1147190108
##########
lib/ConsumerImpl.cc:
##########
@@ -649,10 +649,11 @@ void ConsumerImpl::executeNotifyCallback(Message& msg) {
void ConsumerImpl::notifyBatchPendingReceivedCallback(const BatchReceiveCallback& callback) {
auto messages = std::make_shared<MessagesImpl>(batchReceivePolicy_.getMaxNumMessages(),
batchReceivePolicy_.getMaxNumBytes());
- Message peekMsg;
- while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
- messageProcessed(peekMsg);
- Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
+ Message msg;
+ while (incomingMessages_.peek(msg) && messages->canAdd(msg)) {
+ incomingMessages_.pop(msg);
Review Comment:
Correctness is more important than performance. The race condition problem is hard to analysis if it happened. We don't have to copy everything from the Java client, including the bugs.
BTW, the `peek` and `pop` methods in C++ client already acquire the lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org