Improve the efficiency of ReplicaManager
Reviewers: Jun Rao Ismael Juma Dong Lin
@@ -830,10 +830,15 @@ class ReplicaManager(val config: KafkaConfig,
val logReadResults = readFromLog()
// check if this fetch request can be satisfied right away
- val logReadResultValues = logReadResults.map { case (_, v) => v }
- val bytesReadable = logReadResultValues.map(_.info.records.sizeInBytes).sum
- val errorReadingData = logReadResultValues.foldLeft(false) ((errorIncurred, readResult) =>
- errorIncurred || (readResult.error != Errors.NONE))
+ var bytesReadable: Long = 0
+ var errorReadingData = false
+ val logReadResultMap = new mutable.HashMap[TopicPartition, LogReadResult]
+ logReadResults.foreach { case (topicPartition, logReadResult) =>
+ if (logReadResult.error != Errors.NONE)
+ errorReadingData = true
+ bytesReadable = bytesReadable + logReadResult.info.records.sizeInBytes
+ logReadResultMap.put(topicPartition, logReadResult)
+ }
// respond immediately if 1) fetch request does not want to wait
// 2) fetch request does not require any data
@@ -847,11 +852,12 @@ class ReplicaManager(val config: KafkaConfig,
responseCallback(fetchPartitionData)
} else {
// construct the fetch results from the read results
- val fetchPartitionStatus = logReadResults.map { case (topicPartition, result) =>
- val fetchInfo = fetchInfos.collectFirst {
- case (tp, v) if tp == topicPartition => v
- }.getOrElse(sys.error(s"Partition $topicPartition not found in fetchInfos"))
- (topicPartition, FetchPartitionStatus(result.info.fetchOffsetMetadata, fetchInfo))
+ val fetchPartitionStatus = new mutable.ArrayBuffer[(TopicPartition, FetchPartitionStatus)]
+ fetchInfos.foreach { case (topicPartition, partitionData) =>
+ logReadResultMap.get(topicPartition).map(logReadResult => {
+ val logOffsetMetadata = logReadResult.info.fetchOffsetMetadata
+ fetchPartitionStatus += (topicPartition -> FetchPartitionStatus(logOffsetMetadata, partitionData))
+ })
}
val fetchMetadata = FetchMetadata(fetchMinBytes, fetchMaxBytes, hardMaxBytesLimit, fetchOnlyFromLeader,
fetchIsolation, isFromFollower, replicaId, fetchPartitionStatus)
