KAFKA-18135: ShareConsumer HB UnsupportedVersion msg mixed with Consumer HB #18101

peterxcli · 2024-12-08T18:01:33Z

UnsupportedVersion error is handled in the parent AbstractHeartbeatRequestManager, so used by consumer and share consumer. If a share consumer gets the errors, it will end up with a msg that is currently specific to consumer

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

Lines 384 to 386 in 6fd951a

    
           case UNSUPPORTED_VERSION: 
        
               message = "The cluster doesn't yet support the new consumer group protocol." + 
        
                       " Set group.protocol=classic to revert to the classic protocol until the cluster is upgraded.";

Handle the UnsupportedVersion separately in the existing
handleSpecificError (note that the unsupported version for consumer may also end up containing a msg specific to SubscriptionPattern not supported in HB v0, if regex is used without the required v1)

add onSpecificFailure for consumerHB and shareHB
add testUnsupportedVersionGeneratedOnTheBroker to cover both handleSpecificError and handleSpecificFailure/exception

AndrewJSchofield · 2024-12-08T20:11:56Z

...c/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java

@@ -100,6 +100,9 @@ public abstract class AbstractHeartbeatRequestManager<R extends AbstractResponse
        "group protocol. Set group.protocol=classic on the consumer configs to revert to the CLASSIC protocol " +
        "until the cluster is upgraded.";

+    public static final String SHARE_PROTOCOL_NOT_SUPPORTED_MSG = "The cluster does not support the new SHARE " +


I would change it to "The cluster does not support the share group protocol. To use share groups, the cluster must have the share group protocol enabled." I think this message will remain valid over the next few releases without us needing to update it.

Also, I would expect this constant should go in the ShareMembershipMgr right?

AndrewJSchofield · 2024-12-08T20:13:03Z

We should get @lianetm to make sure she's happy too.

lianetm

Thanks for the changes! Took a first look, here are some comments

lianetm · 2024-12-08T22:37:43Z

...c/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java

@@ -100,6 +100,9 @@ public abstract class AbstractHeartbeatRequestManager<R extends AbstractResponse
        "group protocol. Set group.protocol=classic on the consumer configs to revert to the CLASSIC protocol " +
        "until the cluster is upgraded.";

+    public static final String SHARE_PROTOCOL_NOT_SUPPORTED_MSG = "The cluster does not support the new SHARE " +


Also, I would expect this constant should go in the ShareMembershipMgr right?

lianetm · 2024-12-08T22:41:13Z

...c/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java

+                if (!handleSpecificError(response, currentTimeMs)) {
+                    logger.error("{} failed due to {}: {}", heartbeatRequestName(), error, errorMessage);
+                    handleFatalFailure(error.exception(errorMessage));
+                }


I wonder if it would be simpler to remove the UnsupportedVersion handling from here completely? In the end we're duplicating the logic we already have in the default case further down right? We would only need the logic in the Share/Consumer managers, simply because even if the error is not specific the handling is. Thoughts?

You're right, I think the logic in default and UnsupportedVersion handling are the same. Thanks!

lianetm · 2024-12-08T23:06:29Z

...c/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java

+                if (errorMessage.contains(REGEX_RESOLUTION_NOT_SUPPORTED_MSG)) {
+                    // If the error is about regex subscription, use the original error message
+                    message = errorMessage;
+                }


I would expect we don't need this here. This regex error is generated on the client when building the request (it generates a response with exception) so the handling lands on the onFailure path (not on the onResponse path that uses this handleSpeficifError).

I would expect we just need to have here the same handling we had for the UnsupportedVersion in the parent class before this PR:

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

Lines 411 to 412 in ee42644

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

handleFatalFailure(error.exception(CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG));

That being said, we do need to ensure that we handle responses for each consumer, to cover this bit that is now on the abstract mgr (so it could wrongly apply a consumer msg to the share consumer):

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

Lines 322 to 325 in ee42644

