shounakmk219 commented on code in PR #19116:
URL: https://github.com/apache/pinot/pull/19116#discussion_r3704412397
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -140,18 +139,35 @@ public List<PartitionGroupMetadata>
computePartitionGroupMetadata(String clientI
partitionIds = _partitionIdSubset;
}
- StreamConsumerFactory streamConsumerFactory =
StreamConsumerFactoryProvider.create(streamConfig);
+ // Partitions already covered by a consumption status reuse its offset;
the remaining partitions have their
+ // offsets fetched from the stream in a single batched call. Kafka's
beginningOffsets/endOffsets/offsetsForTimes
+ // accept a collection of partitions and resolve them in one broker
round-trip, so we avoid creating a fresh
+ // consumer per partition (previously hundreds of serial ~1s consumer
creations on high-partition tables, all
+ // executed inside the controller's ideal-state update lock).
+ List<Integer> partitionIdsToFetch = new ArrayList<>(partitionIds.size());
+ for (Integer partitionId : partitionIds) {
+ if (!partitionIdToEndOffset.containsKey(partitionId)) {
+ partitionIdsToFetch.add(partitionId);
+ }
+ }
+ Map<Integer, StreamPartitionMsgOffset> fetchedOffsets =
+ fetchOffsetsForPartitions(partitionIdsToFetch,
streamConfig.getOffsetCriteria(), timeoutMillis);
+
List<PartitionGroupMetadata> result = new ArrayList<>(partitionIds.size());
for (Integer partitionId : partitionIds) {
if (partitionIdToEndOffset.containsKey(partitionId)) {
result.add(new PartitionGroupMetadata(partitionId,
partitionIdToEndOffset.get(partitionId)));
} else {
- try (StreamMetadataProvider partitionMetadataProvider =
- streamConsumerFactory.createPartitionMetadataProvider(
- StreamConsumerFactory.getUniqueClientId(clientId),
partitionId)) {
- StreamPartitionMsgOffset startOffset =
partitionMetadataProvider.fetchStreamPartitionOffset(
- streamConfig.getOffsetCriteria(), timeoutMillis);
+ StreamPartitionMsgOffset startOffset = fetchedOffsets.get(partitionId);
+ if (startOffset != null) {
result.add(new PartitionGroupMetadata(partitionId, startOffset));
+ } else {
+ // The stream returned no offset for this partition (it does not
exist or has no data). Skip it gracefully
+ // so the remaining partitions are still processed; it is retried on
the next validation run. Note this is
+ // a new, more resilient behavior: the previous per-partition
implementation would fail the entire fetch
+ // here instead of dropping a single partition.
+ LOGGER.warn("No offset returned for topic: {} partition: {};
skipping it in partition group metadata",
Review Comment:
good catch, updated to throw TransientConsumerException
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
