KAFKA-16460: New consumer times out consuming records in multiple consumer_test.py system tests

[FrankYang0529]

When a consumer receives group heartbeat with revoked partitions. If the consumer enable auto commit, it will send a OffsetCommitRequest before removing revoked partitions from subscription state. There are two potential race conditions in AsyncKafkaConsumer:

• If there is data in FetchBuffer from previous FetchRequest, users can still get data from poll function. However, the auto commit function doesn't work for these data, because the consumer mark revoked partitions with pendingRevocation=true after we receives OffsetCommitResponse. At that time, there is no further OffsetCommitRequest will be sent by auto commit feature.
• The FetchRequestManager still can send new FetchRequest before the consumer receives OffsetCommitResponse. The reason is same as above.

To fix this, we have to mark pendingRevocation=true before the consumer sends last OffsetCommitRequest, so FetchCollector can't get revoked partition data from FetchBuffer and FetchRequestManager doesn't send new FetchReqeust with revoked partitions.

I run consumer_test.py.test_broker_failure 10 times without error.

> TC_PATHS="tests/kafkatest/tests/client/consumer_test.py::OffsetValidationTest.test_broker_failure" /bin/bash tests/docker/run_tests.sh
================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.12.0
session_id:       2024-11-15--010
run time:         6 minutes 57.245 seconds
tests run:        12
passed:           12
flaky:            0
failed:           0
ignored:          0
================================================================================

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[lianetm]

Nice catch @FrankYang0529 ! This makes sense to me. It was actually the initial intention
https://github.com/lianetm/kafka/blob/e227a9cac887a201eed9105ed47337fb35acb0ae/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java#L872-L886

I guess we lost it with a refactoring and didn't have test coverage for that corner case.

[lianetm]

do we still need the other call now that we have this at the beginning of the reconciliation before commits?

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

Line 1117 in cc20e78

markPendingRevocationToPauseFetching(revokedPartitions);

[FrankYang0529]

I think it's better to keep markPendingRevocationToPauseFetching in AbstractMembershipMember#revokePartitions, because ConsumerMembershipManager#invokeOnPartitionsRevokedOrLostToReleaseAssignment also uses the function.

Although it's redundant for consumer heartbeat response path, it's still required for ConsumerMembershipManager#signalMemberLeavingGroup path.

[lianetm]

makes sense.

[lianetm]

makes sense.

[lianetm]

could we extend this test to ensure that the markPending happens before the commit is triggered? (Maybe using InOrder like I'm doing here)

kafka/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java

Lines 434 to 436 in 64d020a

	InOrder inOrder = inOrder(backgroundEventHandler, membershipManager);
	inOrder.verify(backgroundEventHandler).add(any(ErrorEvent.class));
	inOrder.verify(membershipManager).onHeartbeatFailure(false);

[FrankYang0529]

Updated it. Thanks for the suggestion.

[lianetm]

Thanks for the good catch and improved test coverage @FrankYang0529! LGTM.