apache / apache/pinot

optimize kafka `computePartitionGroupMetadata` with KafkaAdminClient

Open
#11,001 1 comment 0 reactions 1 assignee Claimed by @navina View on GitHub
Dominant language
Java
Stars
6.1k
Forks
1.5k
Avg merge
2d 55m
Merged PRs (30d)
182

Description

we recently had some dns instability issues w/ our kafka cluster that caused consumer creation to fail since it couldn't resolve broker dns. while investigating, we noticed that we were creating N consumers every time a new realtime segment was created where N is the number of partitions on the topic. we have some topics w/ a high partition count like 200.

```
[2023-06-27 02:50:30.459797] INFO [KafkaConsumer] [HelixTaskExecutor-message_handle_thread_27:17] [Consumer clientId=example_table-example_topic-5, groupId=pinot-table-group1] Subscribed to partition(s): example_topic-100
[2023-06-27 02:50:30.459840] INFO [KafkaConsumer] [HelixTaskExecutor-message_handle_thread_27:17] [Consumer clientId=example_table-example_topic-5, groupId=pinot-table-group1] Subscribed to partition(s): example_topic-132
...
```

it seems like the default implementation of [`computePartitionGroupMetadata`](https://github.com/apache/pinot/blob/master/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java#L65) creates all these consumers. i'm wondering if there are any blockers to using the `KafkaAdminClient`'s listOffsets call to achieve the same thing: https://kafka.apache.org/28/javadoc/org/apache/kafka/clients/admin/KafkaAdminClient.html#listOffsets(java.util.Map,org.apache.kafka.clients.admin.ListOffsetsOptions)

here is where it's getting called for new realtime segment creation: https://github.com/apache/pinot/blob/fc26d6d8975b4cd46e26e460236a30e8b1eb2cde/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java#L1547-L1549

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.