nileshkumar3 commented on code in PR #22691:
URL: https://github.com/apache/kafka/pull/22691#discussion_r3498773229
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java:
##########
@@ -264,8 +264,23 @@ public boolean handleResponse(ShareFetchResponse response,
short version) {
if ((response.error() == Errors.SHARE_SESSION_NOT_FOUND) ||
(response.error() == Errors.INVALID_SHARE_SESSION_EPOCH) ||
(response.error() == Errors.SHARE_SESSION_LIMIT_REACHED)) {
- log.info("Node {} was unable to process the ShareFetch request
with {}: {}.",
- node, nextMetadata, response.error());
+ if (response.error() == Errors.SHARE_SESSION_NOT_FOUND) {
+ // Session does not exist on the broker anymore. Recoverable
and self-healing, the client re-sends a full ShareFetch request.
+ if (log.isDebugEnabled()) {
+ log.debug("Node {} returned a {} error; the share session
for member {} was likely evicted from the broker's " +
+ "share session cache. Re-sending a full ShareFetch
request to establish a new session.",
+ node, response.error(), nextMetadata.memberId());
+ }
+ } else {
+ // Other share-session errors (e.g.
INVALID_SHARE_SESSION_EPOCH, SHARE_SESSION_LIMIT_REACHED) are
+ // also recoverable and self-healing: the existing session is
closed and a new one is re-established
+ // with a full ShareFetch.
+ if (log.isDebugEnabled()) {
+ log.debug("Node {} was unable to process the ShareFetch
request with {}: {}. " +
+ "Re-sending a full ShareFetch request, which
closes the existing session on the broker and establishes a new one.",
+ node, nextMetadata, response.error());
+ }
+ }
Review Comment:
Agreed — all the errors are treated the same way (reset to the initial epoch
via nextCloseExistingAttemptNew() and re-send a full request), so splitting
them isn't logical. I've removed the split and replaced it with a single log +
a proper comment.
One small caveat I kept in the comment: for SHARE_SESSION_NOT_FOUND /
INVALID_SHARE_SESSION_EPOCH the retry recovers on the next request, but
SHARE_SESSION_LIMIT_REACHED is a bit different —
ShareSessionCache.maybeCreateSession() does no eviction (unlike the regular
FetchSessionCache), so the full retry only succeeds once the broker's cache has
capacity. The handling is identical, the wording just avoids implying it's
instantly self-healing. Let me know if this is fine or if you'd prefer a fully
generic 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
