satishd commented on code in PR #13535: URL: https://github.com/apache/kafka/pull/13535#discussion_r1170269949
########## core/src/main/scala/kafka/server/ReplicaManager.scala: ########## @@ -1083,48 +1095,100 @@ class ReplicaManager(val config: KafkaConfig, fetchPartitionStatus += (topicIdPartition -> FetchPartitionStatus(logOffsetMetadata, partitionData)) }) } - val delayedFetch = new DelayedFetch( - params = params, - fetchPartitionStatus = fetchPartitionStatus, - replicaManager = this, - quota = quota, - responseCallback = responseCallback - ) - - // create a list of (topic, partition) pairs to use as keys for this delayed fetch operation - val delayedFetchKeys = fetchPartitionStatus.map { case (tp, _) => TopicPartitionOperationKey(tp) } - - // try to complete the request immediately, otherwise put it into the purgatory; - // this is because while the delayed fetch operation is being created, new requests - // may arrive and hence make this operation completable. - delayedFetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys) + + if (remoteFetchInfo.isPresent) { Review Comment: I am not sure line num:1082 is sane as you meant it to be as the file could have been updated. Please clarify. -- 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