[C++] Support key based batching #7996
Merged
Conversation
This file contains hidden or 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
This was referenced Sep 7, 2020
jiazhai
requested review from
codelipenghui,
jiazhai,
merlimat and
wolfstudy
jiazhai
approved these changes
Sep 7, 2020
lbenc135
pushed a commit
to lbenc135/pulsar
that referenced
this pull request
Oct 3, 2020
### Motivation Support key based batching for C++ client. This is a catch up work on apache#4435. In addition, currently the implementation of `BatchMessageContainer` is coupling to `ProducerImpl` tightly. The batch message container registers a timer to producer's executor and the timeout callback is also producer's method. Even its `add` method could call `sendMessage` to send batch to producer's pending queue. These should be producer's work. ### Modifications - Add a `MessageAndCallbackBatch` to store a `MessageImpl` of serialized single messages and a callback list. - Add a `BatchMessageContainerBase` to provide interface methods and methods like update/clear message number/bytes, create `OpSendMsg`. - Let `ProducerImpl` manage the batch timer and determine whether to create `OpSendMsg` from `BatchMessageContainerBase` and send. - Make `BatchMessageContainer` inherit `BatchMessageContainerBase`, it only manages a `MessageAndCallbackBatch`. - Add a `BatchMessageKeyBasedContainer` that inherits `BatchMessageContainerBase`, it manages a map of message key and `MessageAndCallbackBatch`. - Add a producer config to change batching type. - Add some units tests for key based batching and a unit test for key shared subscription. ### Verifying this change This change added tests and can be verified as follows: - Current tests affected by default batching type, like `BatchMessageTest.*` and `BasicEndToEndTest.*` - Newly added tests: `KeyBasedBatchingTest.*` and `KeySharedConsumerTest.testKeyBasedBatching` * Change ClientConnection::getMaxMessageSize() to static method * Refactor the BatchMessageContainer * Add batching type to producer config * Fix testBigMessageSizeBatching error * Add key based batching container * Make batching type config work * Add unit tests for key based batching and key shared subscription * Fix flush test error possibly caused by timeout overflow * Make BatchMessageKeyBasedContainer work for a single batch
Member
|
Move this change to 2.6.2, because #8191 depends on it. |
merlimat
pushed a commit
to merlimat/pulsar
that referenced
this pull request
Dec 19, 2020
### Motivation Support key based batching for C++ client. This is a catch up work on apache#4435. In addition, currently the implementation of `BatchMessageContainer` is coupling to `ProducerImpl` tightly. The batch message container registers a timer to producer's executor and the timeout callback is also producer's method. Even its `add` method could call `sendMessage` to send batch to producer's pending queue. These should be producer's work. ### Modifications - Add a `MessageAndCallbackBatch` to store a `MessageImpl` of serialized single messages and a callback list. - Add a `BatchMessageContainerBase` to provide interface methods and methods like update/clear message number/bytes, create `OpSendMsg`. - Let `ProducerImpl` manage the batch timer and determine whether to create `OpSendMsg` from `BatchMessageContainerBase` and send. - Make `BatchMessageContainer` inherit `BatchMessageContainerBase`, it only manages a `MessageAndCallbackBatch`. - Add a `BatchMessageKeyBasedContainer` that inherits `BatchMessageContainerBase`, it manages a map of message key and `MessageAndCallbackBatch`. - Add a producer config to change batching type. - Add some units tests for key based batching and a unit test for key shared subscription. ### Verifying this change This change added tests and can be verified as follows: - Current tests affected by default batching type, like `BatchMessageTest.*` and `BasicEndToEndTest.*` - Newly added tests: `KeyBasedBatchingTest.*` and `KeySharedConsumerTest.testKeyBasedBatching` * Change ClientConnection::getMaxMessageSize() to static method * Refactor the BatchMessageContainer * Add batching type to producer config * Fix testBigMessageSizeBatching error * Add key based batching container * Make batching type config work * Add unit tests for key based batching and key shared subscription * Fix flush test error possibly caused by timeout overflow * Make BatchMessageKeyBasedContainer work for a single batch
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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.
Motivation
Support key based batching for C++ client. This is a catch up work on #4435.
In addition, currently the implementation of
BatchMessageContaineris coupling toProducerImpltightly. The batch message container registers a timer to producer's executor and the timeout callback is also producer's method. Even itsaddmethod could callsendMessageto send batch to producer's pending queue. These should be producer's work.Modifications
MessageAndCallbackBatchto store aMessageImplof serialized single messages and a callback list.BatchMessageContainerBaseto provide interface methods and methods like update/clear message number/bytes, createOpSendMsg.ProducerImplmanage the batch timer and determine whether to createOpSendMsgfromBatchMessageContainerBaseand send.BatchMessageContainerinheritBatchMessageContainerBase, it only manages aMessageAndCallbackBatch.BatchMessageKeyBasedContainerthat inheritsBatchMessageContainerBase, it manages a map of message key andMessageAndCallbackBatch.Verifying this change
This change added tests and can be verified as follows:
BatchMessageTest.*andBasicEndToEndTest.*KeyBasedBatchingTest.*andKeySharedConsumerTest.testKeyBasedBatchingDocumentation