Introduce batch message container framework and support key based batching container #4435

codelipenghui · 2019-06-01T02:54:07Z

Motivation

Introduce batch message container framework to support multiple ways to do message batch.
Currently, pulsar support a most basic batch message container, use the batch message container framework can quickly implement other types batch message container, even users can customize their own batch message container.

Add a new batch message container named BatchMessageKeyBasedContainer to support batching message in key_shared subscription mode.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (yes)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (no)

codelipenghui · 2019-06-04T07:45:58Z

relate to #4077

codelipenghui · 2019-06-04T07:57:40Z

/cc @wolfstudy please help review the changes

codelipenghui · 2019-06-04T09:40:55Z

run Integration Tests

codelipenghui · 2019-06-04T10:36:12Z

run java8 tests

codelipenghui · 2019-06-04T10:51:01Z

run Integration Tests

codelipenghui · 2019-06-04T22:41:48Z

run Integration Tests
run java8 tests

sijie · 2019-06-04T18:51:25Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatchMessageContainer.java

+     *
+     * @param ex cause
+     */
+    void handleException(Exception ex);


a better name might be discard? this method is basically releasing payload and the container when encountering exceptions, no?

sijie · 2019-06-04T18:52:01Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatchMessageContainer.java

+     *
+     * @return message batch size in bytes
+     */
+    long getCurrentBatchSizeBytes();


nit: getCurrentBatchSize is probably good enough. because size is typically in bytes.

sijie · 2019-06-05T06:28:08Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java

