You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "DL1231 (via GitHub)" <gi...@apache.org> on 2025/10/25 03:10:39 UTC
[PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
DL1231 opened a new pull request, #20768:
URL: https://github.com/apache/kafka/pull/20768
see https://github.com/apache/kafka/pull/19876#discussion_r2461128452
Initialize `fetchPartitionStatus` as a `Map` type to reduce unnecessary collection conversions.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2462673270
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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
Review Comment:
Is it possible to have duplicate TPs in this "fetch" path?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2479962860
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -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
Review Comment:
Thanks for catching this! I've fixed it. PTAL when you get a chance.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2480191795
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala:
##########
@@ -341,4 +342,9 @@ class ReplicaManagerQuotasTest {
quota
}
+ def createFetchPartitionStatusMap(tpId: TopicIdPartition, status: FetchPartitionStatus): util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus] = {
+ val map = new util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus]
Review Comment:
nit: rename variable `map` -> `statusMap` or something similar
##########
core/src/test/scala/integration/kafka/server/DelayedFetchTest.scala:
##########
@@ -267,4 +269,9 @@ class DelayedFetchTest {
error)
}
+ def createFetchPartitionStatusMap(tpId: TopicIdPartition, status: FetchPartitionStatus): util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus] = {
Review Comment:
ditto
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala:
##########
@@ -341,4 +342,9 @@ class ReplicaManagerQuotasTest {
quota
}
+ def createFetchPartitionStatusMap(tpId: TopicIdPartition, status: FetchPartitionStatus): util.LinkedHashMap[TopicIdPartition, FetchPartitionStatus] = {
Review Comment:
can the method access specifier be private?
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -66,8 +67,7 @@ class DelayedFetch(
*/
override def tryComplete(): Boolean = {
var accumulatedSize = 0
- fetchPartitionStatus.foreach {
- case (topicIdPartition, fetchStatus) =>
+ fetchPartitionStatus.forEach { (topicIdPartition, fetchStatus) =>
Review Comment:
nit: could you update the alignment from L71 - L136? remove one tab.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20768:
URL: https://github.com/apache/kafka/pull/20768#issuecomment-3473709912
@DL1231 please rebase code to include latest fixes
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2483854715
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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.iterator.map { case (tp, status) =>
tp -> status.fetchInfo
- }
+ }.toBuffer
val logReadResults = replicaManager.readFromLog(
Review Comment:
Perhaps `readFromLog` could accept a `util.Iterator` instead of a `Seq`. This would help avoid creating an extra collection in this hot operation.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2483898264
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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.iterator.map { case (tp, status) =>
tp -> status.fetchInfo
- }
+ }.toBuffer
val logReadResults = replicaManager.readFromLog(
Review Comment:
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.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2483902546
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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.iterator.map { case (tp, status) =>
tp -> status.fetchInfo
- }
+ }.toBuffer
val logReadResults = replicaManager.readFromLog(
Review Comment:
fair enough
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2483898264
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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.iterator.map { case (tp, status) =>
tp -> status.fetchInfo
- }
+ }.toBuffer
val logReadResults = replicaManager.readFromLog(
Review Comment:
Iterators can be exhausted and hence are a bit more brittle. I would only use them if there is a meaningful difference in performance.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2481213782
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -66,8 +67,7 @@ class DelayedFetch(
*/
override def tryComplete(): Boolean = {
var accumulatedSize = 0
- fetchPartitionStatus.foreach {
- case (topicIdPartition, fetchStatus) =>
+ fetchPartitionStatus.forEach { (topicIdPartition, fetchStatus) =>
Review Comment:
Done
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2479956474
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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
Review Comment:
The source of `fetchPartitionStatus` comes from `fetchData()`.
Its type is LinkedHashMap<TopicIdPartition, PartitionData>, so there will be no duplicate TPs.
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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
Review Comment:
The source of `fetchPartitionStatus` comes from `fetchData()`.
Its type is LinkedHashMap<TopicIdPartition, PartitionData>, so there will be no duplicate TPs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2479958965
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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
Review Comment:
The source of `fetchPartitionStatus` comes from `fetchData()`.
https://github.com/apache/kafka/blob/53e117253d21b2cd00e69d6ab91dae0ee84432f2/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java#L395-L420
Its type is `LinkedHashMap<TopicIdPartition, PartitionData>`, so there will be no duplicate TPs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20768:
URL: https://github.com/apache/kafka/pull/20768
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2482202135
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1733,12 +1734,15 @@ 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.keySet()
+ .stream()
+ .map(new TopicPartitionOperationKey(_))
+ .collect(Collectors.toList[TopicPartitionOperationKey]())
Review Comment:
I think you can use the `Stream.toList()` directly (no need for `collect`). This method was added to `Stream` in Java 16.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2462904655
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -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
Review Comment:
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).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #20768:
URL: https://github.com/apache/kafka/pull/20768#discussion_r2462900116
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -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
Review Comment:
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.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org