jeqo commented on code in PR #19197:
URL: https://github.com/apache/kafka/pull/19197#discussion_r2001527540
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1544,12 +1544,14 @@ class ReplicaManager(val config: KafkaConfig,
fetchPartitionStatus: Seq[(TopicIdPartition,
FetchPartitionStatus)]): Option[LogReadResult] = {
val key = new
TopicPartitionOperationKey(remoteFetchInfo.topicPartition.topic(),
remoteFetchInfo.topicPartition.partition())
val remoteFetchResult = new CompletableFuture[RemoteLogReadResult]
- var remoteFetchTask: Future[Void] = null
+ var remoteFetchFuture: RemoteFetchFuture = null
try {
- remoteFetchTask = remoteLogManager.get.asyncRead(remoteFetchInfo,
(result: RemoteLogReadResult) => {
+ val remoteLogReader =
remoteLogManager.get.remoteLogReaderTask(remoteFetchInfo, (result:
RemoteLogReadResult) => {
remoteFetchResult.complete(result)
delayedRemoteFetchPurgatory.checkAndComplete(key)
})
+ val remoteFetchTask = remoteLogManager.get.asyncRead(remoteLogReader)
+ remoteFetchFuture = new RemoteFetchFuture(remoteLogReader,
remoteFetchTask)
Review Comment:
I think you mean the result future right? If yes; agree, I missed it.
I'm passing both future as input now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
