[PIP-38] Support batch receive in java client. #4621

codelipenghui · 2019-06-27T10:49:41Z

Motivation

Support messages batch receiving, some application scenarios can be made simpler. Users often increase application throughput through batch operations. For example, batch insert or update database.

At present, we provide the ability to receive a single message. If users want to take advantage of batch operating advantages, need to implement a message collector him self. So this proposal aims to provide a universal interface and mechanism for batch receiving messages.

For example:

Messages messages = consumer.batchReceive();
insertToDB(messages);
consumer.acknowledge(messages);

Verifying this change

Added new UT to verify this change.

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? (yes)
If yes, how is the feature documented? (docs and JavaDocs)

sijie

@codelipenghui this is a great change. left a few comments.

Additionally it might be good to have a few more followup changes to optimize this further.

The current pulsar client breaks a message batch to individual messages and collect multiple message into a Messages. There is a lot of unuseful object conversations.

Ideally the pulsar client implementation should

a) keep a queue of Messages. Each Messages is a message batch or multiple message batches.
b) on receiving individual message, it polls a Messages from the queue, and poll a message out of the Messages.

This can allow lazy deserialization and object creation, and it will increase the throughput using batch receive api because your cpu cycles can be reduced.

sijie · 2019-06-27T13:15:36Z

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

+    /**
+     * Max size of message for a single batch receive, 0 or negative means no limit.
+     */
+    private long maxSizeOfMessages;


maxSizeOfMessages sounds a bit confusing.

I would suggest maxNumMessages and maxNumBytes to replace maxNumberOfMessages and maxSizeOfMessages. thoughts?

sijie · 2019-06-27T13:22:02Z

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

