Add BatchedMessageIdImpl to acknowledge batched messages #132
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
BewareMyPower
force-pushed
the
bewaremypower/remove-ack-tracker
branch
from
c12340a to
1066111
Compare
Fixes apache#130 ### Motivation It's a catchup for apache/pulsar#1424 ### Modifications Migrate `BitSet` implementation from JDK. Though we have `vector<bool>` or `boost::dynamic_bitset`, to support batch index ACK in future, it's better to have the same method to convert a bit set to the underlying long array. Add `BatchedMessageIdImpl` to maintain a `BatchMessageAcker`, which is shared by messages in the same batch. The acker is responsible to record which messages are acknowledged. Only if all messages in the batch are acknowledged will the message id be acknowledged. The stats for individual ACKs only updates after the whole batch are acknowledged. Before this PR, each time the single message is acknowledged, the stats increase by one.
BewareMyPower
force-pushed
the
bewaremypower/remove-ack-tracker
branch
from
1066111 to
5df2b47
Compare
RobertIndie
reviewed
Dec 5, 2022
shibd
reviewed
Dec 6, 2022
| auto batchedMessageIdImpl = std::dynamic_pointer_cast<BatchedMessageIdImpl>(messageIdImpl); | ||
|
|
||
| auto batchSize = messageId.batchSize(); | ||
| if (!batchedMessageIdImpl || batchedMessageIdImpl->ackIndividual(messageId.batchIndex())) { |
There was a problem hiding this comment.
When messageId is not BatchedMessageIdImpl, the !batchedMessageIdImpl condition is true, right?
There was a problem hiding this comment.
Yes. In this case batchedMessageIdImpl will be a nullptr (0x0), see https://en.cppreference.com/w/cpp/memory/shared_ptr/pointer_cast
RobertIndie
approved these changes
Dec 7, 2022
BewareMyPower
added a commit
to BewareMyPower/pulsar-client-cpp
that referenced
this pull request
Dec 22, 2022
### Motivation The serialization and deserialization of `MessageId` became wrong after apache#132. 1. The batch size is not serialized. 2. `BatchedMessageIdImpl` could never be deserialized. The wrong behaviors could lead to a result that all MessageId objects created from deserialization does not have a batch size, which might make `ReaderTest.testReaderOnSpecificMessageWithBatches` fail when the cmake build type is `Debug`. What's worse is that a MessageId created from deserialization is always treated as a `MessageIdImpl`, on which the acknowledgment will have wrong behavior. ### Modifications Serialize the batch size if it's valid. In deserialization, create a `BatchedMessageIdImpl` when the batch index and the batch size are valid as a batched message. There is a problem that if a `MessageId` is created from deserialization, it cannot share a `BatchMessageAcker` with other `MessageId` objects. In this case, create a fake `BatchMessageAcker` that returns false for both `ackIndividual` and `ackCumulative` methods. It will make acknowledgment always fail but will fall back to batch index ACK if batch index ACK is enabled. Add the `-DCMAKE_BUILD_TYPE=Debug` for tests to enable assertions.
RobertIndie
pushed a commit
that referenced
this pull request
Dec 28, 2022
* [fix] Fix MessageId serialization when it's a batched message ### Motivation The serialization and deserialization of `MessageId` became wrong after #132. 1. The batch size is not serialized. 2. `BatchedMessageIdImpl` could never be deserialized. The wrong behaviors could lead to a result that all MessageId objects created from deserialization does not have a batch size, which might make `ReaderTest.testReaderOnSpecificMessageWithBatches` fail when the cmake build type is `Debug`. What's worse is that a MessageId created from deserialization is always treated as a `MessageIdImpl`, on which the acknowledgment will have wrong behavior. ### Modifications Serialize the batch size if it's valid. In deserialization, create a `BatchedMessageIdImpl` when the batch index and the batch size are valid as a batched message. There is a problem that if a `MessageId` is created from deserialization, it cannot share a `BatchMessageAcker` with other `MessageId` objects. In this case, create a fake `BatchMessageAcker` that returns false for both `ackIndividual` and `ackCumulative` methods. It will make acknowledgment always fail but will fall back to batch index ACK if batch index ACK is enabled. Add the `-DCMAKE_BUILD_TYPE=Debug` for tests to enable assertions. * Add virtual destructor
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Fixes #130
Motivation
It's a catchup for apache/pulsar#1424
Modifications
Migrate
BitSetimplementation from JDK. Though we havevector<bool>orboost::dynamic_bitset, to support batch index ACK in future, it's better to have the same method to convert a bit set to the underlying long array.Add
BatchedMessageIdImplto maintain aBatchMessageAcker, which is shared by messages in the same batch. The acker is responsible to record which messages are acknowledged. Only if all messages in the batch are acknowledged will the message id be acknowledged.The stats for individual ACKs only updates after the whole batch are acknowledged. Before this PR, each time the single message is acknowledged, the stats increase by one.