Offset commits at key level to ensure no event loss (how offset committing works)

[pdeole]

Hello team,

I am mainly interested in key-level parallelism of parallel consumer however our application is data critical and hence would like to ask a simple question on below scenario and how it is handled to ensure no event loss in parallel consumer:

event1 and event2 are produced on partition1 in given order

1. since event1 with key1 and event2 with key2 are both part of the same partition1
2. key1 event has offset 30 while key2 has offset 40
3. key2 is processed by background thread and offset committed which is 40
4. before key1 gets processed by background thread, the consumer application instance/node goes down
5. partition1 gets rebalanced to node2 and start processing ahead of offset 40, thus losing key1

[astubbs]

Sorry for the delay in answering our question! It's a great question and really gets at the heart of what this library provides.

However, that's not how offset committing works :) Have you had a chance to look at the readme?

1. Let's insert event key1 with has offset 29, which was previously complete and committed.
2. Offset 40 will not be committed, as there are offsets previous to it, which have not completed (offset 30). What will happen is, offset 29 will be committed again, and in it's commit message there is a metadata payload - that payload will carry the information that offset 30 is not complete, while offset 40 is complete.
3. instance goes down, power cut
4. partition is reassigned to another consumer
5. consumer with new assignment, downloads the offset data from broker
6. consumer unpacks offset meta data payload, and decodes the information: offset is 30, payload says: 30 not complete, 40 complete
7. consumer polls partition, downloads offset 30 and 40
8. consumer knows 40 is previously completed, so skips processing message with offset 40
9. consumer processes message 30 concurrently as it has a different key
10. message 30 completes
11. consumer begins an offset committing process in background
12. consumer sees that 30 and 40 are now complete, commits offset 41 with empty metadata

Hopefully that clears things up for you? Let me know if you have any further questions! Let me know how you get on...

You can actually try this out- send some messages, thread.sleep on message 30, complete message 40, ask the PC to commit, then kill the process. Start up again, and observe that message 40 will be skipped, while message 30 will be skipped.

[nguyenbanh1]

If processes in worker pool is successfully but the controller is not committed to Kafka-broker and then instances/nodes is crashed and the result is between committing and processing message is not consistent.
My approach:
First: Still accept committing and processing is not consistent but processing must be idempotent
Second: both insert offset to offset-map and insert offset to a persistent(databases). If nodes have crash and then restart still keep consistent offset.

Could you explain more about how to consistent between commit offset and worker pool's processing? Thank a lot.

[astubbs]

What if your database crashes or is unavailable?
By adding another state store, you are adding extra resiliency (as long as the code continues even if it can’t write to the database) along with extra complexity, but it’s basically the same as committing it to Kafka.

Remember though that if you can’t commit to Kafka, you probably can’t read from it either. So is the is this really adding anything?

The only way to be strictly idempotent, is for each downstream system to record their own offset, not the upstream systems. Ie the adding of a database won’t help, if you’re function calls a web service. Even committing the offsets to Kafka doesn’t mean you’re strictly idempotent, in this situation.

Would you like a committing mode where offsets are committed after each user function finishes processing? Instead of once a second in the background? It would be a lot slower, but would minimise replays a lot more in the case of a failure.

(Again however the only way to be strictly idempotent is for each consuming system to be responsible for it - as Kafka doesn’t support XA transactions yet).

[pdeole]

Thanks Antony for detailed explanation.. Is there integration of parallel consumer API with spring-kafka? Or any example how it can be used if existing code is using spring-kafka API and we want to use Parallel consumer in that code?

…

On Mon, Jul 26, 2021 at 3:21 PM Antony Stubbs ***@***.***> wrote: Sorry for the delay in answering our question! It's a great questions and really gets at the heart of what this library provides. However, that's not how offset committing works :) Have you had a chance to look at the readme? 1. Let's insert event key1 with has offset 29, which was previously complete and committed. 2. Offset 40 will not be committed, as there are offsets previous to it, which have not completed (offset 30). What will happen is, offset 29 will be committed again, and in it's commit message there is a metadata payload - that payload will carry the information that offset 30 is not complete, while offset 40 is complete. 3. instance goes down, power cut 4. partition is reassigned to another consumer 5. consumer with new assignment, downloads the offset data from broker 6. consumer unpacks offset meta data payload, and decodes the information: offset is 29, payload says: 30 not complete, 40 compete 7. consumer polls partition, downloads offset 30 and 40 8. consumer knows 40 is previously completed, so skips processing message with offset 40 9. consumer processes message 30 concurrently as it has a different key 10. message 30 completes 11. consumer begins an offset committing process in background 12. consumer sees that 30 and 40 are now complete, commits offset 41 with empty metadata Hopefully that clears things up for you? Let me know if you have any further questions! Let me know how you get on... — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#135 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AU4J3ZYPGSFCVODBJJGT37TTZUVZNANCNFSM5AR4PIZQ> .

[astubbs]

@pdeole did you work it out? We have some captivity atm so given how popular spring is I may get an example created.

[pdeole]

@astubbs i didn't get a chance to work on service that utilize spring-kafka in our application, First we will be trying out the parallel consumer for services that use Kafka API without spring-kafka integration, so at this point I don't need the parallel consumer integration with Spring

[pdeole]

Hello @astubbs i think the parallel consumer is a very useful utility and has potential to become quite popular due to its advantages over traditional consumer approach. However given that many people using spring over kafka (spring-kafka integration), it would help if parallel consumer utility is also supported with spring-kafka. If you could get this prioritized in upcoming releases of parallel consumer, then that would greatly help.

[astubbs]

Issue linked here: #171

[pdeole]

@astubbs nice to have conversation with you again... we are able to use confluent parallel consumer for service using springboot and spring-kafka, just by making registering a listener for ApplicationStartedEvent and then instantiating a parallel consumer from there on.

Would like to ask you if there is any change in the policy of confluent wherein is this library now officially part of confluent lib suite and whether there will support available for issues faced with this library?