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

Merged
merged 5 commits into from
Nov 2, 2025
Merged

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
07e5ae9
reduce collection conversions
DL1231 Oct 25, 2025
a2f25df
fix comment
DL1231 Oct 31, 2025
c6ef4e8
fix comment
DL1231 Oct 31, 2025
842735d
fix comment
DL1231 Nov 1, 2025
4d35d89
Merge remote-tracking branch 'origin/trunk' into reduce-collection-co…
DL1231 Nov 1, 2025
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

@chia7712 chia7712 Oct 25, 2025

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

@ijuma ijuma Oct 25, 2025

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).

Copy link
Contributor Author

@DL1231 DL1231 Oct 31, 2025

Choose a reason for hiding this comment

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

The source of fetchPartitionStatus comes from fetchData().

kafka/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java

Lines 395 to 420 in 53e1172

public Map<TopicIdPartition, PartitionData> fetchData(Map<Uuid, String> topicNames) {
final LinkedHashMap<TopicIdPartition, PartitionData> fetchData = new LinkedHashMap<>();
final short version = version();
data.topics().forEach(fetchTopic -> {
String name;
if (version < 13) {
name = fetchTopic.topic(); // can't be null
} else {
name = topicNames.get(fetchTopic.topicId());
}
fetchTopic.partitions().forEach(fetchPartition ->
// Topic name may be null here if the topic name was unable to be resolved using the topicNames map.
fetchData.put(new TopicIdPartition(fetchTopic.topicId(), new TopicPartition(name, fetchPartition.partition())),
new PartitionData(
fetchTopic.topicId(),
fetchPartition.fetchOffset(),
fetchPartition.logStartOffset(),
fetchPartition.partitionMaxBytes(),
optionalEpoch(fetchPartition.currentLeaderEpoch()),
optionalEpoch(fetchPartition.lastFetchedEpoch())
)
)
);
});
return fetchData;
}

Its type is LinkedHashMap<TopicIdPartition, PartitionData>, so there will be no duplicate TPs.


val logReadResults = replicaManager.readFromLog(
Copy link
Member

@chia7712 chia7712 Nov 1, 2025

Choose a reason for hiding this comment

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

Perhaps readFromLog could accept a util.Iterator instead of a Seq. This would help avoid creating an extra collection in this hot operation.

Copy link
Member

@ijuma ijuma Nov 1, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

Iterators can be exhausted and hence are a bit more brittle. I would only use them as a method parameter if there is a meaningful difference in performance.

Copy link
Member

@chia7712 chia7712 Nov 1, 2025

Choose a reason for hiding this comment

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

fair enough

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

@ijuma ijuma Oct 25, 2025

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.

Copy link
Contributor Author

@DL1231 DL1231 Oct 31, 2025

Choose a reason for hiding this comment

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

Thanks for catching this! I've fixed it. PTAL when you get a chance.


// 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
}
}