Copilot commented on code in PR #23262:
URL: https://github.com/apache/kafka/pull/23262#discussion_r3918107188
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -649,6 +650,21 @@ private void onResponse(final
StreamsGroupHeartbeatResponse response, long curre
private void onSuccessResponse(final StreamsGroupHeartbeatResponse
response, final long currentTimeMs) {
final StreamsGroupHeartbeatResponseData data = response.data();
+
+ final List<String> unknownStatuses = unknownStatuses(data.status());
+ if (!unknownStatuses.isEmpty()) {
+ final String errorMessage = String.format(
+ "The group coordinator returned status %s in the Streams group
heartbeat response, which this client "
+ + "cannot interpret; it knows the status codes %s. A
status code must not be sent to a client that "
+ + "cannot interpret it, so this is a bug on the group
coordinator.",
+ unknownStatuses,
+ StreamsGroupHeartbeatResponse.Status.knownCodes()
+ );
+ logger.error(errorMessage);
+ handleFatalFailure(new KafkaException(errorMessage));
+ return;
+ }
Review Comment:
In the unknown-status early return, the request has completed but
`heartbeatRequestState` is never marked as received (neither
`onSuccessfulAttempt` nor `onFailedAttempt`). Since `onSendAttempt` already set
`requestInFlight=true`, leaving it unchanged makes the request state
inconsistent and can block future sends for the wrong reason (it will look like
a request is still in-flight). Consider clearing the in-flight flag (and
keeping failure bookkeeping consistent) before transitioning to fatal; also
call `membershipManager.onFatalHeartbeatFailure()` for parity with other fatal
paths that record rebalance-failed metrics.
--
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]
