Does parallel-consumer support manual commit?

[otherhy]

Hello,
I hope to improve the performance of my application by upgrading parallel-consumer, but manual commit is required in the business logic. I did not find the corresponding method in the documentation. Can you please confirm if this feature is supported?

Thank you.

[rkolesnev]

@otherhy - no, parallel consumer does not support manual commit right now.
There is an option to request a commit - through requestCommitAsap() method to trigger it sooner.
Given that Parallel Consumer is processing multiple messages in parallel - i am not sure how manual commit could be used with it.
Is manual commit actually, really required? I struggle to imagine a scenario where manual commit is warranted / has advantage over auto commit.

[shreeramkulkarni]

Hey @rkolesnev, I'm responding to a closed discussion but would like to add my thoughts in the same context. (Not sure if I want to reopen the thread)

We have use case where our message processing is not really idempotent and system failures cause duplicate processing, affecting the net effect of ordered processing.

1. A,B,C to be processed in order.
2. A,B processed but no offsets committed.
3. system failure.
4. A and B are again processed.
5. Net ordered processing A->B->A->B->C

With periodic commits there's still a gap for duplicate consumption. Committing offsets individually would bring the behavior closer to exactly-once semantics.

I wonder if any consideration was given to providing users with the option to commit offsets after processing messages individually, especially for those willing to incur the potential I/O overhead?

Would appreciate if you can share any thoughts.
Thanks

[rkolesnev]

Hi @shreeramkulkarni - the original answer still applies.

If you are using Parallel Consumer to increase throughput of processing - and Ordering mode by KEY - then lets say you have keys A and B and 3 messages for each key A1,A2,A3 and B1,B2,B3 that are being processed in order but in parallel.
Lets say commit after each message is processed is triggered.
A1 processed and commit triggered
in parallel B1 is processed as well
A1 commit successful
crash happens
recovery - commit data read
B1 is processed again.
What i am trying to highlight is introducing additional parallelisation increases potential duplication on crash / restarts.
And normally level of parallelism is quite high - so if processing 100 or 1000 messages from single partition in parallel - doing a commit after each message is processed is not really feasible (as in parallel mode - more than 1 message will be processed concurrently anyway).

Stepping back from Parallel Consumer - even with vanilla Kafka clients - performing commit after each processed message does not guarantee no duplicates would be introduced as application may crash just after processing a message but before commit is performed.

But for most distributed system implementation where there are network calls etc - there are those potential failure windows / modes where duplicate processing may be introduced.

Usually the processing step (or external system if processing involves calling out to external system) would be coded to handle possible duplicates in various ways.
For example is processing results in sending message downstream to kafka topic - deduplication could be applied to that topic using KStreams / KSQL or Flink etc.

[shreeramkulkarni]

Thanks for the response @rkolesnev !

introducing additional parallelisation

Not sure if you are referring to any parallelization other than whats already there in the library.
And yes I agree, because of Kafka's offset sequential offset management, I see that if there are multiple messages from the same partition being processed concurrently (KEY, UNORDERED) then committing them individually is not really possible.

But when messages are processed in order, PARTTION wise, it should be possible to commit individually without losing concurrency (with some additional network io overhead)? and could have lesser chances of duplicate consumption?

[rkolesnev]

I am referring to additional parallelism as compared to vanilla Kafka Consumer - exactly as you said - in KEY and UNORDERED modes.
For PARTITION mode - it would be just easier to use vanilla Kafka Consumer instead - with the option for individual commit available there.
Implementing this in ParallelConsumer would require the commit behaviour in ParallelConsumer to differ based on processing mode - in addition - the Kafka Consumer object is not thread safe - so even in PARTITION processing order - when concurrently processing messages from multiple partitions - doing commits on individual partition states would be hard and need synchronisation.

In this case - having dedicated Kafka Consumer per partition would be more straightforward to implement.

[shreeramkulkarni]

Thanks @rkolesnev , got it the rational and limitations behind sequential commits!

So, the project that I work on has a use case where we want to make sure that the net ordering is not impacted in any system failure scenarios.
Let me explain what I mean here,
For key A, these are the message that need to be consumed in order A1->A2->A3->A4->A5. (KEY ordering)

--start-->A1--processed-->A2--processed--offset committed-->A3--processed-->A4--processed---application crashed before period commit---X
A1 to A4 are processed but offset is committed for A1->A2.
On restart, A3 and A4 are processed again and eventually A5.

ref doc,

Across a system failure, only completed messages not stored as such in the last offset payload commit will be replayed

Library has a nice mechanism to keep track of all the messages and also has PERIODIC_CONSUMER_SYNC that prevents from undesired ordered processing as long as it's clean restart but not system failure.
So, if we look at the final order in which the messages were processed is: A1->A2->A3->A4->A3->A4->A5

Since we can't trigger a Kafka commit after processing every message, if each message were to be immediately marked and committed to some metadata (outside the application process, may be Kafka) then there would be only single message reprocessed and not the chain.
Our usecase is fine if the same message can be reprocessed but not beyond the chain.

Would like to know if there would be any possibility to safeguard against this scenario within the library?

TL;DR
Allowed repetitions
A1->A2->A3->A4->A5 (ideal case)
A1->A2->A3->A4->A4->A5
A1->A2->A3->A4->A4->A4->A5
...
Undesired repetitions
A1->A2->A3->A4->A3->A4->A5 or
A1->A2->A3->A4->A5->A3->A4->A5