You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "PavelZeger (via GitHub)" <gi...@apache.org> on 2026/05/16 09:50:45 UTC
[I] [PIP] Add `BatchReceive` to `Consumer` [pulsar-client-go]
PavelZeger opened a new issue, #1496:
URL: https://github.com/apache/pulsar-client-go/issues/1496
**Is your feature request related to a problem? Please describe.**
The Java client exposes a batched receive mode on every `Consumer`:
```java
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:
```go
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?**
- [x] I'm willing to submit a PR!
--
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.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [PIP] Add `BatchReceive` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1496:
URL: https://github.com/apache/pulsar-client-go/issues/1496#issuecomment-4582293031
> Thanks for the detailed analysis [@nodece](https://github.com/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?
Hi @nodece. Appreciate for the review of my latest comment/suggestion above.
--
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
Re: [I] [PIP] Add `BatchReceive` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1496:
URL: https://github.com/apache/pulsar-client-go/issues/1496#issuecomment-4509568715
Hi @nodece , @crossoverJie , @shibd , @freeznet , @RobertIndie ! Appreciate for a review of this PIP.
--
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
Re: [I] [PIP] Add `BatchReceive` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1496:
URL: https://github.com/apache/pulsar-client-go/issues/1496#issuecomment-4466510274
I would like to add this ability to the client
--
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
Re: [I] [PIP] Add `BatchReceive` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1496:
URL: https://github.com/apache/pulsar-client-go/issues/1496#issuecomment-4519572532
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?
--
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
Re: [I] [PIP] Add `BatchReceive` to `Consumer` [pulsar-client-go]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on issue #1496:
URL: https://github.com/apache/pulsar-client-go/issues/1496#issuecomment-4517108954
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.
--
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
Re: [I] [PIP] Add `BatchReceive` to `Consumer` [pulsar-client-go]
Posted by "PavelZeger (via GitHub)" <gi...@apache.org>.
PavelZeger commented on issue #1496:
URL: https://github.com/apache/pulsar-client-go/issues/1496#issuecomment-4846737916
Hi @nodece , @crossoverJie , @shibd , @freeznet , @RobertIndie ! Appreciate for a re-review of this PIP.
--
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