KAFKA-18736: Handle errors in the Streams group heartbeat request manager

[cadonna]

This commit adds error handling to the Streams heartbeat request manager.

Errors can occur while sending a heartbeat request and when a response with an error code that is not NONE is received.

Some errors are handled explicitly to recover from them or to log specific messages. All the others are handled as fatal errors.

Pull Request Overview

This PR addresses KAFKA-18736 by enhancing error handling in the Streams heartbeat request manager so that errors during heartbeat requests are handled explicitly and appropriately. Key changes include:

• Adding new error handling logic in the StreamsGroupHeartbeatRequestManager for various error conditions.
• Refactoring error and failure handling methods (onErrorResponse and onFailure) to differentiate retriable vs fatal error cases.
• Enhancing and expanding test cases to validate the new behavior across several error scenarios.

Reviewed Changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.

File	Description
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java	Introduces new error handling logic with dedicated error cases and logging for unsupported versions, coordinator issues, and authorization failures.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java	Adds comprehensive tests covering various error and failure conditions during heartbeat request processing.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java	Refactors and consolidates tests for heartbeat failure handling into a parameterized test.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java	Updates the heartbeat failure methods, splitting retriable and fatal error handling for clarity.

Comments suppressed due to low confidence (2)

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:558

• Consider adding additional context (e.g., request id or coordinator info) to the error log in the default case so that unexpected errors are easier to diagnose in production.

logger.error("StreamsGroupHeartbeatRequest failed due to unexpected error {}: {}", error, errorMessage);

clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java:1481

• [nitpick] Consider renaming the boolean parameter (e.g., to 'isRetriable') in the parameterized test to improve clarity on its intended meaning.

@ParameterizedTest
    @ValueSource(booleans = {true, false})