+     * client.newConsumer().batchReceivePolicy(BatchReceivePolicy.builder()
+     *              .maxNumberOfMessages(100)
+     *              .maxSizeOfMessages(5 * 1024 * 1024)
+     *              .timeout(100)


I think it is better to have a builder method timeout(long timeout, TimeUnit timeoutUnit) rather than having two separated methods.

merlimat · 2019-06-27T15:21:50Z

@codelipenghui Since this is a significant addition to the client API, can you create a PIP with description and examples of the code?
Also, I'd mark this for 2.5 rather that 2.4.1 which should only contain bug-fixes (or very small improvements).

codelipenghui · 2019-06-28T03:50:00Z

@merlimat I have already move it to 2.5.0 and will create a PIP soon.

codelipenghui · 2019-06-29T15:58:00Z

@merlimat @sijie I have already create PIP-38 in wiki and google doc, please take a look.

Hope to get your advice on google doc, i will sync the update to the PIP wiki.

codelipenghui · 2019-07-01T02:48:20Z

@sijie I have addressed your comments, please review again.

codelipenghui · 2019-07-01T06:55:07Z

run java8 tests
run Integration Tests

codelipenghui · 2019-07-02T00:04:16Z

run java8 tests
run Integration Tests

codelipenghui · 2019-07-02T06:09:25Z

run java8 tests
run Integration Tests

codelipenghui · 2019-07-02T08:19:23Z

run java8 tests
run Integration Tests

codelipenghui · 2019-07-02T13:22:49Z

run java8 tests
run Integration Tests

codelipenghui · 2019-07-02T15:56:53Z

run java8 tests
run Integration Tests

codelipenghui · 2019-07-03T00:09:31Z

run java8 tests
run Integration Tests

sijie · 2019-07-03T01:48:06Z

I have already create PIP-38 in wiki and google doc, please take a look.

@codelipenghui can you send an email to the dev@ mailing list to start the discussion?

codelipenghui · 2019-07-03T08:43:37Z

@sijie Oh, sorry i forgot it, i will send a email soon.

codelipenghui · 2019-07-23T07:11:53Z

@merlimat
Please help take a look PIP-38 when you have time, here is the discuss thread: https://lists.apache.org/thread.html/3e2a87d31bf8a98142bd68545714cdbf5d87011b4ae3909c5c9f43b9@%3Cdev.pulsar.apache.org%3E
Thanks.

codelipenghui · 2019-08-01T06:43:45Z

run java8 tests
run Integration Tests

codelipenghui · 2019-08-10T01:42:02Z

run java8 tests
run integration tests

sijie

@codelipenghui overall looks good. left some comments there.

sijie · 2019-08-10T03:06:25Z

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

+     * Max number of bytes: 10MB
+     * Timeout: 100ms
+     */
+    public static final BatchReceivePolicy DEFAULT_POLICY = new BatchReceivePolicy(


in general I am not in favor of using "number of messages" in any configuration or policies. In a multi-tenant system, message size varies between tenants and applications. so I would actually remove the limit of number of message just rely on number of bytes for a default policy.

Hence my recommendation would be:

BatchReceivePolicy DEFAULT_POLICY = new BatchReceivePolicy(-1, 10 * 1024 * 1024, 100, TimeUnit.MILLISECONDS);

sijie · 2019-08-10T03:07:24Z

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

+    /**
+     * Max number of messages for a single batch receive, 0 or negative means no limit.
+     */
+    private int maxNumMessages;


since you already used builder pattern, just make these variables final

sijie · 2019-08-10T03:09:30Z

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

+    /**
+     * Get the list {@link Message}
+     */
+    List<Message<T>> getMessageList();


do we need to expose this? my feeling is to avoid exposing such method until it is really needed.

sijie · 2019-08-10T03:10:31Z

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

@@ -314,6 +315,14 @@ public ConsumerBuilderImpl(PulsarClientImpl client, Schema<T> schema) {
        return this;
    }

+    @Override
+    public ConsumerBuilder<T> batchReceivePolicy(BatchReceivePolicy batchReceivePolicy) {
+        checkArgument(batchReceivePolicy != null, "batchReceivePolicy must not be null.");


@codelipenghui we have already moved all the validation to ConsumerConfigurationData. Can you move the validation there?

I see all the validations are still in ConsumerBuilderImpl.

sijie · 2019-08-10T03:12:10Z

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

+    /**
+     * Max bytes of messages for a single batch receive, 0 or negative means no limit.
+     */
+    private long maxNumBytes;


I don't think we need long here. an int should be good enough. Because you cannot really hold a "long"-sized buffer in memory.

sijie · 2019-08-10T03:26:21Z

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

+        } catch (InterruptedException | ExecutionException e) {
+            State state = getState();
+            if (state != State.Closing && state != State.Closed) {
+                stats.incrementNumReceiveFailed();


do we need a new metric for batchReceive?

Yes, i will add metrics for batch receive.

sijie · 2019-08-10T03:27:20Z

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

+    protected CompletableFuture<Messages<T>> internalBatchReceiveAsync() {
+        CompletableFuture<Messages<T>> result = new CompletableFuture<>();
+        try {
+            lock.writeLock().lock();


should we use a writeLock or a readLock?

Use read lock may cause Messages return early(have not reached capacity yet), so use write lock here.

sijie · 2019-08-10T03:31:40Z

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

+            try {
+                msg = incomingMessages.poll(0L, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException e) {
+                // ignore


@codelipenghui Can we just use poll()?

sijie · 2019-08-10T03:32:15Z

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

+    void notifyPendingBatchReceivedCallBack(OpBatchReceive<T> opBatchReceive) {
+        MessagesImpl<T> messages = new MessagesImpl<>(batchReceivePolicy.getMaxNumMessages(),
+                batchReceivePolicy.getMaxNumBytes());
+        Message<T> msgPeeked = incomingMessages.peek();


I see the similar pattern in a few places. Can we make a function for it?

sijie · 2019-08-10T03:33:46Z

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

@@ -80,6 +91,11 @@ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurat
        this.pendingReceives = Queues.newConcurrentLinkedQueue();
        this.schema = schema;
        this.interceptors = interceptors;
+        this.batchReceivePolicy = conf.getBatchReceivePolicy();
+        this.pendingBatchReceives = Queues.newConcurrentLinkedQueue();


Can you delay the creation of this queue util the first batchReceive is called? I am try to reduce creating the queue if the consumer is not using batchReceive.

codelipenghui · 2019-11-13T06:39:11Z

@sijie rebased

codelipenghui · 2019-11-13T08:36:29Z

run java8 tests

codelipenghui · 2019-11-13T09:33:48Z

run java8 tests

ssrkr · 2020-11-10T12:32:20Z

Anyone working on adding the support for batchRecieve to Golang client?
People apparently are expecting it already. Please check https://stackoverflow.com/questions/62173028/reading-messages-in-bulk-through-a-pulsar-consumer

codelipenghui requested review from sijie, merlimat and jiazhai June 27, 2019 10:50

codelipenghui self-assigned this Jun 27, 2019

codelipenghui added this to the 2.4.1 milestone Jun 27, 2019

codelipenghui added area/client type/feature The PR added a new feature or issue requested a new feature labels Jun 27, 2019

sijie reviewed Jun 27, 2019

View reviewed changes

codelipenghui modified the milestones: 2.4.1, 2.5.0 Jun 28, 2019

sijie mentioned this pull request Jun 30, 2019

Why pulsar java client doesn't support batch message consuming? #4405

Closed

codelipenghui changed the title ~~Support batch receive in java client.~~ [PIP-38] Support batch receive in java client. Jul 1, 2019

codelipenghui force-pushed the batch_message_receiving branch from fee1af0 to d201d8a Compare August 10, 2019 01:44

sijie reviewed Aug 10, 2019

View reviewed changes

codelipenghui added 19 commits November 13, 2019 14:36

Support batch receive in java client.

eb9e125

Fix UT tests.

cfc06ec

Fix UT tests.

f8cb5e3

Fix comments.

840f5cf

Fix java client doc.

b34f360

Fix unit test issues.

b595310

fix zero queue size consumer unit tests.

cabe1cf

fix comments

f99953a

fix unit tests.

4fdcc96

fix unit tests.

d03e1bd

Add jcip license

e047d89

Fix NOTICE.bin.txt

7a3f927

fix comments

090f048

Fix license.

ea478c8

Fix license.

04c86c6

Fix check style

875f627

Fix imports

8e72d1c

Fix document.

8aff8fa

Fix java doc.

5f5692f

codelipenghui force-pushed the batch_message_receiving branch from 4e7a9ec to 5f5692f Compare November 13, 2019 06:38

Fix license header

87df913

jiazhai approved these changes Nov 19, 2019

View reviewed changes

jiazhai merged commit 56517d5 into apache:master Nov 19, 2019

BewareMyPower mentioned this pull request Jan 12, 2021

Fix NPE when MultiTopicsConsumerImpl receives null value messages #9113

Merged

1 task

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

shuoqingding mentioned this pull request Nov 29, 2023

[Bug] Java consumer blocked by message larger than batchReceivePolicy.maxNumBytes when calling batchReceiveAsync #21631

Open

2 tasks

[PIP-38] Support batch receive in java client. #4621

[PIP-38] Support batch receive in java client. #4621

Conversation

codelipenghui commented Jun 27, 2019 • edited Loading

Motivation

Verifying this change

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

Documentation

sijie left a comment • 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

merlimat commented Jun 27, 2019

codelipenghui commented Jun 28, 2019

codelipenghui commented Jun 29, 2019

codelipenghui commented Jul 1, 2019

codelipenghui commented Jul 1, 2019

codelipenghui commented Jul 2, 2019

codelipenghui commented Jul 2, 2019

codelipenghui commented Jul 2, 2019

codelipenghui commented Jul 2, 2019

codelipenghui commented Jul 2, 2019

codelipenghui commented Jul 3, 2019

sijie commented Jul 3, 2019

codelipenghui commented Jul 3, 2019

codelipenghui commented Jul 23, 2019

codelipenghui commented Aug 1, 2019

codelipenghui commented Aug 10, 2019

sijie left a 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 Aug 11, 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Nov 13, 2019

codelipenghui commented Nov 13, 2019

codelipenghui commented Nov 13, 2019

ssrkr commented Nov 10, 2020

codelipenghui commented Jun 27, 2019 •

edited

Loading

sijie left a comment •

edited

Loading

codelipenghui Aug 11, 2019 •

edited

Loading