Jackie-Jiang commented on code in PR #15641:
URL: https://github.com/apache/pinot/pull/15641#discussion_r2067522671
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1851,6 +1854,10 @@ private StreamPartitionMsgOffset
fetchStreamOffset(OffsetCriteria offsetCriteria
}
try {
return
_partitionMetadataProvider.fetchStreamPartitionOffset(offsetCriteria,
maxWaitTimeMs);
+ } catch (IllegalStateException ise) {
Review Comment:
This is not robust, and will ignore other illegal state exceptions. In what
scenario are we fetching offset after consumer is closed? Is there a better way
to handle it?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java:
##########
@@ -56,6 +56,8 @@ public abstract class KafkaPartitionLevelConnectionHandler {
protected final Consumer<String, Bytes> _consumer;
protected final TopicPartition _topicPartition;
protected final Properties _consumerProp;
+ // Admin client for thread-safe offset lookups
+ protected final AdminClient _adminClient;
Review Comment:
What is the overhead of creating the `AdminClient`? Should we create both
`Consumer` and `AdminClient` just when needed, and close immediately after use?
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
