mjsax commented on code in PR #23262:
URL: https://github.com/apache/kafka/pull/23262#discussion_r4419435994
##########
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:
Makes sense. However, Claud judges the impact slight differently:
```
▎ Thanks. Addressed all of this in the latest commit, but some details
differ from the description above:
▎
▎ - In-flight flag: It doesn't block anything in practice.
handleFatalFailure() moves the member to FATAL, which is terminal. poll() and
maximumTimeToWait() return early
▎ on shouldSkipHeartbeat(), which covers FATAL, before they look at the
request state. Still, the unknown-status path now does the same bookkeeping as
the fatal path in
▎ onFailure(): onFailedAttempt() and heartbeatState.reset().
▎ - onFatalHeartbeatFailure(): Added, so the rebalance-failed metric is
recorded as on the other fatal paths.
▎ - Pending leave: Good catch that the leave future is left uncompleted,
but calling onFatalHeartbeatFailure() would not complete it. By then
transitionToFatal() has
▎ already set the state to FATAL, and onHeartbeatFailure() only completes
the leave for an UNSUBSCRIBED member. The gap is in
▎ StreamsMembershipManager.transitionToFatal() itself: its UNSUBSCRIBED
branch returns without completing the pending leave, which the consumer's
▎ AbstractMembershipManager.transitionToFatal() does. So it affects every
fatal error on a leave heartbeat (e.g. GROUP_AUTHORIZATION_FAILED), not only an
unknown status
▎ code. A leave response never carries a status today: the coordinator
sends an empty list. Since the gap is the same on all branches we cherry-pick
to, I fixed it in
▎ this PR too, and added a test for 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
