(kafka) branch trunk updated: KAFKA-20713: Lower level for client log on self-healing session not found (#22617)

[li...@apache.org]

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 85b0e80272d KAFKA-20713: Lower level for client log on self-healing session not found (#22617)
85b0e80272d is described below

commit 85b0e80272d1c594d01a37273d68e555c636765d
Author: Lianet Magrans <98...@users.noreply.github.com>
AuthorDate: Fri Jun 19 09:14:44 2026 -0400

    KAFKA-20713: Lower level for client log on self-healing session not found (#22617)
    
    Lower to DEBUG and update message for the fetch session not found
    (evicted) scenario. The previous INFO log reported this as an error, but
    on the client side it is self-healing and not actionable.
    
    Fetch sessions are managed on the broker, which already exposes the
    visibility needed (metrics and logs) to detect this situation and
    correct it (broker configs for fetch session cache)
    
    Reviewers: Andrew Schofield <as...@confluent.io>
---
 .../java/org/apache/kafka/clients/FetchSessionHandler.java   | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java b/clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java
index b5150b2d7f1..ab108ceab73 100644
--- a/clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java
@@ -526,11 +526,19 @@ public class FetchSessionHandler {
      */
     public boolean handleResponse(FetchResponse response, short version) {
         if (response.error() != Errors.NONE) {
-            log.info("Node {} was unable to process the fetch request with {}: {}.",
-                node, nextMetadata, response.error());
             if (response.error() == Errors.FETCH_SESSION_ID_NOT_FOUND) {
+                // Session does not exist on the broker anymore. Recoverable and self-healing, the client re-sends a full fetch request.
+                log.debug("Node {} returned a {} error; the fetch session {} was likely evicted from the broker's " +
+                    "fetch session cache. Re-sending a full fetch request to establish a new session.",
+                    node, response.error(), nextMetadata.sessionId());
                 nextMetadata = FetchMetadata.INITIAL;
             } else {
+                // Other fetch-session errors (e.g. INVALID_FETCH_SESSION_EPOCH, FETCH_SESSION_TOPIC_ID_ERROR) are
+                // also recoverable and self-healing: the existing session is closed and a new one is re-established
+                // with a full fetch.
+                log.debug("Node {} was unable to process the fetch request with {}: {}. " +
+                    "Re-sending a full fetch request, which closes the existing session on the broker and establishes a new one.",
+                    node, nextMetadata, response.error());
                 nextMetadata = nextMetadata.nextCloseExistingAttemptNew();
             }
             return false;