MINOR: Fix race conditions in KafkaStreams when topic is missing.

[chickenchickenlove]

Background

There is race condition in ConsumerNetworkThread for Kafka Streams with new_protocol.
So it make HandlingSourceTopicDeletionIntegrationTest#shouldThrowErrorAfterSourceTopicDeleted be flaky.
Please refer to details in attached sequence diagram.

CC. @mjsax

Result

• Fixes flaky test HandlingSourceTopicDeletionIntegrationTest#shouldThrowErrorAfterSourceTopicDeleted.

Related

• https://issues.apache.org/jira/browse/KAFKA-19511
• https://issues.apache.org/jira/browse/KAFKA-19521
• KAFKA-19511: Fix flaky test HandlingSourceTopicDeletionIntegrationTest.shouldThrowErrorAfterSourceTopicDeleted #20183

[chickenchickenlove]

@mjsax sorry to mention.
Could you take a look this PR?
It is related with Kafka streams!

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[chickenchickenlove]

I’d greatly appreciate it if someone could kindly review this PR when they have a chance.

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[chickenchickenlove]

Gently Ping, could you take a look this PR?

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[mjsax]

We are somewhat swamped recently. Sorry for the delay... Don't know when I will get to it. But please keeping nagging us about it :)

[chickenchickenlove]

@mjsax Thanks for your comment even if you are in busy!
If you know persons who can review this PR, could you mention him/her ?
Thanks for your help and effort always! 🙇‍♂️

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[lianetm]

Hey @chickenchickenlove , thanks for looking in to this, interesting. Help me understand this issue better (the proposed change is harmless as I see it and could fix the flakiness, but want to make sure it is not hiding something else).

High level, from the consumer POV, the 2 sides involved happen in the background (so safe there):

• updating assignment when a reconciliation completes (topic delete)
• process unsubscribe

So how exactly do we end up in step 8??? Is it that the partitions passed to the unsubscribe here (taken from the currentAssignment) are not in the subscriptionState anymore?

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java

Line 921 in 05999c7

return revokeActiveTasks(toTaskIdSet(currentAssignment.activeTasks));

This bit it different between streams and the consumer mgrs actually (the partitions to revoked passed in the consumer are taken directly from the subscriptionState, not from the currentAssignment like the streamsMgr does)

[chickenchickenlove]

@lianetm , Thanks for your feedback!
I’ll dig into it and let you know If I find something suspicious. 🙇‍♂️

I have a question.
The changes and scenario that I wrote is reasonable to you?

[lianetm]

The changes and scenario that I wrote is reasonable to you?

yes, I have no concern with the fix really, I just don't fully understand the issue (want us to make sure we're not covering an issue with the assignment passed to the task revocation)

Thanks for looking into it!

[chickenchickenlove]

I see!
Let me check!

[chickenchickenlove]

@lianetm

It looks like a commit masked the issue I was trying to fix.

1. I tested and debugged to solve this problem based on commit hash (986322d)
2. Currently, this PR based on commit hash (8861629)

In this branch, I reverted this PR(38e3359) to check HandlingSourceTopicDeletionIntegrationTest can be success with new consumer protocol.
But, it never success.

However, If I tested it based on 986322d, HandlingSourceTopicDeletionIntegrationTest is always success.

In other words, one of the PRs between those two commits made the HandlingSourceTopicDeletionIntegrationTest#shouldThrowErrorAfterSourceTopicDeleted test permanently fail.

That is, it succeeds under the legacy protocol but fails under the new protocol.
To put it differently, in the legacy protocol, when a topic is deleted, all Stream Threads are expected to transition to the Error state, whereas in the new protocol, when a topic is deleted, all Stream Threads remain in the Running state.

Is this the expected behavior for the new protocol in Kafka Streams?

[chickenchickenlove]

@lianetm
I think this commit affects.
handleMissingSourceTopicsWithTimeout(...)
03190e4

In my local, timer set the value `maxPollTimeMs` (300000ms). If I set the timer as `1000`ms, same problem occur and this patch will work.

So I believe this change is masking the issue.
If you’d like me to temporarily shorten the timer interval to investigate further, please let me know.
Currently, the value in the trunk branch is set to twice the heartbeat interval, which is already quite short, so I think it’s meaningful to dig into it.

So, Please let me know your opinion!
Thanks a lot.

FYI,
When I rebase to current trunk branch, and I tested it again and it succeed!!!

[chickenchickenlove]

There was happening 😅

In summary,

1. The issue could not be reproduced on the current PR’s base branch due to the previous commit 03190e4
2. However, after rebasing the base branch onto the latest trunk branch, the issue reappeared. When I reverted commit 38e3359 and tested again with the new protocol, all tests passed successfully.

So, Let me dig into more.

[chickenchickenlove]

@lianetm
I dug into more.

I think this problem has been solved by this PR. (#19400)
Because, StreamThread don't send a UNSUBSCRIBE events anymore.

So, Step 5 disappear, ConsumerNetworkThread will not handle UNSUBSCRIBE event.
Therefore, there is no race condition any more.

What do you think?

[github-actions]

This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

[github-actions]

This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.