[PIP] Add `BatchReceive` to `Consumer`

[PavelZeger]

Is your feature request related to a problem? Please describe.

The Java client exposes a batched receive mode on every Consumer:

Messages<T> msgs = consumer.batchReceive();
CompletableFuture<Messages<T>> f = consumer.batchReceiveAsync();

Behaviour is governed by BatchReceivePolicy, configured on the builder via consumerBuilder.batchReceivePolicy(...). The policy lets the user say "give me up to N messages, or up to S bytes, or wait at most T, whichever happens first."

Here the Consumer interface only offers single-message receive:

Receive(context.Context) (Message, error)

There is no equivalent of batchReceive. Users who want batched processing have to loop over Receive themselves, but that approach:

• Cannot honour a "max bytes" boundary without re-implementing accounting.
• Cannot signal "I want partial batches if the timeout fires" cleanly.
• Forces an extra goroutine if you want to bound wait time independently of the per-message context deadline.

Batched receive is a major feature for high-throughput sinks: connectors, ETL processors, anything that wants to amortize a downstream write across many messages. It is one of the most-requested features when teams compare the Go client against the Java client.

Describe the solution you'd like

No breaking change. Add a BatchReceivePolicy and two methods on the Consumer interface. To stay compatible with existing implementations, expose the new methods with default no-op implementations on the multi-topic consumer first and roll out per-impl support.

Describe alternatives you've considered

Edge cases for testing

• The first message arrives, then the channel becomes empty for longer than Timeout — return the partial batch.
• A MaxNumBytes of 1 with a payload of 100 — should return that one message, not get stuck.
• BatchReceive on a multi-topic consumer must round-robin or merge fairly, not starve some topics.
• Cancelling the parent ctx mid-batch should return what we already have, plus the context error.

Additional context

• Java: pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java (batchReceive, batchReceiveAsync)
• Java: pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatchReceivePolicy.java

Are you willing to submit a PR?

• I'm willing to submit a PR!

[PavelZeger]

Hi @nodece , @crossoverJie , @shibd , @freeznet , @RobertIndie ! Appreciate for a review of this PIP.

[nodece]

Thanks for the suggestion.

After analyzing the batchReceive implementation in the Java client, we found that it is essentially a client-side message accumulation wrapper — it drains messages from the internal incomingMessages queue into a container based on a BatchReceivePolicy (maxNumMessages / maxNumBytes / timeout), then returns the batch to the caller.

Key observations:

1. No broker-side optimization — batchReceive does not reduce the number of requests to the broker. Messages are still pushed one-by-one (or unpacked from broker-level batches individually) into the receiver queue. The "batch" is purely a client-side grouping.

2. Equivalent to a receive() loop — The same behavior can be achieved by calling receive() in a loop with your own accumulation logic and timeout control. The API is a convenience wrapper, not a performance optimization.

3. Go client context — In the Go client, users already have full control over how they consume messages via channels or callbacks. Adding a batchReceive API would just wrap Receive() / Chan() in a loop without providing additional value at the protocol level.

If there's a specific use case where you believe batch receive would provide tangible benefits (e.g., reducing lock contention, simplifying consumption patterns), we'd be happy to discuss further. Otherwise, we'd recommend implementing the accumulation logic in application code, which gives more flexibility over batching strategy.

[PavelZeger]

Thanks for the detailed analysis @nodece - I've gone through the four Consumer implementations (consumer, multiTopicConsumer, regexConsumer, zeroQueueConsumer) and I agree on the first three: a BatchReceive here would only be a for-loop around Chan() - no real benefit.

So I withdraw the proposal to add BatchReceive to the Consumer interface for these cases.

Where I think there's still a real benefit - ZeroQueueConsumer:
zeroQueueConsumer.Receive works differently:

1. It takes a lock,
2. calls availablePermits.inc() to grant exactly one permit,
3. waits for that one message,
4. repeats.

Consuming N messages with the current API costs N sequential permit grants and N round-trips to the broker. A native BatchReceive(N) could grant N permits upfront in a single Flow command and return when N messages have arrived (or the deadline fires). That's an actual optimization, not a wrapper - and it's specific to the Go client because of how zeroQueueConsumer is wired.

Proposed compromise:

1. Utility helper for the common cases. A free function in a small pulsar/util (or batch) subpackage:
   func BatchReceive(ctx context.Context, c pulsar.Consumer, policy BatchReceivePolicy) ([]pulsar.Message, error). It would wrap Chan() with correct timer semantics, a documented byte-size definition, partial-batch return on timeout/ctx cancel, etc. No changes to the Consumer interface. It's for users who want batch ergonomics opt in, everyone else is unaffected.

2. Native batch only for ZeroQueueConsumer. Add BatchReceive to a narrower interface (or to ZeroQueueConsumer directly) where it earns its place by saving real round-trips. This is the only spot where the API gives the user something they can't replicate from outside.

If this looks for you as a benefit, I'd be happy to create the PR to reflect it - utility helper for default/multi-topic/regex and/or native batch for zero-queue only. WDYT?