Re: [PR] [feat][client] Implement batched read in Reader [pulsar]

["Denovo1998 (via GitHub)" <gi...@apache.org>]

Denovo1998 commented on code in PR #26700:
URL: https://github.com/apache/pulsar/pull/26700#discussion_r4230669307


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsReaderImpl.java:
##########
@@ -182,6 +183,13 @@ public Message<T> readNext() throws PulsarClientException {
         return msg;
     }
 
+    @Override
+    public Messages<T> batchReadNext() throws PulsarClientException {
+        Messages<T> msg = multiTopicsConsumer.batchReceive();

Review Comment:
   I noticed that the single-topic and multi-topic readers currently use different batch receive policies.
   
   `ReaderImpl` explicitly sets `maxNumMessages=1` with no timeout, while `MultiTopicsReaderImpl` uses the default Consumer policy (10 MiB / 100 ms).
   
   So this method can return an empty batch after the timeout, even though the new Reader API says it blocks until a message is available.
   
   Could we make the batch policy consistent between the two implementations and clarify the empty-batch behavior in the API contract? We already have `BatchReceivePolicy`, so we probably don't need another mechanism for this.



##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java:
##########
@@ -75,6 +85,23 @@ public interface Reader<T> extends Closeable {
      */
     CompletableFuture<Message<T>> readNextAsync();
 
+    /**
+     * Read asynchronously the next messages in the topic.
+     *
+     * <p>{@code batchReadNextAsync()} should be called subsequently once returned {@code CompletableFuture} gets complete
+     * with received messages. Else it creates <i> backlog of receive requests </i> in the application.
+     *
+     * <p>The returned future can be cancelled before completion by calling {@code .cancel(false)}
+     * ({@link CompletableFuture#cancel(boolean)}) to remove it from the the backlog of receive requests. Another
+     * choice for ensuring a proper clean up of the returned future is to use the CompletableFuture.orTimeout method
+     * which is available on JDK9+. That would remove it from the backlog of receive requests if receiving exceeds
+     * the timeout.
+     *
+     * @return a future that will yield a messages (when it's available) or {@link PulsarClientException} if the reader
+     *         is already closed.
+     */
+    CompletableFuture<Messages<T>> batchReadNextAsync();

Review Comment:
   One more thing: adding these two abstract methods breaks source compatibility for existing `Reader` implementations.
   
   There's already an example in this repo: `ReaderHandlerTest.IllegalReader` in the `pulsar-websocket` module implements `org.apache.pulsar.client.api.Reader`, but it doesn't implement either new method, so compiling that test source should fail.
   
   Since `Reader` is a public Stable interface, this can also affect external implementations.
   
   Could we address the existing implementations and check whether default methods would be more appropriate for compatibility?



##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java:
##########
@@ -207,6 +208,21 @@ public Message<T> readNext() throws PulsarClientException {
         return msg;
     }
 
+    @Override
+    public Messages<T> batchReadNext() throws PulsarClientException {
+        Messages<T> msg = consumer.batchReceive();

Review Comment:
   I think we're still missing an important part here.
   
   `ReaderImpl` configures the underlying consumer with `DISABLED_BATCH_RECEIVE_POLICY`, which has `maxNumMessages=1` and `timeout=0`.
   
   Since `batchReceive()` uses that policy, and `MessagesImpl.canAdd()` enforces the message count limit, this method can only return one message per call.
   
   That means it doesn't really provide the batching benefit described in the PR.
   
   Could we use a proper batch receive policy here and add a regression test that verifies a single `batchReadNext()` call can return multiple messages?



##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java:
##########
@@ -207,6 +208,21 @@ public Message<T> readNext() throws PulsarClientException {
         return msg;
     }
 
+    @Override
+    public Messages<T> batchReadNext() throws PulsarClientException {
+        Messages<T> msg = consumer.batchReceive();
+
+        // Acknowledge message immediately because the reader is based on non-durable subscription. When it reconnects,
+        // it will specify the subscription position anyway
+        consumer.acknowledgeAsync(msg).exceptionally(ex -> {

Review Comment:
   Should we keep cumulative acknowledgment here?
   
   The existing `readNext()` uses `acknowledgeCumulativeAsync()`, but `acknowledgeAsync(Messages)` uses individual acknowledgments.
   
   There's also a difference in `MultiTopicsReaderImpl`: synchronous batch reads use cumulative acks for each message, while asynchronous batch reads use individual acks.
   
   For Reader, acknowledging only the last message in the batch (or the last message per topic/partition for multi-topic reads) seems more consistent with the existing behavior and would avoid unnecessary individual acknowledgment state.
   
   Was there a reason to change the ack semantics here?



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