Skip to content

MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions #20768

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: trunk
Choose a base branch
from
+30 −17
Open

MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions #20768

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions core/src/main/scala/kafka/server/DelayedFetch.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import org.apache.kafka.server.purgatory.DelayedOperation
import org.apache.kafka.server.storage.log.{FetchIsolation, FetchParams, FetchPartitionData}
import org.apache.kafka.storage.internals.log.{FetchPartitionStatus, LogOffsetMetadata}

import java.util
import scala.collection._
import scala.jdk.CollectionConverters._

Expand All @@ -39,7 +40,7 @@ import scala.jdk.CollectionConverters._
*/
class DelayedFetch(
params: FetchParams,
fetchPartitionStatus: Seq[(TopicIdPartition, FetchPartitionStatus)],
fetchPartitionStatus: util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus],
replicaManager: ReplicaManager,
quota: ReplicaQuota,
responseCallback: Seq[(TopicIdPartition, FetchPartitionData)] => Unit
Expand All @@ -66,8 +67,7 @@ class DelayedFetch(
*/
override def tryComplete(): Boolean = {
var accumulatedSize = 0
fetchPartitionStatus.foreach {
case (topicIdPartition, fetchStatus) =>
fetchPartitionStatus.forEach { (topicIdPartition, fetchStatus) =>
val fetchOffset = fetchStatus.startOffsetMetadata
val fetchLeaderEpoch = fetchStatus.fetchInfo.currentLeaderEpoch
try {
Expand Down Expand Up @@ -154,9 +154,9 @@ class DelayedFetch(
* Upon completion, read whatever data is available and pass to the complete callback
*/
override def onComplete(): Unit = {
val fetchInfos = fetchPartitionStatus.map { case (tp, status) =>
val fetchInfos = fetchPartitionStatus.asScala.map { case (tp, status) =>
tp -> status.fetchInfo
}
}.toSeq
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible to have duplicate TPs in this "fetch" path?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, we are doing two conversions here - you can avoid one by using a view or iterator after asScala and using toBuffer instead of toSeq (the latter can result in a lazy collection being created which can result in problems).


val logReadResults = replicaManager.readFromLog(
params,
Expand Down
12 changes: 6 additions & 6 deletions core/src/main/scala/kafka/server/ReplicaManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1628,7 +1628,7 @@ class ReplicaManager(val config: KafkaConfig,
params: FetchParams,
responseCallback: Seq[(TopicIdPartition, FetchPartitionData)] => Unit,
logReadResults: util.LinkedHashMap[TopicIdPartition, LogReadResult],
fetchPartitionStatus: Seq[(TopicIdPartition, FetchPartitionStatus)]): Unit = {
fetchPartitionStatus: util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus]): Unit = {
val remoteFetchTasks = new util.HashMap[TopicIdPartition, Future[Void]]
val remoteFetchResults = new util.HashMap[TopicIdPartition, CompletableFuture[RemoteLogReadResult]]

Expand All @@ -1643,7 +1643,7 @@ class ReplicaManager(val config: KafkaConfig,
remoteFetchResults,
remoteFetchInfos,
remoteFetchMaxWaitMs,
fetchPartitionStatus.toMap.asJava,
fetchPartitionStatus,
params,
logReadResults,
tp => getPartitionOrException(tp),
Expand Down Expand Up @@ -1710,17 +1710,17 @@ class ReplicaManager(val config: KafkaConfig,
responseCallback(fetchPartitionData)
} else {
// construct the fetch results from the read results
val fetchPartitionStatus = new mutable.ArrayBuffer[(TopicIdPartition, FetchPartitionStatus)]
val fetchPartitionStatus = new util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus]
fetchInfos.foreach { case (topicIdPartition, partitionData) =>
val logReadResult = logReadResultMap.get(topicIdPartition)
if (logReadResult != null) {
val logOffsetMetadata = logReadResult.info.fetchOffsetMetadata
fetchPartitionStatus += (topicIdPartition -> new FetchPartitionStatus(logOffsetMetadata, partitionData))
fetchPartitionStatus.put(topicIdPartition, new FetchPartitionStatus(logOffsetMetadata, partitionData))
}
}

if (!remoteFetchInfos.isEmpty) {
processRemoteFetches(remoteFetchInfos, params, responseCallback, logReadResultMap, fetchPartitionStatus.toSeq)
processRemoteFetches(remoteFetchInfos, params, responseCallback, logReadResultMap, fetchPartitionStatus)
} else {
// If there is not enough data to respond and there is no remote data, we will let the fetch request
// wait for new data.
Expand All @@ -1733,7 +1733,7 @@ class ReplicaManager(val config: KafkaConfig,
)

// create a list of (topic, partition) pairs to use as keys for this delayed fetch operation
val delayedFetchKeys = fetchPartitionStatus.map { case (tp, _) => new TopicPartitionOperationKey(tp) }.toList
val delayedFetchKeys = fetchPartitionStatus.asScala.map { case (tp, _) => new TopicPartitionOperationKey(tp) }.toList
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we convert to a Scala List if we convert to a Java collection right after? We should probably use Java's Stream here and avoid the Scala collections altogether.


// 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
Expand Down
15 changes: 11 additions & 4 deletions core/src/test/scala/integration/kafka/server/DelayedFetchTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ import org.junit.jupiter.params.provider.ValueSource
import org.mockito.ArgumentMatchers.{any, anyInt}
import org.mockito.Mockito.{mock, when}

import java.util

class DelayedFetchTest {
private val maxBytes = 1024
private val replicaManager: ReplicaManager = mock(classOf[ReplicaManager])
Expand All @@ -59,7 +61,7 @@ class DelayedFetchTest {

val delayedFetch = new DelayedFetch(
params = fetchParams,
fetchPartitionStatus = Seq(topicIdPartition -> fetchStatus),
fetchPartitionStatus = createFetchPartitionStatusMap(topicIdPartition, fetchStatus),
replicaManager = replicaManager,
quota = replicaQuota,
responseCallback = callback
Expand Down Expand Up @@ -105,7 +107,7 @@ class DelayedFetchTest {

val delayedFetch = new DelayedFetch(
params = fetchParams,
fetchPartitionStatus = Seq(topicIdPartition -> fetchStatus),
fetchPartitionStatus = createFetchPartitionStatusMap(topicIdPartition, fetchStatus),
replicaManager = replicaManager,
quota = replicaQuota,
responseCallback = callback
Expand Down Expand Up @@ -145,7 +147,7 @@ class DelayedFetchTest {

val delayedFetch = new DelayedFetch(
params = fetchParams,
fetchPartitionStatus = Seq(topicIdPartition -> fetchStatus),
fetchPartitionStatus = createFetchPartitionStatusMap(topicIdPartition, fetchStatus),
replicaManager = replicaManager,
quota = replicaQuota,
responseCallback = callback
Expand Down Expand Up @@ -196,7 +198,7 @@ class DelayedFetchTest {

val delayedFetch = new DelayedFetch(
params = fetchParams,
fetchPartitionStatus = Seq(topicIdPartition -> fetchStatus),
fetchPartitionStatus = createFetchPartitionStatusMap(topicIdPartition, fetchStatus),
replicaManager = replicaManager,
quota = replicaQuota,
responseCallback = callback
Expand Down Expand Up @@ -267,4 +269,9 @@ class DelayedFetchTest {
error)
}

def createFetchPartitionStatusMap(tpId: TopicIdPartition, status: FetchPartitionStatus): util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus] = {
val map = new util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus]
map.put(tpId, status)
map
}
}
10 changes: 8 additions & 2 deletions core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ import org.mockito.ArgumentMatchers.{any, anyBoolean, anyInt, anyLong}
import org.mockito.Mockito.{mock, when}
import org.mockito.{AdditionalMatchers, ArgumentMatchers}

import java.util
import scala.jdk.CollectionConverters._

class ReplicaManagerQuotasTest {
Expand Down Expand Up @@ -186,7 +187,7 @@ class ReplicaManagerQuotasTest {

new DelayedFetch(
params = fetchParams,
fetchPartitionStatus = Seq(tp -> fetchPartitionStatus),
fetchPartitionStatus = createFetchPartitionStatusMap(tp, fetchPartitionStatus),
replicaManager = replicaManager,
quota = null,
responseCallback = null
Expand Down Expand Up @@ -237,7 +238,7 @@ class ReplicaManagerQuotasTest {

new DelayedFetch(
params = fetchParams,
fetchPartitionStatus = Seq(tidp -> fetchPartitionStatus),
fetchPartitionStatus = createFetchPartitionStatusMap(tidp, fetchPartitionStatus),
replicaManager = replicaManager,
quota = null,
responseCallback = null
Expand Down Expand Up @@ -341,4 +342,9 @@ class ReplicaManagerQuotasTest {
quota
}

def createFetchPartitionStatusMap(tpId: TopicIdPartition, status: FetchPartitionStatus): util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus] = {
val map = new util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus]
map.put(tpId, status)
map
}
}