Parallel async threads

[okomen-foreflight]

Hey there,

I have a topic with 20 partitions, which I have created a stream from where i make a ForeachAsync. But I've found that the messages in the for each is processed sequentially. So when if an async calls takes 5 seconds, then it will take 5 seconds before the next message is processed. This is despite ParallelProcessing set to true and MaxDegreeOfParallelism set to default.

So my question is the following: Is it possible to make the External stream thread parallel while still respecting sequential processing per kafka key. If so what am I missing?

Best regards

[LGouellec]

Hey @okomen-foreflight,

Today ForeachAsync processed sequentially the messages, because let's suppose the processing is really asynchronous, I can commit a newer offset before having the return of an oldest offset. And in case of the oldest offset fail, the message is already committed.
But I understand what you mean, maybe I can add another parameter to allow really asynchronous processing, but you assume that you can loose messages.

Btw, ParallelProcessing and MaxDegreeOfParallelism is only used for the classic DSL.

[okomen-foreflight]

Ahhh right of course, that's fair.
My application doesn't actually need that option now that I think about it. I will make a safer mechanism to process parallel if needed. Thanks for the input!

Also what is the classic DSL in this case and what would it do?

[LGouellec]

Each processor knows the next processor, like a composite pattern :

The forward method looks like :

foreach(var next in NextProcessors)
  next.Process(key, value);

If you enable ParallelProcessing, the forward method looks like :

Parallel.ForEach(NextProcessors,
                        new ParallelOptions { MaxDegreeOfParallelism =configuration.MaxDegreeOfParallelism },
                        Process);

TL;DR : This ParallelProcessing allow you to forward in parallel to a different sub-stream processors the record, otherwise this forward method is syncrhonous. First the next processor, then the second one.

[okomen-foreflight]

Hey @LGouellec
I've come to think about it, and I've started to wonder why it isn't possible to do parallel processing when we have multiple stream threads and presumably multiple consumers.

Kafka offsets are commited per partition not per topic, so if I have 20 partitions and 20 stream threads, it should be possible to allow ForeachAsync to run in parallel with a degree of 20 🤔

[LGouellec]

Hi @okomen-foreflight ,

Please find bellow the threading model :

In other words, if you have 20 partitions with 20 stream threads, you have 1 partition per thread = 1 consumer per partition, the topology isn't shared across all your threads.
The topology is duplicate for each task (= partition), so if you have 1 task per thread, the topology is executed with a parallelism of 20 I mean.

ForeachAsync, MapAsync ... must be used only if you want to interact with another tiered system (Http API, Database, etc ...) otherwise you can use ForEach, Map, ... and the unit of parallelism is the number of threads ..

[okomen-foreflight]

Yeah I realize that, but it seems it should be possible to also make the external stream thread parallel to the degree of partitions right?

[okomen-foreflight]

I could make my own consumer and producers, one for each streamThread/partition and then make them call an external system and produce to a new record.
Doing so would allow me to get parallelism even with async calls.
So I'm unsure why your streamiz couldn't do the same thing?

[LGouellec]

Today I have only one external stream thread (thread in green) to execute the external call per instance.

Yes you can scale the external thread to the number of partitions, but today it's not possible with the current status. Please feel free to create a GH issue to track this new feature.

[okomen-foreflight]

Alright cool! I will definitely do that
Thank you @LGouellec for all your hard work on this library. It's amazing