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 <apoorvmittal10@gmail.com>

Author: smjn

core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java

@@ -115,8 +115,15 @@ public CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>> readAsy
         }
 
         // 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<>();

core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java

@@ -381,6 +381,23 @@ public void testReadAsyncReturnsErrorFromLocalRead() throws Exception {
         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();