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 0e5fb238f7a KAFKA-20783: LogReader interface cleanup (2/N) (#22795)
0e5fb238f7a is described below
commit 0e5fb238f7a2a83382e4c03e33501be3663d6c6f
Author: Apoorv Mittal <[email protected]>
AuthorDate: Thu Jul 9 15:37:46 2026 +0100
KAFKA-20783: LogReader interface cleanup (2/N) (#22795)
Minor PR to cleanup LogReader interface as now DleayedShareFetch is
moved to async handling of reads.
Reviewers: Abhinav Dixit <[email protected]>, Andrew Schofield
<[email protected]>
---
.../server/share/ReplicaManagerLogReader.java | 82 +++++++++++-----------
.../org/apache/kafka/server/share/LogReader.java | 15 ----
2 files changed, 41 insertions(+), 56 deletions(-)
diff --git a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
index 14a6ecdbd29..fbc7e734255 100644
--- a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
+++ b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
@@ -61,47 +61,6 @@ public class ReplicaManagerLogReader implements LogReader {
this.replicaManager = replicaManager;
}
- @Override
- public LinkedHashMap<TopicIdPartition, LogReadResult> read(
- FetchParams fetchParams,
- Set<TopicIdPartition> partitionsToFetch,
- LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
- LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes) {
-
- if (partitionsToFetch.isEmpty()) {
- return new LinkedHashMap<>();
- }
-
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData>
topicPartitionData = new LinkedHashMap<>();
- topicPartitionFetchOffsets.forEach((topicIdPartition, fetchOffset) ->
- topicPartitionData.put(topicIdPartition,
- new FetchRequest.PartitionData(
- topicIdPartition.topicId(),
- fetchOffset,
- 0,
- partitionMaxBytes.get(topicIdPartition),
- Optional.empty())
- ));
-
- Seq<Tuple2<TopicIdPartition, LogReadResult>> responseLogResult =
replicaManager.readFromLog(
- fetchParams,
- CollectionConverters.asScala(
- partitionsToFetch.stream().map(topicIdPartition ->
- new Tuple2<>(topicIdPartition,
topicPartitionData.get(topicIdPartition))).collect(Collectors.toList())
- ),
- QuotaFactory.UNBOUNDED_QUOTA,
- true);
-
- LinkedHashMap<TopicIdPartition, LogReadResult> responseData = new
LinkedHashMap<>();
- responseLogResult.foreach(tpLogResult -> {
- responseData.put(tpLogResult._1(), tpLogResult._2());
- return BoxedUnit.UNIT;
- });
-
- log.trace("Data successfully retrieved by replica manager: {}",
responseData);
- return responseData;
- }
-
@Override
public CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readAsync(
FetchParams fetchParams,
@@ -162,6 +121,47 @@ public class ReplicaManagerLogReader implements LogReader {
});
}
+ // Visible for testing
+ LinkedHashMap<TopicIdPartition, LogReadResult> read(
+ FetchParams fetchParams,
+ Set<TopicIdPartition> partitionsToFetch,
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
+ LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes
+ ) {
+ if (partitionsToFetch.isEmpty()) {
+ return new LinkedHashMap<>();
+ }
+
+ LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData>
topicPartitionData = new LinkedHashMap<>();
+ topicPartitionFetchOffsets.forEach((topicIdPartition, fetchOffset) ->
+ topicPartitionData.put(topicIdPartition,
+ new FetchRequest.PartitionData(
+ topicIdPartition.topicId(),
+ fetchOffset,
+ 0,
+ partitionMaxBytes.get(topicIdPartition),
+ Optional.empty())
+ ));
+
+ Seq<Tuple2<TopicIdPartition, LogReadResult>> responseLogResult =
replicaManager.readFromLog(
+ fetchParams,
+ CollectionConverters.asScala(
+ partitionsToFetch.stream().map(topicIdPartition ->
+ new Tuple2<>(topicIdPartition,
topicPartitionData.get(topicIdPartition))).collect(Collectors.toList())
+ ),
+ QuotaFactory.UNBOUNDED_QUOTA,
+ true);
+
+ LinkedHashMap<TopicIdPartition, LogReadResult> responseData = new
LinkedHashMap<>();
+ responseLogResult.foreach(tpLogResult -> {
+ responseData.put(tpLogResult._1(), tpLogResult._2());
+ return BoxedUnit.UNIT;
+ });
+
+ log.trace("Data successfully retrieved by replica manager: {}",
responseData);
+ return responseData;
+ }
+
/**
* Returns a copy of {@code base} with its read data ({@code info}) and
{@code error} replaced, preserving
* all other read metadata (high watermark, log offsets, etc.). Used to
wrap a remote read into a
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 3ae9500f9f4..67362188e6a 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
@@ -29,21 +29,6 @@ import java.util.concurrent.CompletableFuture;
*/
public interface LogReader {
- /**
- * Read records for the given partitions starting at the specified offsets.
- *
- * @param fetchParams The fetch parameters (isolation level,
maxBytes, etc.)
- * @param partitionsToFetch The set of partitions to actually fetch
(after filtering erroneous ones)
- * @param topicPartitionFetchOffsets The fetch offset per partition
- * @param partitionMaxBytes The max bytes per partition
- * @return A map of partition to log read result
- */
- LinkedHashMap<TopicIdPartition, LogReadResult> read(
- FetchParams fetchParams,
- Set<TopicIdPartition> partitionsToFetch,
- LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
- LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes);
-
/**
* Read records for the given partitions starting at the specified
offsets, combining the local read
* and - when {@code readRemote} is true and the requested data has been
tiered off the local log - the
