A way to consume multiple messages as a single collection (e.g. List<T>)

[Songzhizong]

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

I am using RabbitMQ Stream for high-throughput scenarios, where the bottleneck often shifts from network I/O to the processing logic, specifically when interacting with downstream systems like databases (ClickHouse, Elasticsearch) or calling external APIs.

The current ConsumerBuilder only supports a single-message callback via messageHandler(BiConsumer<Context, Message> messageHandler).
This forces the application to process messages one by one. In scenarios where I need to perform bulk inserts or batch processing to optimize IOPS and transaction overhead, the current API is limiting.

Describe the solution you'd like

I would like to request native support for batch consumption in the Java Client.
Ideally, the ConsumerBuilder could offer a mechanism to receive a batch of messages.

For example, an API similar to this:

Environment environment = Environment.builder().build();

environment.consumerBuilder()
    .stream("my-stream")
    .batchSize(500) // Trigger when 500 messages are accumulated
    .batchTimeout(Duration.ofMillis(100)) // Or trigger when time window elapses
    .batchMessageHandler((context, messages) -> { // messages is List<Message>
        // Here I can perform bulk insert efficiently
        myDatabase.insertAll(messages); 
    })
    .build();

Describe alternatives you've considered

Currently, we have to implement a manual "Accumulator/Buffer" pattern on the client side:

1. Receive message in the single-message callback.
2. Push it into a thread-safe List/Queue.
3. Check if the list size reaches a threshold or a timer expires.
4. Flush the list to the business logic.

Why this alternative is not ideal:

• Complexity: Every user has to reinvent the wheel (concurrency control, timer management, offset tracking).
• Risk: It increases the risk of message loss or duplication if the buffering logic is not handled correctly during graceful shutdowns.
• Performance: A native implementation within the client library (closer to the network layer/Netty buffer) would likely be more efficient than application-level buffering.

Additional context

Many other streaming clients (like Kafka's poll returning a generic collection, or Spring's batch listeners) support this out of the box. Since RabbitMQ Stream is designed for high throughput, adding batch consumption support would significantly enhance its capability in "Big Data" or "Data Pipeline" use cases.

[acogoluegnes]

I converted the original issue into a conversation. A new issue can be created if we come up with more precise actionable items.

The current API provides a good isolation from the underlying protocol, as messages arrive in a "chunk", within 1 delivery frame. A chunk can contain 1 to several thousands messages, depending on the ingress.

We could choose to pack all the messages of a chunk in a list, so the batch size would vary.

To enforce a setting like a batch size, the list of messages would need to span several chunks in some cases.

• Performance: A native implementation within the client library (closer to the network layer/Netty buffer) would likely be more efficient than application-level buffering.

I don't think making it "closer to the network" would make a big difference, except making the code base more complex.

Obviously we would trade latency for throughput with this approach.

A first prototype could be implemented as a regular MessageHandler that adapts the one-message approach to the batch-oriented approach. It could contain all the features you mentioned: enforcing a batch size or chunk-oriented batch, timeout release (which implies having a background thread), failure and recovery management.

No users requested this feature in over 4 years, but we would consider contributions if they are aligned with the current library design.

[michaelklishin]

I agree that the gain here will likely be API ergonomics first and foremost, and not so much the consumption throughput. Or at least that's my intuition that Stream PerfTest can easily prove right or wrong.

[Songzhizong]

I would like to provide a concrete use case to highlight why the batch pull functionality is critical for us.

Use Case: Metering Service
We are implementing a metering service for a medium-sized system. Specific API calls generate metering events that are published to the stream. A downstream service consumes these events to generate hourly aggregated usage data per user.

Throughput & Database IOPS Analysis
To achieve high consumption throughput, processing data in batches is essential to minimize database round-trips. Below is a comparison of the impact when processing 1,000 events belonging to 10 different users:

1. With Batch Pull (Target Scenario):
   The service retrieves 1,000 events in a single batch. We can categorize them in memory and interact with the database efficiently: 1 IN query (to fetch states for the 10 users) and 10 update operations.

   • Total DB Operations: ~11

2. Without Batch Pull (Current Limitation):
   Without native batch pull support (and assuming we avoid implementing complex and error-prone buffering logic at the application level), we are forced to process events one by one. Each event triggers a "select + update" cycle.

   • Total DB Operations: 2,000 (1,000 queries + 1,000 updates)

Conclusion:
Without native batch support, the database IOPS load increases drastically (by a factor of ~180 in this example), which severely limits the overall system throughput.

[michaelklishin]

@Songzhizong you keep saying that "batch pulling" will provide better throughput to the users of this client without any evidence. The main contributor to this client has pretty openly challenged this assumption in the general case.

You are not forced to process events one by one. You just need to do the buffering in your own code. Being "closer to the network" will not result in any improvements.

So please stop repeating that argument without any backing. We would believe a PR and benchmarks-backed evidence (that we will try to reproduce) but not repeated claims that "batch pull" or "being closer to the network" would make a meaningful difference. You can parallelize processing today, the problem of batch processing is as old as concurrent programming.

I can see how this can be an API ergonomics improvement and for that reason, we would consider a certain change.

[michaelklishin]

Without native batch support, the database IOPS load increases drastically (by a factor of ~180 in > this example), which severely limits the overall system throughput.

Addressing that is perfectly possible in your own code, today, without any changes to this library.
Consume N messages into a collection of fixed size, process the batch however you like, then reset the buffer and consume N more messages. Rinse and repeat.

It's certainly not this client's problem to solve that your application does not currently do that, or that it causes the ages old N+1 query scenario elsewhere. Not a single other user complaining about this in four years is solid evidence.

Any case, you @Songzhizong need this feature badly, so it's on you to contribute the changes, benchmarking results and convince our team that we should accept the changes and maintain them forever (since most contributors will never again help maintain their code after originally submitting it).

[Songzhizong]

Apologies for the delayed response.

I completely agree with your technical assessment. You are right that from a pure I/O perspective, buffering in the application layer versus the client layer doesn't fundamentally change the load on the database. Our current solution is indeed implementing a buffer in our business logic to address this.

However, my primary motivation for this proposal isn't just about raw performance, but rather API ergonomics and standardization.

Scenarios like writing to databases (e.g., ClickHouse, Elasticsearch) almost always require batch processing. While users can implement their own accumulators, building a robust one—handling batch sizes, flush intervals, offset management, and especially data safety during graceful shutdowns—is non-trivial and error-prone.

If the client provided a standard implementation (or a helper utility), it would significantly improve the developer experience and lower the barrier for adoption.

Regarding the contribution: I'm not entirely confident in my ability to modify the core logic of such a high-quality project, but I will try to look into it and see if I can propose a PoC or a benchmark as you suggested.

[acogoluegnes]

@Songzhizong Thanks for explaining the benefits of batching to the RabbitMQ team. My comments above stand:

• start with a re-usable MessageHandler and share the implementation with the open-source community
• explore the code and see how such a feature could fit in the existing design and API of this library (we consider contributions)

As @michaelklishin wrote, the RabbitMQ team can look conservative about significant contributions, but 1-time requirements and 1-time contributors come and go and we have to maintain RabbitMQ code on the long term.

[Songzhizong]

Thank you for the guidance, @acogoluegnes .

I fully understand and respect the team's conservative approach regarding new features. You are right that long-term maintainability is paramount, especially when dealing with contributions from the community.

I will follow your advice and start by implementing a reusable BatchMessageHandler (likely as a utility/helper class) that encapsulates the buffering logic. I'll take some time to explore the source code to ensure the design fits naturally with the existing API, and then I'll share the implementation or submit a PR for your review.