+    }
+
+    @Override
+    public List<OpSendMsg> createOpSendMsgs() throws IOException {


nit: it seems that we have to create one list for each batch when using normal batching mechanism. is it possible that we can reduce this overhead when key based batching is not used?

Good point, i think can add a method isMultiBatch, if isMultiBatch is true call createOpSendMsgs() , otherwise call createOpSendMsg()

sijie · 2019-06-05T06:29:47Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java

+        return msg.getKey();
+    }
+
+    private class KeyBasedPart {


can we make the class static?

Also is KeyedBatch a better name?

sijie · 2019-06-05T06:30:32Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java

+        return result;
+    }
+
+    private String peekKey(MessageImpl<?> msg) {


peek sounds like an action to retrieve a key from a queue. why not just call it getKey?

sijie · 2019-06-05T06:33:41Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java

+        if (part == null) {
+            part = new KeyBasedPart();
+            part.addMsg(msg, callback);
+            batches.putIfAbsent(key, part);


I think there is a race condition here. if concurrent puts happen, one put will fail to add the message to the part. hence there is data loss.

Before add message to the container, producer already with synchronized semantics. BatchMessageKeyBasedContainer and BatchMessageContainerImpl is not thread-safe, if we want to make sure they are thread-safe, we need to change some other fields to thread-safe(e.g. numMessagesInBatch, currentBatchSizeBytes), is it necessary for us to do so?

@codelipenghui if it is not threadsafe, why do we need to use ConcurrentHashMap? Can't we just use a HashMap?

A good practice is - if you are using ConcurrentMap, it might be worth making it thread-safe. Otherwise it will be hard to maintain the code, since people will be confused about why a ConcurrentMap is used here.

Yes, we can just use a HashMap.

sijie · 2019-06-05T06:34:04Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java

+    @Override
+    public void add(MessageImpl<?> msg, SendCallback callback) {
+        if (log.isDebugEnabled()) {
+            log.debug("[{}] [{}] add message to batch, num messages in batch so far {}", topicName, producerName,


Suggested change

log.debug("[{}] [{}] add message to batch, num messages in batch so far {}", topicName, producerName,

log.debug("[{}] [{}] add message to batch, num messages in batch so far is {}", topicName, producerName,

sijie · 2019-06-05T06:41:19Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java

@@ -223,6 +224,13 @@ private ProducerBuilderImpl(PulsarClientImpl client, ProducerConfigurationData c
        return this;
    }

+    @Override
+    public ProducerBuilder<T> batchingContainerBuilder(BatchMessageContainerBuilder batchingContainerBuilder) {


I am not sure if batchingContainerBuilder is the best name. maybe MessageBatcher?

I've also struggled with this problem, if we named messageBatcher, It looks like it's provided with a message batcher, but in fact, it provides the ability to build a message batcher. because if it is a partitioned topic, each partitions producer can not use same message batcher, so i named batchingContainerBuilder. This naming seems a bit complicated, how about batcherBuilder ?

Batcher sounds like a better name.

sijie · 2019-06-05T06:41:58Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java

+     *          batch message container builder
+     * @return the producer builder instance
+     */
+    ProducerBuilder<T> batchingContainerBuilder(BatchMessageContainerBuilder batchingContainerBuilder);


add a comment that this is only be used when batching is enabled

codelipenghui · 2019-06-10T06:49:48Z

@sijie I'm already address all your comments, PTAL.

codelipenghui · 2019-06-10T09:22:58Z

run cpp tests
run java8 tests

codelipenghui · 2019-06-10T10:54:52Z

run java8 tests

codelipenghui · 2019-06-11T08:27:02Z

run cpp tests
run java8 tests

codelipenghui · 2019-06-13T08:24:49Z

@merlimat @jiazhai Please help review this PR if you have time.

sijie · 2019-06-17T00:09:57Z

ping @merlimat @jiazhai

jiazhai

+1. lgtm.
Thanks for the work.

### Motivation Support key based batching for C++ client. This is a catch up work on #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

### 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

codelipenghui changed the title ~~Introduce batch message container framework~~ [WIP] Key based batch message container Jun 1, 2019

codelipenghui force-pushed the batch_message_container_framework branch from d2bcfec to c5a04b5 Compare June 1, 2019 13:43

codelipenghui changed the title ~~[WIP] Key based batch message container~~ Introduce batch message container framework and support key based batching container Jun 4, 2019

codelipenghui requested review from merlimat, sijie and jiazhai June 4, 2019 07:41

codelipenghui mentioned this pull request Jun 4, 2019

PIP-34 New subscription type Key_Shared #4077

Closed

12 tasks

sijie assigned codelipenghui Jun 4, 2019

sijie added area/broker area/client type/feature The PR added a new feature or issue requested a new feature labels Jun 4, 2019

sijie requested changes Jun 5, 2019

View reviewed changes

codelipenghui requested a review from sijie June 6, 2019 03:02

codelipenghui and others added 9 commits June 10, 2019 10:42

Introduce batch message container framework.

e4dfa86

Fix UI issues.

b4d404e

Add batch message container builder.

3b986f9

Add key based batch message container.

5d5806e

Add document for batch message containers.

199ef13

Fix configuration data test issue.

b5c341b

Fix storm UT issue

9b648f0

fix review comments.

1657d24

clean code of UT.

3a6e788

codelipenghui force-pushed the batch_message_container_framework branch from 134f9dd to 3a6e788 Compare June 10, 2019 03:05

fix review comments.

c8810ff

rename to batcher builder.

7089cbe

remove unused import.

5a5df34

remove unused code.

a84f88a

sijie approved these changes Jun 12, 2019

View reviewed changes

codelipenghui added this to the 2.4.0 milestone Jun 13, 2019

jiazhai approved these changes Jun 19, 2019

View reviewed changes

sijie merged commit b45736a into apache:master Jun 19, 2019

sijie mentioned this pull request Dec 27, 2019

ISSUE-4077: PIP-34 New subscription type Key_Shared streamnative/pulsar-archived#32

Closed

15 tasks

wolfstudy mentioned this pull request Jan 14, 2020

Support key based batcher for key_shared apache/pulsar-client-go#172

Closed

sijie mentioned this pull request Jan 15, 2020

ISSUE-172: Support key based batcher for key_shared streamnative/pulsar-client-go#22

Closed

BewareMyPower mentioned this pull request Sep 7, 2020

[C++] Support key based batching #7996

Merged

1 task

codelipenghui deleted the batch_message_container_framework branch May 19, 2021 05:37

BewareMyPower mentioned this pull request May 6, 2022

[Java Client] Fix wrong behavior of deduplication for key based batching #15413

Closed

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce batch message container framework and support key based batching container #4435

Introduce batch message container framework and support key based batching container #4435

codelipenghui commented Jun 1, 2019 •

edited

Loading

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

sijie Jun 4, 2019

sijie Jun 4, 2019

sijie Jun 5, 2019

codelipenghui Jun 5, 2019

sijie Jun 5, 2019

sijie Jun 5, 2019

sijie Jun 5, 2019

codelipenghui Jun 5, 2019 •

edited

Loading

sijie Jun 10, 2019

codelipenghui Jun 10, 2019

sijie Jun 5, 2019

sijie Jun 5, 2019

codelipenghui Jun 5, 2019 •

edited

Loading

sijie Jun 10, 2019

sijie Jun 5, 2019

codelipenghui commented Jun 10, 2019

codelipenghui commented Jun 10, 2019

codelipenghui commented Jun 10, 2019

codelipenghui commented Jun 11, 2019

codelipenghui commented Jun 13, 2019

sijie commented Jun 17, 2019

jiazhai left a comment

	log.debug("[{}] [{}] add message to batch, num messages in batch so far {}", topicName, producerName,
	log.debug("[{}] [{}] add message to batch, num messages in batch so far is {}", topicName, producerName,

Introduce batch message container framework and support key based batching container #4435

Introduce batch message container framework and support key based batching container #4435

Conversation

codelipenghui commented Jun 1, 2019 • edited Loading

Motivation

Does this pull request potentially affect one of the following parts:

Documentation

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

codelipenghui commented Jun 4, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui Jun 5, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui Jun 5, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Jun 10, 2019

codelipenghui commented Jun 10, 2019

codelipenghui commented Jun 10, 2019

codelipenghui commented Jun 11, 2019

codelipenghui commented Jun 13, 2019

sijie commented Jun 17, 2019

jiazhai left a comment

Choose a reason for hiding this comment

codelipenghui commented Jun 1, 2019 •

edited

Loading

codelipenghui Jun 5, 2019 •

edited

Loading

codelipenghui Jun 5, 2019 •

edited

Loading