Add get many in streams as dependency injection

[marcosschroh]

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

Currently there is not a way to get many records at once when we are using the dependency injection approach.

With the current state we consume events once by one, this is wonderful for most applications, but there are use cases that we need to perform certain actions inside the consumption handler that can take too long, for example saving the events to a database or maybe doing a HTTP request, for example:

from kstreams import stream, ConsumerRecord

@stream(topic)
async def my_stream(cr: ConsumerRecord):
    # this takes 1 second, better to do it in bulk
    save_to_db(cr)

Describe the solution you'd like

Get a List[ConsumerRecord] every time that the consumption handler is called. The number of elements in the list can be specified, also a timeout.

from kstreams import stream, ConsumerRecord, GetMany

@stream(topic)
async def my_stream(crs: List[ConsumerRecord] = GetMany(records=10, timeout=10)):
    # this takes 1 second, better to do it in bulk
    for cr in crs:
        print(cr)
    
    save_bulk_to_db(crs)

Describe alternatives you've considered

Keep using the get_many from the Stream:

@stream_engine.stream(topic, ...)
async def stream(stream: Stream):
    while True:
        data = await stream.getmany(max_records=5)
        print(data)

[woile]

Nice idea!

I like the GetMany configuration approach.

A potential limitation is that you cannot modify the records and timeout. What if you wanted to to implement an exponential backoff. However, how often will the user have to re-configure the amount of records / timeouts?

An alternative could be to replace List with Many, where many could be a custom object that can be modified. Though for the initial implementation I would skip this.

[woile]

I've been discussing this issue with Marcos via call.
Because it's not trivial coding, I'm posting here so we can continue the discussion.
It's a bit more complex than we previously thought.

Possible issues

Adding a crs: list[ConsumerRecord] = GetMany, from now on: "looping solution", can lead to a lot of hard to prevent and identify problems:

The most simple one, how do you design a Dead Letter Queue?

If we include the list[ConsumerRecord] into our current solution, that would mean
that Middlewares should be supported. People may grab our DLQ examples, and try to
use them with a list[ConsumerRecord] creating a lot of potential issues.

Let's see the DLQ from our docs:

class DLQMiddleware(middleware.BaseMiddleware):
    def __init__(self, *, topic: str, **kwargs) -> None:
        super().__init__(**kwargs)
        self.topic = topic

    async def __call__(self, cr: ConsumerRecord):
        try:
            return await self.next_call(cr)
        except ValueError:
            await self.send(self.topic, key=cr.key, value=cr.value)

Here, the middleware protocol accepts only a single cr record.

What do we do?

1. Should we execute the middlewares for each of the list[ConsumerRecord] and then pass it to the user function?
2. Or should we change the protocol to a ConsumerRecord | list[ConsumerRecord]?

The 1st would mean increasing the execution to O(n^2), and on top that, if a record N, where N > 0 index, fails,
then you'd be performing a commit, skipping the previous records.

The 2nd option, would increase the complexity of the middlewares, and they should all be refactored. Most CI pipelines would fail with mypy, if they update kstreams, but keep using a middleware with just ConsumerRecord.
Furthermore, it would probably imply a O(n^2) complexity, as every record would be deserialized first in a middleware,
and it would then give the list to the user function.

In conclusion, Middlewares with the "looping solution" would create a lot of complexity.

Now, forget about the DLQ, and let's formalize the failure scenarios:

1. How to handle failures when a single record fails in a middleware or the user function?
2. How to handle failures when a single record fails in multiple middlewares and the user function?
3. How to handle failures when multiple records fail in a single middleware or user function?
4. How to handle failures when multiple records fail in multiple middlewares and the user function?

All this problems, remind us, why we move away from the original stream design:

@stream_engine.stream(topic)
async def consume(stream):
    async for cr in stream:
        print(cr.value)

It makes it hard to create reusable components, as most things should happen inside the for..in.
That's why we no longer recommend it and we want to deprecate it.
With the new approach, we were able to introduce Middlewares, as a common, reusable interface.

Alternative solutions

Right now, my suggested solution would be to introduce the option to fetch multiple messages,
but nothing changes for the user code:

@stream(..., max_records=5, timeout_ms=1000)
def user_fn(cr: ConsumerRecord):
    print(cr)

This change would allow users to potential improve perforamnce by fetching many messages at once,
thus, reducing latency.

Internally, kstream would use getmany instead of getone, but it would loop over the received messages,
executing for each the middlewares and the user_fn.

[Masqueey]

Should we execute the middlewares for each of the list[ConsumerRecord] and then pass it to the user function?

I don't entirely get how the Alternative solution is any different from the method suggested here. Is the only difference that it does the middlewares and the user function for every event instead of everything but the user function?

[woile]

The difference is that the amount of messages retrieved are hidden to the user, and it would be an implementation detail on kstreams:

Here's were we do the getone

kstreams/kstreams/streams.py

Lines 371 to 376 in 2d9dd78

	async def func_wrapper_with_typing(self) -> None:
	while self.running:
	try:
	cr = await self.getone()
	async with self.is_processing:
	await self.func(cr)

Example of possible implementation on kstreams:

async def func_wrapper_with_typing(self) -> None:
    while self.running:
        try:
            if SHOULD_GET_MANY:
                crs = await self.getone()
                async with self.is_processing:
                    for cr in crs:
                        await self.func(cr)
            else:
                cr = await self.getone()
                async with self.is_processing:
                    await self.func(cr)

The self.func(cr) contains all the middlewares + the user defined function

In this example SHOULD_GET_MANY would be somehow identified by kstreams based on the settings provided.
On top of that, you no longer have the O(n^2) issue, as the user function + all the middlewares are executed for each of the messages once.

Let's look at the DLQ:
If it fails, you have alraedy processed the previous ones, so no issue here.

caveat
I don't know what a .commit() would do in this scenario, does the consumer think it should commit after all the getmany? @marcosschroh ?

[marcosschroh]

I don't know what a .commit() would do in this scenario, does the consumer think it should commit after all the getmany? @marcosschroh ?

I think the commit should be done for the specific offset of the TopicPartition that the ConsumerRecord belongs to. Otherwise it will commit all the offset for records that have been not processed.

@stream(..., max_records=5, timeout_ms=1000)
def user_fn(cr: ConsumerRecord, stream: Stream):
    print(cr)
    tp = TopicPartition(
        topic=transactional_topic,
        partition=cr.partition,
    )
    await stream.commit({tp: cr.offset + 1})

We can conclude that:

• The implementation of GetMany might improve the performance, as we will use the germany from the driver.
• GetMany will fetch multiple records but they will go one by one through the middlewares
• The end user is responsible to accumulate multiple records. This will require extra effort but it is the best that we can do.

class Accumulator(BaseModel):
    records: List = []

    def add(self, cr: ConsumerRecord):
        ...

    def ready(self):
        ...

    async def do_something(self):
        await save_bulk(self.records)
        await do_request(self.records)

accumulator = Accumulator()

@stream(..., max_records=5, timeout_ms=1000)
def user_fn(cr: ConsumerRecord):
    accumulator.add(cr)
    
    if accumulator.ready():
        accumulator.do_something()
   

It is possible to fetch multiple records for specifics TopicPartitions, so we should add the partition option too. The default is all partitions.

What do you think?

[woile]

I think that's a great approach.

On top of that, if a user configures getmany, we could throw an exception if you try to do stream.commit() without any parameters.