(kafka) branch trunk updated: KAFKA-20678: Exception handling for read call in ReplicaManagerLogReader.readAsync. (#22753)

[mittal]

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

apoorvmittal10 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 1db1ae701c0 KAFKA-20678: Exception handling for read call in 
ReplicaManagerLogReader.readAsync. (#22753)
1db1ae701c0 is described below

commit 1db1ae701c0cecbf5d868fae2f376246410511c9
Author: Sushant Mahajan <smaha...@confluent.io>
AuthorDate: Wed Jul 8 01:52:02 2026 +0530

    KAFKA-20678: Exception handling for read call in 
ReplicaManagerLogReader.readAsync. (#22753)
    
    The `read` call in `readAsync` was not protected by a try catch block
    which could throw an exception in a non-async fashion to the caller.
    This PR surrounds the call in try-catch and adds appropriate tests.
    
    Reviewers: Apoorv Mittal <apoorvmitta...@gmail.com>
---
 .../kafka/server/share/ReplicaManagerLogReader.java     | 11 +++++++++--
 .../kafka/server/share/ReplicaManagerLogReaderTest.java | 17 +++++++++++++++++
 2 files changed, 26 insertions(+), 2 deletions(-)

diff --git a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java 
b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
index 70e798b6eb2..14a6ecdbd29 100644
--- a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
+++ b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
@@ -115,8 +115,15 @@ public class ReplicaManagerLogReader implements LogReader {
         }
 
         // Perform the local read for all partitions once; remote follow-ups 
(if any) are issued per partition.
-        LinkedHashMap<TopicIdPartition, LogReadResult> localReadResults =
-            read(fetchParams, partitionsToFetch, topicPartitionFetchOffsets, 
partitionMaxBytes);
+        // read() is expected to convert per-partition failures into a 
LogReadResult carrying an error code
+        // rather than throwing, but guard against an unexpected throw here 
anyway - readAsync must surface
+        // failures via the returned future, not by throwing synchronously out 
of this method.
+        LinkedHashMap<TopicIdPartition, LogReadResult> localReadResults;
+        try {
+            localReadResults = read(fetchParams, partitionsToFetch, 
topicPartitionFetchOffsets, partitionMaxBytes);
+        } catch (Exception e) {
+            return CompletableFuture.failedFuture(e);
+        }
 
         // Only look at partitions with non-null read results.
         LinkedHashMap<TopicIdPartition, CompletableFuture<LogReadResult>> 
futures = new LinkedHashMap<>();
diff --git 
a/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java 
b/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
index fb93cba205b..498298f621d 100644
--- a/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
+++ b/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
@@ -381,6 +381,23 @@ public class ReplicaManagerLogReaderTest {
         verify(replicaManager, never()).remoteLogManager();
     }
 
+    @Test
+    public void testReadAsyncCompletesExceptionallyWhenLocalReadThrows() {
+        ReplicaManager replicaManager = mockReplicaManager();
+        RuntimeException readError = new RuntimeException("read failed");
+        when(replicaManager.readFromLog(any(), any(), any(), 
anyBoolean())).thenThrow(readError);
+
+        ReplicaManagerLogReader logReader = new 
ReplicaManagerLogReader(replicaManager);
+        CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>> 
future =
+            logReader.readAsync(fetchParams(), Set.of(TOPIC_ID_PARTITION),
+                offsets(TOPIC_ID_PARTITION), maxBytes(TOPIC_ID_PARTITION), 
true);
+
+        // readAsync must surface the failure via the returned future rather 
than throwing synchronously.
+        assertTrue(future.isCompletedExceptionally());
+        ExecutionException exception = assertThrows(ExecutionException.class, 
() -> future.get(10, TimeUnit.SECONDS));
+        assertSame(readError, exception.getCause());
+    }
+
     @Test
     public void 
testReadAsyncReturnsNoneErrorWhenRemoteReadFailsButLocalSucceeds() throws 
Exception {
         ReplicaManager replicaManager = mockReplicaManager();