You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lianetm (via GitHub)" <gi...@apache.org> on 2024/11/29 20:05:11 UTC
[PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
lianetm opened a new pull request, #17989:
URL: https://github.com/apache/kafka/pull/17989
Client side validation to fail HB request with SubscriptionPattern if the HB version does not support it (v0)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1866509662
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -602,6 +602,44 @@ public void testHeartbeatResponseOnErrorHandling(final Errors error, final boole
}
}
+ @Test
+ public void testUnsupportedVersion() {
+ mockErrorResponse(Errors.UNSUPPORTED_VERSION, Errors.UNSUPPORTED_VERSION.message());
+ ArgumentCaptor<ErrorEvent> errorEventArgumentCaptor = ArgumentCaptor.forClass(ErrorEvent.class);
+ verify(backgroundEventHandler).add(errorEventArgumentCaptor.capture());
+ ErrorEvent errorEvent = errorEventArgumentCaptor.getValue();
+
+ // UnsupportedApiVersion in HB response without any custom message. It's considered as new protocol not supported.
+ String hbNotSupportedMsg = "The cluster does not support the new consumer group protocol. Set group" +
+ ".protocol=classic on the consumer configs to revert to the classic protocol until the cluster is upgraded.";
+ assertInstanceOf(Errors.UNSUPPORTED_VERSION.exception().getClass(), errorEvent.error());
+ assertEquals(hbNotSupportedMsg, errorEvent.error().getMessage());
+ clearInvocations(backgroundEventHandler);
+
+ // UnsupportedApiVersion in HB response with custom message. Specific to required version not present, should
+ // keep the custom message.
+ String hbVersionNotSupportedMsg = "The cluster does not support resolution of SubscriptionPattern on version 0. " +
Review Comment:
I am sorry but I don’t understand what you mean. On the server, we basically handle the field if it is provided.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1869683864
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -313,12 +319,29 @@ private void onFailure(final Throwable exception, final long responseTimeMs) {
logger.debug(message);
} else {
logger.error("{} failed due to fatal error: {}", heartbeatRequestName(), exception.getMessage());
- handleFatalFailure(exception);
+ 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));
+ } else {
+ // This is the case where building the request fails even though the node supports the API (ex.
+ // required version 1 not available when regex in use).
+ handleFatalFailure(exception);
+ }
}
// Notify the group manager about the failure after all errors have been handled and propagated.
membershipManager().onHeartbeatFailure(exception instanceof RetriableException);
}
+ /***
+ * @return True if the exception is the UnsupportedVersion generated on the client, before sending the request,
+ * when checking if the API is available on the broker.
+ */
+ private boolean isHBApiUnsupportedErrorMsg(Throwable exception) {
+ return exception instanceof UnsupportedVersionException &&
+ exception.getMessage().equals("The node does not support " + ApiKeys.CONSUMER_GROUP_HEARTBEAT);
Review Comment:
nice suggestion, makes sense to me, filed https://issues.apache.org/jira/browse/KAFKA-18157, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1866530959
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -602,6 +602,44 @@ public void testHeartbeatResponseOnErrorHandling(final Errors error, final boole
}
}
+ @Test
+ public void testUnsupportedVersion() {
+ mockErrorResponse(Errors.UNSUPPORTED_VERSION, Errors.UNSUPPORTED_VERSION.message());
+ ArgumentCaptor<ErrorEvent> errorEventArgumentCaptor = ArgumentCaptor.forClass(ErrorEvent.class);
+ verify(backgroundEventHandler).add(errorEventArgumentCaptor.capture());
+ ErrorEvent errorEvent = errorEventArgumentCaptor.getValue();
+
+ // UnsupportedApiVersion in HB response without any custom message. It's considered as new protocol not supported.
+ String hbNotSupportedMsg = "The cluster does not support the new consumer group protocol. Set group" +
+ ".protocol=classic on the consumer configs to revert to the classic protocol until the cluster is upgraded.";
+ assertInstanceOf(Errors.UNSUPPORTED_VERSION.exception().getClass(), errorEvent.error());
+ assertEquals(hbNotSupportedMsg, errorEvent.error().getMessage());
+ clearInvocations(backgroundEventHandler);
+
+ // UnsupportedApiVersion in HB response with custom message. Specific to required version not present, should
+ // keep the custom message.
+ String hbVersionNotSupportedMsg = "The cluster does not support resolution of SubscriptionPattern on version 0. " +
Review Comment:
Apologies for the unclear comment. I initially assumed this test case was intended to verify that the server responds with UNSUPPORTED_VERSION when a SubscriptionPattern is combined with version 0. However, it appears the test case is actually used to verify the "custom error message."
In short, please disregard my previous comment. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1866317609
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -59,6 +60,9 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
@Override
public ConsumerGroupHeartbeatRequest build(short version) {
+ if (version == 0 && data.subscribedTopicRegex() != null) {
+ throw new UnsupportedVersionException("Broker does not support resolution of SubscriptionPattern on version " + version);
Review Comment:
you're right, we have the same exception but in 2 diff cases, so I made the changes to separate the messages, one for the case of HB API not supported (protocol is not enabled), the other one for when a required HB version is not supported (ex. v1 for regex). Also changed the msgs a bit to make them consistent, thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17989:
URL: https://github.com/apache/kafka/pull/17989#issuecomment-2520776896
Hey here, I opened a PR to fix a unit test that went in broken from this PR, but honestly I don't understand the situation. I see all green on the github actions liked to the merge (obviously I'm missing something about it)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1866397636
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -602,6 +602,44 @@ public void testHeartbeatResponseOnErrorHandling(final Errors error, final boole
}
}
+ @Test
+ public void testUnsupportedVersion() {
+ mockErrorResponse(Errors.UNSUPPORTED_VERSION, Errors.UNSUPPORTED_VERSION.message());
+ ArgumentCaptor<ErrorEvent> errorEventArgumentCaptor = ArgumentCaptor.forClass(ErrorEvent.class);
+ verify(backgroundEventHandler).add(errorEventArgumentCaptor.capture());
+ ErrorEvent errorEvent = errorEventArgumentCaptor.getValue();
+
+ // UnsupportedApiVersion in HB response without any custom message. It's considered as new protocol not supported.
+ String hbNotSupportedMsg = "The cluster does not support the new consumer group protocol. Set group" +
+ ".protocol=classic on the consumer configs to revert to the classic protocol until the cluster is upgraded.";
+ assertInstanceOf(Errors.UNSUPPORTED_VERSION.exception().getClass(), errorEvent.error());
+ assertEquals(hbNotSupportedMsg, errorEvent.error().getMessage());
+ clearInvocations(backgroundEventHandler);
+
+ // UnsupportedApiVersion in HB response with custom message. Specific to required version not present, should
+ // keep the custom message.
+ String hbVersionNotSupportedMsg = "The cluster does not support resolution of SubscriptionPattern on version 0. " +
Review Comment:
Do we have a guard on the server side? I assumed that the `SubscriptionPattern` check would be within `throwIfConsumerGroupHeartbeatRequestIsInvalid`, but I was unable to locate it.
https://github.com/apache/kafka/blob/6fd951a9c0aa773060cd6bbf8a8b8c47ee9d2965/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java#L1153
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -59,6 +60,9 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
@Override
public ConsumerGroupHeartbeatRequest build(short version) {
+ if (version == 0 && data.subscribedTopicRegex() != null) {
+ throw new UnsupportedVersionException("Broker does not support resolution of SubscriptionPattern on version " + version);
Review Comment:
Pardon me, why do we need to handle the version check manually instead of relying on the generated code? Alternatively, is this PR intended to add more meaningful error messages? If so, could you please include the explicit valid version number (or broker version) in the error message?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1866507930
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -59,6 +60,9 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
@Override
public ConsumerGroupHeartbeatRequest build(short version) {
+ if (version == 0 && data.subscribedTopicRegex() != null) {
+ throw new UnsupportedVersionException("Broker does not support resolution of SubscriptionPattern on version " + version);
Review Comment:
We could indeed rely on the auto-generated code but the error message is not so great.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1865582471
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -59,6 +60,9 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
@Override
public ConsumerGroupHeartbeatRequest build(short version) {
+ if (version == 0 && data.subscribedTopicRegex() != null) {
+ throw new UnsupportedVersionException("Broker does not support resolution of SubscriptionPattern on version " + version);
Review Comment:
We need to check how we handle this error in the hb request manager. If I recall correctly, we assume that the protocol is not supported when we get it. It may be confusing.
I also note that we actually handle UNSUPPORTED_VERSION in the AbstractHeartbeatRequestManager and the error message is specific to the consumer group hb request.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1868097845
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -313,12 +315,32 @@ private void onFailure(final Throwable exception, final long responseTimeMs) {
logger.debug(message);
} else {
logger.error("{} failed due to fatal error: {}", heartbeatRequestName(), exception.getMessage());
- handleFatalFailure(exception);
+ 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.
+ String msg = "The cluster does not support the new consumer group protocol. " +
Review Comment:
Should we use the same error message here? We could perhaps have it in a constant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1866536192
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -602,6 +602,44 @@ public void testHeartbeatResponseOnErrorHandling(final Errors error, final boole
}
}
+ @Test
+ public void testUnsupportedVersion() {
+ mockErrorResponse(Errors.UNSUPPORTED_VERSION, Errors.UNSUPPORTED_VERSION.message());
+ ArgumentCaptor<ErrorEvent> errorEventArgumentCaptor = ArgumentCaptor.forClass(ErrorEvent.class);
+ verify(backgroundEventHandler).add(errorEventArgumentCaptor.capture());
+ ErrorEvent errorEvent = errorEventArgumentCaptor.getValue();
+
+ // UnsupportedApiVersion in HB response without any custom message. It's considered as new protocol not supported.
+ String hbNotSupportedMsg = "The cluster does not support the new consumer group protocol. Set group" +
+ ".protocol=classic on the consumer configs to revert to the classic protocol until the cluster is upgraded.";
+ assertInstanceOf(Errors.UNSUPPORTED_VERSION.exception().getClass(), errorEvent.error());
+ assertEquals(hbNotSupportedMsg, errorEvent.error().getMessage());
+ clearInvocations(backgroundEventHandler);
+
+ // UnsupportedApiVersion in HB response with custom message. Specific to required version not present, should
+ // keep the custom message.
+ String hbVersionNotSupportedMsg = "The cluster does not support resolution of SubscriptionPattern on version 0. " +
Review Comment:
ok, this is intended to verify that the client generates the Unsupported error with a custom msg (on build, so it really never makes it to the broker)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1869333289
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -313,12 +319,29 @@ private void onFailure(final Throwable exception, final long responseTimeMs) {
logger.debug(message);
} else {
logger.error("{} failed due to fatal error: {}", heartbeatRequestName(), exception.getMessage());
- handleFatalFailure(exception);
+ 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));
+ } else {
+ // This is the case where building the request fails even though the node supports the API (ex.
+ // required version 1 not available when regex in use).
+ handleFatalFailure(exception);
+ }
}
// Notify the group manager about the failure after all errors have been handled and propagated.
membershipManager().onHeartbeatFailure(exception instanceof RetriableException);
}
+ /***
+ * @return True if the exception is the UnsupportedVersion generated on the client, before sending the request,
+ * when checking if the API is available on the broker.
+ */
+ private boolean isHBApiUnsupportedErrorMsg(Throwable exception) {
+ return exception instanceof UnsupportedVersionException &&
+ exception.getMessage().equals("The node does not support " + ApiKeys.CONSUMER_GROUP_HEARTBEAT);
Review Comment:
Comparing strings is not stable since they can be unintentionally changed. Could we create a subclass of `UnsupportedVersionException` to represent unsupported fields? Perhaps name it `UnsupportedProtocolFieldException`. Additionally, this exception could provide more accurate information by indicating the specific field, the current version, and the supported versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1867236561
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -59,6 +60,11 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
@Override
public ConsumerGroupHeartbeatRequest build(short version) {
+ if (version == 0 && data.subscribedTopicRegex() != null) {
+ throw new UnsupportedVersionException("The cluster does not support regular expressions resolution " +
+ "on version " + version + ". It must be upgraded to version >= 1 to allow to subscribe to a " +
+ "SubscriptionPattern.");
Review Comment:
nit: I wonder whether we should precise that the version is the version of the ConsumerGroupHeartbeat API. Users may confused it with the version of the client or the server. What do you think?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -382,8 +382,16 @@ private void onErrorResponse(final R response, final long currentTimeMs) {
break;
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.";
+ if (errorMessage == null) {
+ message = "The cluster does not support the new consumer group protocol. " +
+ "Set group.protocol=classic on the consumer configs to revert to " +
+ "the classic protocol until the cluster is upgraded.";
+ } else {
+ // Keep custom message present in the error. This could be the case where
+ // HB version required for a feature being used is not available
+ // (ex. regex requires HB v>0), or any other custom message included in the response.
+ message = errorMessage;
+ }
Review Comment:
As a second thought, I believe that we don't need to change this code at all. My understanding is that we reach this code only if the server returns a response containing the UNSUPPORTED_VERSION error. If the request building fails because the required version of the API is not available, it lands in `onFailure` directly. Hence, I believe that your custom error goes through directly.
This also means that the custom message that we use here for when the new protocol is not available is only provided if the request goes to the server. If it fails locally, it won't be used at all. Do you confirm?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17989:
URL: https://github.com/apache/kafka/pull/17989#issuecomment-2516069465
Filed https://issues.apache.org/jira/browse/KAFKA-18135 to review and refactor where the UnsupportedVersion is handled, mainly for share consumer, that is currently inheriting the msgs defined in the parent AbstractMembershipMgr which do not apply to it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1871637163
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -313,12 +319,29 @@ private void onFailure(final Throwable exception, final long responseTimeMs) {
logger.debug(message);
} else {
logger.error("{} failed due to fatal error: {}", heartbeatRequestName(), exception.getMessage());
- handleFatalFailure(exception);
+ 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));
+ } else {
+ // This is the case where building the request fails even though the node supports the API (ex.
+ // required version 1 not available when regex in use).
+ handleFatalFailure(exception);
+ }
}
// Notify the group manager about the failure after all errors have been handled and propagated.
membershipManager().onHeartbeatFailure(exception instanceof RetriableException);
}
+ /***
+ * @return True if the exception is the UnsupportedVersion generated on the client, before sending the request,
+ * when checking if the API is available on the broker.
+ */
+ private boolean isHBApiUnsupportedErrorMsg(Throwable exception) {
+ return exception instanceof UnsupportedVersionException &&
+ exception.getMessage().equals("The node does not support " + ApiKeys.CONSUMER_GROUP_HEARTBEAT);
Review Comment:
Note that this one is not about the unsupported fields but about the unsupported api. I agree that we could great a sub class which carry on more information (e.g. the api key and version).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17989:
URL: https://github.com/apache/kafka/pull/17989#issuecomment-2516077579
Also filed https://issues.apache.org/jira/browse/KAFKA-18154 given that a previous build here failed with a test timeout on ShareConsumerTest `destroyCluster`, that seems unrelated to me. The test does take really long locally, both on trunk and on this branch, around 12 mins (with a 20 min timeout).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1866520435
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -59,6 +60,9 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
@Override
public ConsumerGroupHeartbeatRequest build(short version) {
+ if (version == 0 && data.subscribedTopicRegex() != null) {
+ throw new UnsupportedVersionException("Broker does not support resolution of SubscriptionPattern on version " + version);
Review Comment:
Agree, this allows us to show an error msg specific to the field and version required (and it's also detecting it while building the request, so avoiding the actual HB request doomed to fail).
I updated the msg to add the version required as suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17989:
URL: https://github.com/apache/kafka/pull/17989#discussion_r1868073149
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -382,8 +382,16 @@ private void onErrorResponse(final R response, final long currentTimeMs) {
break;
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.";
+ if (errorMessage == null) {
+ message = "The cluster does not support the new consumer group protocol. " +
+ "Set group.protocol=classic on the consumer configs to revert to " +
+ "the classic protocol until the cluster is upgraded.";
+ } else {
+ // Keep custom message present in the error. This could be the case where
+ // HB version required for a feature being used is not available
+ // (ex. regex requires HB v>0), or any other custom message included in the response.
+ message = errorMessage;
+ }
Review Comment:
you're right:
1. If the request fails to build on the client, it lands directly into the `onFailure` because the response is created with the `UnsupportedVersion` exception
https://github.com/apache/kafka/blob/a8cdbaf4b30a119c66a9a85f6cab32610cd17fa9/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L588-L590 (meaning that the custom msg for v1 required for regex goes out directly, nothing extra needed)
2. the custom msg for protocol not supported that was here won't actually come out if the request fails locally because the client will fail when checking the supported apis, and go directly to `onFailure` again, with this:
https://github.com/apache/kafka/blob/a8cdbaf4b30a119c66a9a85f6cab32610cd17fa9/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java#L157 (We still need to keep the msg here to make sure that it shows in the case where the broker does support the HB api but the new protocol is not enabled)
So, I made the changes for both. I added different msgs for the case where protocol is not supported (old broker), vs protocol not enabled, thoughts?
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -59,6 +60,11 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
@Override
public ConsumerGroupHeartbeatRequest build(short version) {
+ if (version == 0 && data.subscribedTopicRegex() != null) {
+ throw new UnsupportedVersionException("The cluster does not support regular expressions resolution " +
+ "on version " + version + ". It must be upgraded to version >= 1 to allow to subscribe to a " +
+ "SubscriptionPattern.");
Review Comment:
makes sense , done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18127: Validate SubscriptionPattern used on v0 HB [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #17989:
URL: https://github.com/apache/kafka/pull/17989
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org