@m1a2st Thanks for your comments.
I have a question.
This is an extremely rare case, so writing test code won’t be easy.
Nevertheless, I’ll give it a try.
Even if I use Mockito or similar tools, the tests may end up quite convoluted.
Could you please take that into consideration?

Thanks for the reply. I’ll try to reproduce the issue on my local machine by following the steps in spring-projects/spring-kafka#3703.

There has been the issue that events skipped in group rebalancing

If this is correct, it's a serious bug, and we should not just fix it as MINOR, but file a Jira ticket. \cc @dajac

@m1a2st thanks for your answer!
There is no example to reliably reproduce that situation.
I was able to reproduce it once again on my local by using two IDE and debug mode, but it was very difficult to reproduce it.
Because these call stacks are handled by two other thread.

If you can build kafka consumer client and your local and use it on your local dependencies,
I think that adding an Thread.sleep(...) to SyncGroupResponseHandler#handle(...) side is help for you to reproduce it.

@mjsax
I think especially CoperativeStikcy strategy will be affected.
Because there’s a high probability that the consumer still owns the previous partition, calling Fetcher#fetchRecords(...) will continue to use the records from that partition even if the commit fails.

I’ve documented the detailed scenario here.
Please refer to this comment.

Hey, @mjsax , @m1a2st .
I dug into this problem more deeply.

The issue became clearer, so I reverted the previous commit and added a new commit that introduces wider synchronized block.

Because synchronized in generationIfStable() is insufficient to prevent unexpected race conditions.
So, This situation can cause event skip problem.

