kirktrue commented on code in PR #17411:
URL: https://github.com/apache/kafka/pull/17411#discussion_r1794399171
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1277,10 +1278,13 @@ private void releaseAssignmentAndLeaveGroup(final Timer
timer) {
UnsubscribeEvent unsubscribeEvent = new
UnsubscribeEvent(calculateDeadlineMs(timer));
applicationEventHandler.add(unsubscribeEvent);
try {
- // If users subscribe to an invalid topic name, they will get
InvalidTopicException in error events,
- // because network thread keeps trying to send MetadataRequest in
the background.
- // Ignore it to avoid unsubscribe failed.
- processBackgroundEvents(unsubscribeEvent.future(), timer, e -> e
instanceof InvalidTopicException);
+ // Network thread keeps trying to send MetadataRequest in the
background.
+ // If there is invalid request, there will have many error events.
We would like to ignore it to avoid unsubscribe fail.
+ // Ignore exceptions:
+ // InvalidTopicException: avoid invalid topic name.
+ // UnsupportedVersionException: avoid sending consumer group
protocol to unsupported broker.
+ processBackgroundEvents(unsubscribeEvent.future(), timer,
+ e -> e instanceof InvalidTopicException || e instanceof
UnsupportedVersionException);
Review Comment:
How would we get to the point of unsubscribing if the broker doesn't support
the group protocol the consumer is using? Or is this to handle the case where
`close()` attempts to unsubscribe and hits `UnsupportedVersionException`?
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3796,7 +3796,14 @@ class KafkaApis(val requestChannel: RequestChannel,
if (!isConsumerGroupProtocolEnabled()) {
// The API is not supported by the "old" group coordinator (the
default). If the
// new one is not enabled, we fail directly here.
- requestHelper.sendMaybeThrottle(request,
consumerGroupHeartbeatRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
+ requestHelper.sendMaybeThrottle(request, new
ConsumerGroupHeartbeatResponse(
+ new ConsumerGroupHeartbeatResponseData()
+ .setErrorCode(Errors.UNSUPPORTED_VERSION.code())
+ .setErrorMessage("Failed to handle consumer group heartbeat request
since the " +
+ "consumer group coordinator is not enabled. Please enable the
consumer group coordinator in the brokers configuration by setting " +
+
s"\"${GroupCoordinatorConfig.NEW_GROUP_COORDINATOR_ENABLE_CONFIG}\" to \"true\"
and " +
+
s"\"${GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG}\" to
\"classic, consumer\". " +
+ s"Or setting \"group.protocol\" to \"classic\" in the consumer
configuration.")))
Review Comment:
The message is a bit misleading as is because the group coordinator _is_
enabled, it's just that the `group.protocol` Some minor tweaks:
```suggestion
requestHelper.sendMaybeThrottle(request, new
ConsumerGroupHeartbeatResponse(
new ConsumerGroupHeartbeatResponseData()
.setErrorCode(Errors.UNSUPPORTED_VERSION.code())
.setErrorMessage("An error occurred handling the consumer group
heartbeat request. " +
s"To enable support for the ${Group.GroupType.CONSUMER} group
protocol, update the broker " +
s"configuration for
\"${GroupCoordinatorConfig.NEW_GROUP_COORDINATOR_ENABLE_CONFIG}\" to \"true\"
and " +
s"\"${GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG}\" to
" +
s"\"${Group.GroupType.CLASSIC}, ${Group.GroupType.CONSUMER}\".
Alternatively, update the client " +
s"configuration for \"group.protocol\" to
\"${Group.GroupType.CLASSIC}\".")))
```
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
