lianetm commented on code in PR #22691:
URL: https://github.com/apache/kafka/pull/22691#discussion_r3493615913
##########
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:
is it worth splitting the logic in this share case? The share consumer
handles all session-related errors the same way (log +
nextCloseExistingAttemptNew) vs the consumer which had different handling for
the initial fetch
So what about we simplify here, keep a single log? (maybe with the added
explanation that it sends a full request to establish a new session)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java:
##########
@@ -296,8 +311,22 @@ public boolean handleResponse(ShareFetchResponse response,
short version) {
public boolean handleResponse(ShareAcknowledgeResponse response, short
version) {
if ((response.error() == Errors.SHARE_SESSION_NOT_FOUND) ||
(response.error() == Errors.INVALID_SHARE_SESSION_EPOCH)) {
- log.info("Node {} was unable to process the ShareAcknowledge
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 ShareAcknowledge 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
ShareAcknowledge request to establish a new session.",
+ node, response.error(), nextMetadata.memberId());
+ }
+ } else {
+ // Other share-session errors (e.g.
INVALID_SHARE_SESSION_EPOCH) are also recoverable and self-healing:
+ // the existing session is closed and a new one is
re-established with a full ShareAcknowledge.
+ if (log.isDebugEnabled()) {
+ log.debug("Node {} was unable to process the
ShareAcknowledge request with {}: {}. " +
+ "Re-sending a full ShareAcknowledge request, which
closes the existing session on the broker and establishes a new one.",
+ node, nextMetadata, response.error());
+ }
+ }
Review Comment:
same for this case
--
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]