// In ConsumerCoordinator.sendOffsetCommitRequest(...)
// Main Thread do
1.  generation = generationIfStable();   // <- MemberState.COMPLETING_REBALANCE, 
2. groupInstanceId = rebalanceConfig.groupInstanceId.orElse(null); // <- MemberState.COMPLETING_REBALANCE
3. if (generation == null) {   // <- MemberState.COMPLETING_REBALANCE

// In SyncGroupResponseHandler.handle(...)
// Consumer Coordinator Heartbeat Thread
4. synchronized (AbstractCoordinator.this) { ..  } // <- MemberState.COMPLETING_REBALANCE
5. log.info("Successfully synced group in generation {}", generation); // MemberState.COMPLETING_REBALANCE
...

// In ConsumerCoordinator.sendOffsetCommitRequest(...)
// Main Thread do
6. if (rebalanceInProgress()) { // <- MemberState.COMPLETING_REBALANCE

// In SyncGroupResponseHandler.handle(...)
// Consumer Coordinator Heartbeat Thread
7. state = MemberState.STABLE; // <- MemberState.STABLE

// In ConsumerCoordinator.sendOffsetCommitRequest(...)
// Main Thread do
8. else { return RequestFuture.failure(new CommitFailedException("Offset ...)) } // <- MemberState.STABLE

This race condition can prevent certain records from being committed, even when the previous partition is reassigned to the consumer during rebalancing.

This cause the some problem, I guess.
Please let me know if I’m misunderstanding something.

In terms of At-Most-Once (enable.auto.commit=true)

1. Consumer A read offset 10 for partition 0, however failed to commit offset 10 due to race condition.
2. Then Consumer A down. Consumer B read offset 10 for partition 0.
   It means that it cannot ensure At-Most-Once semantic.

I believe this race condition could cause other side effects beyond what I can foresee.
What do you think?

Hi, @m1a2st sorry to bother you.
I reverted previous commit and made an new one.
Because I felt the previous commit wasn’t effective.

Currently, I just added only synchronized keywords.
IMHO, I believe the existing test cases already cover the current changes.
However, if additional tests are required, I’d appreciate your guidance on how best to approach writing them.

Thanks in advance 🙇‍♂️

@chickenchickenlove Thanks for the patch. Could you please file a Jira for the bug?

For the context, we already had a few of those race conditions in the past and they led us to completely re-architect the consumer internals. The new internals are however only used when the new rebalance protocol is used.

IMHO, I believe the existing test cases already cover the current changes.
However, if additional tests are required, I’d appreciate your guidance on how best to approach writing them.

I agree that adding a specific test case will be hard for this one. I think that we could go without one.

Currently, I just added only synchronized keywords.
IMHO, I believe the existing test cases already cover the current changes.
However, if additional tests are required, I’d appreciate your guidance on how best to approach writing them.

Thanks to @chickenchickenlove for the explanation. I completely agree that this scenario is difficult to test, thus add a test is not necessary.

@dajac
Thanks for your comments. 🙇‍♂️
I made a Jira ticket for this issue. (https://issues.apache.org/jira/browse/KAFKA-19242)
The re-architected rebalncing protocol is KIP-848?

@dajac , @m1a2st
I created a new Jira ticket and made a fresh commit to match it. After doing a git reset --hard, I reapplied the same changes under KAFKA-19242 commit and updated the PR title accordingly.
When you have time, Please take an another look. 🙇‍♂️

@dajac , @m1a2st gently ping. 🙇‍♂️

Thank you @chickenchickenlove for solving this issue.

@m1a2st @dajac Is it acceptable to merge this PR? 🙇
Unfortunately, this happens weekly, forcing teams to hunt down ignored records.

@chickenchickenlove Thanks for the patch and sorry for the delay. I have not had the time to really dive into it yet. Would it be possible to extend the description to better explain what the issue is? I understand that there is a race condition in this area. I would like to really understand the impact. Do we skip events because we somehow commit offsets of unprocessed records due to the race condition?

Do we skip events because we somehow commit offsets of unprocessed records due to the race condition?

Hi @dajac, this is precisely the side effect of this race condition.

Do we skip events because we somehow commit offsets of unprocessed records due to the race condition?

Hi @dajac, this is precisely the side effect of this race condition.

Understood. I don't fully grasp the sequence of events leading to it. The description of the pull request suggests that the commit fails before the offsets are sent to the group coordinator. Intuitively, I would think that it should re-process records instead of skipping records when the partition is re-assigned. I would like to better understand it.

I believe this comment from @chickenchickenlove describes in detail a bad sequence of events between the main thread and the consumer thread.

spring-projects/spring-kafka#3703 (comment)

@dajac
Sorry to make you confused.
Let me define problem clearly.

TL;DR:

• The race condition I mentioned earlier can trigger an unexpected CommitFailedException. However, this issue does not seem to lead to event skipping.
• The wrapper application respects the semantics of exceptions thrown by Apache Kafka. However, if an unexpected exception are thrown from kafka consumer(as in this case, kafka consumer throws CommitFailedException even if it can commit properly), it might cause a boundary condition. In such case, some records might be lost.

I was able to successfully reproduce the issue in two separate steps.
First, an unexpected CommitFailedException was thrown due to a race condition.
Then, I verified that when the wrapper application encounters this CommitFailedException, it ends up losing the record.

The boundary condition between Apache Kafka and Wrapper application seem to lead to event skipping.
Apache Kafka handles CommitFailedException and RetryableCommitException separately.
Therefore, the wrapper application also handles those errors separately.

However, since the CommitFailedException was caused by a race condition and was not anticipated, it appears to have been handled through a different path than intended, resulting in some events being skipped in wrapper application. (Especially, Cooperative Sticky strategy affects mostly.)

IMHO, anyway, it would be better to fix this race condition problem.
Because other wrapper applications rely on the semantics of
CommitFailedException and RetryableCommitException or RebalancingSomethingException and handle it properly.

Does this PR and my description make sense to you?
Please let me know your opinion. 🙇‍♂️

@dajac
Sorry to bother you...!
Gently ping 🙇‍♂️

@chickenchickenlove Thanks for the explanation. I agree that we should fix the race condition. Could you please update the description of the PR and the Jira to better explain the issue based on it? The fundamental issue is that the commit path may not return the correct exception due to the race condition during a rebalance as you explained.

@dajac
Thanks for your time. 🙇‍♂️
I has been updated PR description and Jira (https://issues.apache.org/jira/browse/KAFKA-19242).
Could you take a look?

Merged to trunk, 4.0 and 3.9.

Thanks @chickenchickenlove @dajac @injae-kim

Hi, could you let me know if you plan to release the 3.9 fix to Maven soon? It would be very helpful for our project.

The ticket mentions 3.9.2 as one of the versions containing the fix.
However, as far as I can see, a 3.9.2 was never released, right?