offset commits on Async tasks

[duke-bartholomew]

Hi, I have a question regarding the async (sink)operations in kafka-streams-dotnet.
Assume I have following topology:

    var builder = new StreamBuilder();
    builder.Stream<string, string>("my-topic")
        .MapValues(JsonConvert.DeserializeObject<MyObject>)
        .ForeachAsync(async (record, _) => {
                db.MyObjects.Add(record.Value);
                await db.SaveChangesAsync();
            },
            RetryPolicy.NewBuilder()
                .NumberOfRetry(10)
                .RetryBackOffMs(100)
                .RetriableException<DatabaseException>()
                .RetryBehavior(EndRetryBehavior.FAIL)
                .Build());

.. so writing to a database asynchronously.

What happens under the hood in kafka-streams-dotnet?
I read that an "external streamthread" will be used to handle that action.
Does this mean a new thread per message in this case? (I come from the JVM world, so my perception of threads and the impact of them might be skewed when it comes to .NET threads)

What about offset commits?
Assume 5 messages are in-flight of being inserted in the database (given above example topology). What would the committed offset be in case message 2,3,4 and 5 are properly inserted, but message 1 fails after it's retry period?

Also, assume I have successfully processed messages 1 to 5 and have message 6 to 20 in-flight in the async block.
What offset will be committed in this case?

And a related question:
threading-model.rst states the following:

For each async processor, a request and a response (except for ForEachAsync(..)) topics will be created and the stream thread publish into the request topic and subscribe the response topic.

So this would mean that using ForEachAsync will not create such request/response topics?
(but while doing some quick tests, these topics do pop up on my Kafka cluster)

[LGouellec]

Hi @duke-bartholomew ,

What happens under the hood in kafka-streams-dotnet?
Does this mean a new thread per message in this case? (I come from the JVM world, so my perception of threads and the impact of them might be skewed when it comes to .NET threads)

No a single thread is created and dedicated to consume the internal request, call the external processor with logic retry and so on, and publish the result in the internal response topic. This thread has a logic with pause/resume in case the processor is too long compared to max.poll.interval.ms. Btw, in JAVA the consumer has max.poll.interval.ms to consume a batch records between 0 and max.poll.records messages whereas in .Net the consumer has max.poll.interval.ms to consume one message. It's quite different.
All the logic is present here, here.

What about offset commits?
_Assume 5 messages are in-flight of being inserted in the database (given above example topology). What would the committed offset be in case message 2,3,4 and 5 are properly inserted, but message 1 fails after it's retry period?

Also, assume I have successfully processed messages 1 to 5 and have message 6 to 20 in-flight in the async block.
What offset will be committed in this case?_

The external DSL allow async/await syntax but some reasons like you mentioned, I can't execute the processing fully asynchronously.
Because if the first request throw a timed out and the message is not inserted whereas next message do then I can commit message and "loose" record 1.

The commit is called every commit.interval.ms like the other stream thread after the processing. So for each async/await call, I wait the task is done here.

A new feature could be execute all the batch of records fetched from the consumer asynchronously and waiting all the task before committing the offset and fetch the next batch of record. I keep in mind.

So this would mean that using ForEachAsync will not create such request/response topics?
(but while doing some quick tests, these topics do pop up on my Kafka cluster)

When you use ForEachAsync(...), there is only a request topic. Because you don't have a new stream returning the foreach processor. So you have a internal request topic consumed by the external thread, but no output endpoint.

Hope it's clear

Best regards,

[duke-bartholomew]

Hi @LGouellec

Aha, so it's basically a 2-stepped approach then?

[ source (TOPIC: my-input-stream) ] ---> [ process ] ---> [ ForEachAsync: (TOPIC:  process-000-request-sink)]

and

[ source (TOPIC: process-000-request-sink) ] ---> [ doAction ]
                                                     +=> with 'doAction' working it's way through 
                                                         the messages one-by-one (synchronous)

With offset commits happening automatically (on interval) on the my-input-stream partitions (after they got produced to the request-sink topic) and offset commits on the process-000-request-sink partitions when successfully processed a message.

[LGouellec]

Yes this exactly this king of behavior