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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
