KAFKA-17397: Ensure ClassicKafkaConsumer sends leave request on close even if interrupted

[chickenchickenlove]

Description

This PR addresses KAFKA-17397 by ensuring deterministic behavior in
ClassicKafkaConsumer.close() under interruption for the classic group
protocol.

In CI, PlaintextConsumerTest.testCloseLeavesGroupOnInterrupt() can
fail for ClassicKafkaConsumer because LeaveGroupRequest is sometimes
blocked by NetworkClient.isReady() when a metadata update is due.
Since isReady prioritizes metadata requests, the pending
LeaveGroupRequest can remain unsent. When the calling thread is
already interrupted, ConsumerNetworkClient may throw an
InterruptException before the pending request gets a chance to be
sent. This causes the member to leave only via session.timeout.ms,
resulting in test flakiness.

Changes

Allow LeaveGroupRequest to bypass the metadata-update gating in
NetworkClient.isReady() (while still respecting canSendRequest),
ensuring the request is sent during close() even when a metadata
update is due.

Sequence Diagram

• Steps 9, 10, and 11 describe non-deterministic behavior that prevents
  the ClassicKafkaConsumer from successfully sending a LEAVE_GROUP
  request when it is interrupted.
• This PR makes the sending of the LEAVE_GROUP request deterministic
  by bypassing the metadata update step specifically for LEAVE_GROUP.
  (See, Step 9, 15, 16, 17, 18)

Fixes

• https://issues.apache.org/jira/browse/KAFKA-17397
• https://issues.apache.org/jira/browse/KAFKA-18031
  • Flaky Test -
    https://develocity.apache.org/scans/tests?search.rootProjectNames=kafka&search.timeZoneId=Asia%2FTaipei&tests.container=kafka.api.PlaintextConsumerTest&tests.sortField=FLAKY&tests.test=testCloseLeavesGroupOnInterrupt(String)%5B1%5D

In local re-produce

• With current trunk branch, the test failed 1~2 times out of 20 runs.
• With this PR, all 500 test runs succeeded (although there were a few
  build failures due to busy CPU and full memory.)

$ grep -r "PASSED" | wc -l
498

$ grep -r "FAILED" | wc -l
2

$ grep -r "FAILED"
run-34.log:> Task :clients:compileJava FAILED
run-34.log:BUILD FAILED in 5s

Note on the implementation

I considered adding isReadyForLeaveGroup directly to the KafkaClient
interface to keep the design consistent. However, I opted for the
current approach to avoid modifying the interface, as I wasn't sure if
changing the KafkaClient interface would require a KIP.

If you believe adding it to the interface is preferable (and acceptable
without or with a KIP ), please let me know. I'm happy to refactor it
and wrote KIP.

[chickenchickenlove]

@lucasbru @lianetm
I’ve created a PR that fixes the root cause we identified in the issue you guys reported.
When you have bandwidth, could you please take a look? 🙇‍♂️

[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.