if (isHBApiUnsupportedErrorMsg(exception)) {

// This is expected to be the case where building the request fails because the node does not support

// the API. Propagate custom message.

handleFatalFailure(new UnsupportedVersionException(CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG, exception));

Both consumers need to ensure they propagate their specific msgs when that exception is in a response.

So I guess we should add a handle onSpeficFailure in each consumer just like handleSpecificError did?

Yes. The implementation of that onSpecificFailure/handleSpecificExceptionInResponse on each consumer should cover this bit I expect, with the correct msg:

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

Lines 322 to 325 in 497f500

if (isHBApiUnsupportedErrorMsg(exception)) {

// This is expected to be the case where building the request fails because the node does not support

// the API. Propagate custom message.

handleFatalFailure(new UnsupportedVersionException(CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG, exception));

And in the AbstractHBMgr onFailure probably just need something like:

if (exception instanceof RetriableException) { ... } else if (!onSpecificFailure()) { handleFatalFailure(exception); }

We don't need any specifics for the regex case I would expect. It should land on the handleFatalFailure and propagate the msg it has (it will only ever be generated when building consumer HB, so no risk for the share consumer)

With that seems we cover the specifics of handling responses with exceptions (generated on the client) , and in a very similar way to how we handle responses with error codes (generated on the broker).

AndrewJSchofield

A few more changes requested. I have been working with the code on my local environment, and I think there's still something not totally right. This PR isn't making it worse in any way, but I think there's another improvement we can make here which would improve it more.

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

AndrewJSchofield · 2024-12-12T18:26:41Z

@peterxcli It seems to me that this doesn't quite work as well as it might. In my testing, connecting a share consumer to a broker which does not support SHARE_GROUP_HEARTBEAT does not emit the new message. What actually happens is that org.apache.kafka.common.errors.UnsupportedVersionException: The node does not support SHARE_GROUP_HEARTBEAT is thrown.

The consumer checks the ApiVersions response and notices that SHARE_GROUP_HEARTBEAT is not supported before it's even sent a request. As a result, in AbstractHeartbeatRequestManager.makeHeartbeatRequest, in the whenComplete call, the response is null. This means we end up in onFailure rather than onResponse, and this means the new code is never actually executed.

What do you think @lianetm? Am I missing something?

AndrewJSchofield · 2024-12-12T18:30:22Z

@peterxcli Actually, I've realised that this is actually a temporary problem. In AK 4.0, SHARE_GROUP_HEARTBEAT is marked as an unstable API. If the broker doesn't support the RPC at all, then it will behave as I said. In AK 4.1, I expect we'll mark the API as stable, and then the consumer will send the request. Then the new message will be used if share groups are not enabled. So, I think this is as good as it can be and the approach in the PR is OK.

lianetm · 2024-12-13T19:24:19Z

Just to answer @AndrewJSchofield's comment #18101 (comment) , you got it right, we basically have 2 paths to get the Unsupported:

generated on the client when building the request if the broker does not support the API or version required => HB managers get a client response object with exception
generated on the broker (API supported but not enabled) => HB managers get client response with ErrorCode

For the KafkaConsumer we can have the 2 paths now but Share will always go down 1 as you described until the API becomes stable.

lianetm · 2024-12-13T18:57:04Z

...c/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java

+                if (errorMessage.contains(REGEX_RESOLUTION_NOT_SUPPORTED_MSG)) {
+                    // If the error is about regex subscription, use the original error message
+                    message = errorMessage;
+                }


Yes. The implementation of that onSpecificFailure/handleSpecificExceptionInResponse on each consumer should cover this bit I expect, with the correct msg:

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

Lines 322 to 325 in 497f500

if (isHBApiUnsupportedErrorMsg(exception)) {

// This is expected to be the case where building the request fails because the node does not support

// the API. Propagate custom message.

handleFatalFailure(new UnsupportedVersionException(CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG, exception));

And in the AbstractHBMgr onFailure probably just need something like:

if (exception instanceof RetriableException) { ... } else if (!onSpecificFailure()) { handleFatalFailure(exception); }

We don't need any specifics for the regex case I would expect. It should land on the handleFatalFailure and propagate the msg it has (it will only ever be generated when building consumer HB, so no risk for the share consumer)

With that seems we cover the specifics of handling responses with exceptions (generated on the client) , and in a very similar way to how we handle responses with error codes (generated on the broker).

lianetm · 2024-12-13T19:03:29Z

.../test/java/org/apache/kafka/clients/consumer/internals/ShareHeartbeatRequestManagerTest.java

@@ -422,6 +425,29 @@ public void testHeartbeatResponseOnErrorHandling(final Errors error, final boole
        }
    }

+    @ParameterizedTest
+    @ValueSource(strings = {SHARE_PROTOCOL_NOT_SUPPORTED_MSG})
+    public void testUnsupportedVersion(String errorMsg) {


this is truly testUnsupportedVersionGeneratedOnTheClient (client response with exception). Renaming will maybe clarify, and we should ensure we have coverage for something like testUnsupportedVersionGeneratedOnTheBroker (client response with ErrorCode), is it already covered?

The outcome of both tests should be the same, but they execute diff code paths modified in this PR (handleSpecificError and handleSpecificFailure/exception)

peterxcli · 2024-12-20T15:07:59Z

Rebase CI hotfix #18284.

peterxcli · 2024-12-22T06:26:58Z

Three unrelated tests failed.

UserQuotaTest	testProducerConsumerOverrideLowerQuota(String, String).quorum=kraft.groupProtocol=classic	org.opentest4j.AssertionFailedError: Client with id=QuotasTestProducer-1 should have been throttled, 0.0 ==> expected: but was:	6.92s
PlaintextAdminIntegrationTest	testConsumerGroups(String, String).quorum=kraft.groupProtocol=consumer	org.opentest4j.AssertionFailedError: Expected the offset for partition 0 to eventually become 1.	21.57s
UserClientIdQuotaTest	testThrottledProducerConsumer(String, String).quorum=kraft.groupProtocol=classic	org.opentest4j.AssertionFailedError: Client with id=QuotasTestProducer-!@#$%^&*() should have been throttled, 0.0 ==> expected: but was:	3.31s

AndrewJSchofield

A couple of final comments. This is a nice improvement.

AndrewJSchofield · 2024-12-27T11:25:57Z

...c/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java

@@ -461,15 +434,24 @@ protected void handleFatalFailure(Throwable error) {
        membershipManager().transitionToFatal();
    }

+    /**
+     * Error handling specific failure to a group type when sending the request.


I would add "and no response has been received" to the comment. This is the crucial difference between the usage of handleSpecificFailure and handleSpecificExceptionInResponse.

AndrewJSchofield · 2024-12-27T11:26:34Z

...c/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java

+            String message = CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG;
+            if (errorMessage.equals(REGEX_RESOLUTION_NOT_SUPPORTED_MSG)) {
+                message = REGEX_RESOLUTION_NOT_SUPPORTED_MSG;
+                logger.error("{} failed due to regex resolution not support: {}", heartbeatRequestName(), message);


nit: "regex resolution not supported".

… and consumer

…ove error handling - Move SHARE_PROTOCOL_NOT_SUPPORTED_MSG constant to AbstractHeartbeatRequestManager - Add test case for unsupported version error handling in ShareHeartbeatRequestManager - Improve error message handling for regex resolution errors

…artbeatRequestManager#onErrorResponse

- Split error handling into two distinct methods: handleSpecificFailure and handleSpecificExceptionInResponse - Move UnsupportedVersion handling from onFailure to specific implementations - Rename handleSpecificError to handleSpecificExceptionInResponse for clarity - Improve error logging messages to better distinguish between client-side and broker-side version errors This change makes the error handling more modular and clearer by separating client-side request failures from broker response errors.

- Split UnsupportedVersion tests into client-side and broker-side cases - Add separate test methods for each scenario in both Consumer and Share heartbeat managers - Fix test helper methods to properly simulate client vs broker version errors - Rename test methods to better describe test scenarios This change improves test clarity by explicitly testing both client-side and

AndrewJSchofield

lgtm

AndrewJSchofield · 2024-12-28T15:38:40Z

Failing test is known as flaky and tracked by https://issues.apache.org/jira/browse/KAFKA-18298.

github-actions bot added consumer clients small Small PRs labels Dec 8, 2024

AndrewJSchofield added the ci-approved label Dec 8, 2024

AndrewJSchofield requested changes Dec 8, 2024

View reviewed changes

AndrewJSchofield requested a review from lianetm December 8, 2024 20:12

lianetm reviewed Dec 8, 2024

View reviewed changes

peterxcli force-pushed the k18135 branch from bf1357a to 2e55073 Compare December 11, 2024 08:23

AndrewJSchofield requested changes Dec 12, 2024

View reviewed changes

.../src/main/java/org/apache/kafka/clients/consumer/internals/ShareHeartbeatRequestManager.java Outdated Show resolved Hide resolved

.../src/main/java/org/apache/kafka/clients/consumer/internals/ShareHeartbeatRequestManager.java Outdated Show resolved Hide resolved

lianetm reviewed Dec 13, 2024

View reviewed changes

AndrewJSchofield added the KIP-932 Queues for Kafka label Dec 16, 2024

peterxcli force-pushed the k18135 branch from d595eee to 2e55073 Compare December 18, 2024 18:06

github-actions bot added the small Small PRs label Dec 18, 2024

peterxcli force-pushed the k18135 branch from 2e55073 to 2ab4981 Compare December 18, 2024 18:11

github-actions bot removed the small Small PRs label Dec 20, 2024

peterxcli force-pushed the k18135 branch from 12622ad to 27acf94 Compare December 20, 2024 15:05

peterxcli requested review from AndrewJSchofield and lianetm December 20, 2024 15:08

AndrewJSchofield requested changes Dec 27, 2024

View reviewed changes

peterxcli requested a review from AndrewJSchofield December 28, 2024 02:10

peterxcli added 8 commits December 28, 2024 10:11

Add specific error handling for unsupported version in share consumer…

8f3b783

… and consumer

Address Andrew and Lianet's comment on SHARE_PROTOCOL_NOT_SUPPORTED_MSG

7b4274b

Remove duplicated logic of unsupported version handling in AbstractHe…

f8205cc

…artbeatRequestManager#onErrorResponse

Address Andrew comments on shareheartbeatrequestmanageer

9fa5fba

Address Andrew's comments

b976d2c

peterxcli force-pushed the k18135 branch from edb60f6 to b976d2c Compare December 28, 2024 02:11

AndrewJSchofield approved these changes Dec 28, 2024

View reviewed changes

AndrewJSchofield merged commit be4d1a6 into apache:trunk Dec 28, 2024
8 of 9 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18135: ShareConsumer HB UnsupportedVersion msg mixed with Consumer HB #18101

KAFKA-18135: ShareConsumer HB UnsupportedVersion msg mixed with Consumer HB #18101

peterxcli commented Dec 8, 2024 •

edited

Loading

AndrewJSchofield Dec 8, 2024

lianetm Dec 8, 2024

AndrewJSchofield commented Dec 8, 2024

lianetm left a comment

lianetm Dec 8, 2024

lianetm Dec 8, 2024

peterxcli Dec 9, 2024

lianetm Dec 8, 2024

peterxcli Dec 12, 2024

lianetm Dec 13, 2024

AndrewJSchofield left a comment

AndrewJSchofield commented Dec 12, 2024

AndrewJSchofield commented Dec 12, 2024

lianetm commented Dec 13, 2024

lianetm Dec 13, 2024

lianetm Dec 13, 2024 •

edited

Loading

peterxcli commented Dec 20, 2024 •

edited

Loading

peterxcli commented Dec 22, 2024

AndrewJSchofield left a comment

AndrewJSchofield Dec 27, 2024

AndrewJSchofield Dec 27, 2024

AndrewJSchofield left a comment

AndrewJSchofield commented Dec 28, 2024

	case UNSUPPORTED_VERSION:
	message = "The cluster doesn't yet support the new consumer group protocol." +
	" Set group.protocol=classic to revert to the classic protocol until the cluster is upgraded.";

	logger.error("{} failed due to {}: {}", heartbeatRequestName(), error, errorMessage);
	handleFatalFailure(error.exception(CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG));

	if (isHBApiUnsupportedErrorMsg(exception)) {
	// This is expected to be the case where building the request fails because the node does not support
	// the API. Propagate custom message.
	handleFatalFailure(new UnsupportedVersionException(CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG, exception));

KAFKA-18135: ShareConsumer HB UnsupportedVersion msg mixed with Consumer HB #18101

KAFKA-18135: ShareConsumer HB UnsupportedVersion msg mixed with Consumer HB #18101

Conversation

peterxcli commented Dec 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AndrewJSchofield commented Dec 8, 2024

lianetm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AndrewJSchofield left a comment

Choose a reason for hiding this comment

AndrewJSchofield commented Dec 12, 2024

AndrewJSchofield commented Dec 12, 2024

lianetm commented Dec 13, 2024

Choose a reason for hiding this comment

lianetm Dec 13, 2024 • edited Loading

Choose a reason for hiding this comment

peterxcli commented Dec 20, 2024 • edited Loading

peterxcli commented Dec 22, 2024

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AndrewJSchofield left a comment

Choose a reason for hiding this comment

AndrewJSchofield commented Dec 28, 2024

peterxcli commented Dec 8, 2024 •

edited

Loading

lianetm Dec 13, 2024 •

edited

Loading

peterxcli commented Dec 20, 2024 •

edited

Loading