(kafka) branch trunk updated: KAFKA-20720: Lower level for client log on self-healing share session… (#22691)

[lianetm]

This is an automated email from the ASF dual-hosted git repository.

lianetm pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new eb8f9c916cf KAFKA-20720: Lower level for client log on self-healing 
share session… (#22691)
eb8f9c916cf is described below

commit eb8f9c916cf9d0555d8aabd8b0787ffcbb4783cd
Author: nileshkumar3 <nileshkum...@gmail.com>
AuthorDate: Tue Jun 30 13:29:12 2026 -0500

    KAFKA-20720: Lower level for client log on self-healing share session… 
(#22691)
    
    ## Summary
    Share consumers log recoverable share-session errors at INFO when the
    broker returns `SHARE_SESSION_NOT_FOUND`, `INVALID_SHARE_SESSION_EPOCH`,
    or `SHARE_SESSION_LIMIT_REACHED` on ShareFetch/ShareAcknowledge
    responses. These are broker-managed, self-healing conditions: the client
    closes the existing session and re-establishes it via
    `nextCloseExistingAttemptNew()`.
    This change lowers those logs to DEBUG and updates the messages to
    explain the recovery behavior, matching the approach already taken for
    fetch sessions in `FetchSessionHandler` (KAFKA-20713).
    
    Reviewers - @lianetm
    
    Reviewers: Lianet Magrans <lmagr...@confluent.io>
---
 .../consumer/internals/ShareSessionHandler.java      | 20 ++++++++++++++++----
 1 file changed, 16 insertions(+), 4 deletions(-)

diff --git 
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
 
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
index b73500e2dd4..5b92e871e0e 100644
--- 
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
+++ 
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
@@ -264,8 +264,15 @@ public class ShareSessionHandler {
         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());
+            // These share-session errors are handled the same way: close the 
existing session (if any) and re-establish
+            // it by re-sending a full ShareFetch request. For 
SHARE_SESSION_NOT_FOUND / INVALID_SHARE_SESSION_EPOCH this
+            // recovers on the next request; for SHARE_SESSION_LIMIT_REACHED 
it succeeds once the broker's session cache
+            // has capacity.
+            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());
+            }
             nextMetadata = nextMetadata.nextCloseExistingAttemptNew();
             return false;
         }
@@ -296,8 +303,13 @@ public class ShareSessionHandler {
     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());
+            // These share-session errors are handled the same way: close the 
existing session (if any) and re-establish
+            // it by re-sending a full ShareAcknowledge request, which 
recovers on the next request.
+            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());
+            }
             nextMetadata = nextMetadata.nextCloseExistingAttemptNew();
             return false;
         }