This is an automated email from the ASF dual-hosted git repository.
chia7712 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 2f04be7b0dc MINOR: Updated log reader interface to abstractions
(#22841)
2f04be7b0dc is described below
commit 2f04be7b0dc31ddd840bff0768cfad3e418a6a23
Author: Apoorv Mittal <[email protected]>
AuthorDate: Thu Jul 16 22:58:45 2026 +0100
MINOR: Updated log reader interface to abstractions (#22841)
The minor PR updates the LogReader interface to match the other passed
params abstractions rather specific Map implementation.
Reviewers: Chia-Ping Tsai <[email protected]>
---
.../server/share/ReplicaManagerLogReader.java | 9 +++++----
.../org/apache/kafka/server/share/LogReader.java | 22 ++++++++++++++--------
2 files changed, 19 insertions(+), 12 deletions(-)
diff --git a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
index fbc7e734255..dca782a8ee8 100644
--- a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
+++ b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
@@ -34,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.LinkedHashMap;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -65,8 +66,8 @@ public class ReplicaManagerLogReader implements LogReader {
public CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readAsync(
FetchParams fetchParams,
Set<TopicIdPartition> partitionsToFetch,
- LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
- LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes,
+ Map<TopicIdPartition, Long> topicPartitionFetchOffsets,
+ Map<TopicIdPartition, Integer> partitionMaxBytes,
boolean readRemote) {
if (partitionsToFetch.isEmpty()) {
@@ -125,8 +126,8 @@ public class ReplicaManagerLogReader implements LogReader {
LinkedHashMap<TopicIdPartition, LogReadResult> read(
FetchParams fetchParams,
Set<TopicIdPartition> partitionsToFetch,
- LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
- LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes
+ Map<TopicIdPartition, Long> topicPartitionFetchOffsets,
+ Map<TopicIdPartition, Integer> partitionMaxBytes
) {
if (partitionsToFetch.isEmpty()) {
return new LinkedHashMap<>();
diff --git a/server/src/main/java/org/apache/kafka/server/share/LogReader.java
b/server/src/main/java/org/apache/kafka/server/share/LogReader.java
index 67362188e6a..0842ec22f46 100644
--- a/server/src/main/java/org/apache/kafka/server/share/LogReader.java
+++ b/server/src/main/java/org/apache/kafka/server/share/LogReader.java
@@ -21,6 +21,7 @@ import org.apache.kafka.server.storage.log.FetchParams;
import org.apache.kafka.storage.internals.log.LogReadResult;
import java.util.LinkedHashMap;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -34,17 +35,22 @@ public interface LogReader {
* and - when {@code readRemote} is true and the requested data has been
tiered off the local log - the
* follow-up remote read into a single call.
*
- * <p>This is the asynchronous, remote-aware counterpart to {@link #read}:
it returns a single future
- * holding one {@link LogReadResult} per requested partition. The future
completes once every partition
- * has resolved - partitions available locally (or whose local read
failed) resolve immediately, while
- * partitions whose data is in remote storage resolve later, once the
remote read finishes on the remote
- * storage reader pool, so the caller's thread is never blocked on remote
IO. When {@code readRemote} is
- * false, tiered offsets are simply omitted from the result rather than
fetched.
+ * <p>The read is asynchronous and remote-aware: it returns a single
future holding one {@link LogReadResult}
+ * per requested partition. The future completes once every partition has
resolved - partitions
+ * available locally (or whose local read failed) resolve immediately,
while partitions whose data
+ * is in remote storage resolve later, once the remote read finishes on
the remote storage reader
+ * pool, so the caller's thread is never blocked on remote IO. When {@code
readRemote} is false,
+ * tiered offsets are simply omitted from the result rather than fetched.
*
* <p>Each per-partition {@link LogReadResult} is partial-data tolerant:
the read never fails as a whole,
* allowing callers to use whatever records were retrieved (via {@link
LogReadResult#info()}) and skip
* the rest based on {@link LogReadResult#error()}.
*
+ * <p>If the order of the fetched partitions matters to the caller, it
should pass order-preserving
+ * {@link Set} and {@link Map} implementations (for example {@link
java.util.LinkedHashSet} and
+ * {@link LinkedHashMap}). LogReader implementations must then honour that
iteration order and return
+ * the partitions in the same order in the resulting {@link LinkedHashMap}.
+ *
* @param fetchParams The fetch parameters (isolation
level, maxBytes, etc.)
* @param partitionsToFetch The set of partitions to fetch
* @param topicPartitionFetchOffsets The fetch offset per partition
@@ -56,7 +62,7 @@ public interface LogReader {
CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readAsync(
FetchParams fetchParams,
Set<TopicIdPartition> partitionsToFetch,
- LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
- LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes,
+ Map<TopicIdPartition, Long> topicPartitionFetchOffsets,
+ Map<TopicIdPartition, Integer> partitionMaxBytes,
boolean readRemote);
}
