[
https://issues.apache.org/jira/browse/KAFKA-14336?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17623342#comment-17623342
]
David Schlosnagle commented on KAFKA-14336:
-------------------------------------------
PR for review: https://github.com/apache/kafka/pull/12782
> Reduce allocations from MetadataResponse#convertToNodeArray
> -----------------------------------------------------------
>
> Key: KAFKA-14336
> URL: https://issues.apache.org/jira/browse/KAFKA-14336
> Project: Kafka
> Issue Type: Improvement
> Components: admin
> Reporter: David Schlosnagle
> Priority: Minor
>
> While profiling a Kafka consumer application that utilizes
> `org.apache.kafka.clients.admin.Admin#listOffsets(java.util.Map<org.apache.kafka.common.TopicPartition,org.apache.kafka.clients.admin.OffsetSpec>)`,
> one of the largest aggregate memory allocations are coming from
> `org.apache.kafka.common.requests.MetadataResponse.convertToNodeArray(List,
> Map)` due to the use of stream reference pipeline.
> We can avoid allocating the stream reference pipeline & spliterator for this
> case by explicitly allocating the presized `Node[]` and using a for loop with
> `int` induction over the specified IDs `List` argument.
> {code}
> java.util.stream.ReferencePipeline$3
> at java.util.stream.ReferencePipeline.map(Function)
> at org.apache.kafka.common.requests.MetadataResponse.convertToNodeArray(List,
> Map)
> at
> org.apache.kafka.common.requests.MetadataResponse.toPartitionInfo(MetadataResponse$PartitionMetadata,
> Map)
> at org.apache.kafka.common.requests.MetadataResponse.cluster()
> at
> org.apache.kafka.clients.admin.KafkaAdminClient.getListOffsetsCalls(MetadataOperationContext,
> Map, Map)
> at
> org.apache.kafka.clients.admin.KafkaAdminClient.lambda$listOffsets$21(MetadataOperationContext,
> Map, Map)
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$$Lambda$3539+0x000000080142d120.1364252822.get()
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$23.handleResponse(AbstractResponse)
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(long,
> List)
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests()
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run()
> at java.lang.Thread.run()
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
