noob-se7en commented on code in PR #19116:
URL: https://github.com/apache/pinot/pull/19116#discussion_r3705328297
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -183,44 +199,75 @@ public Map<Integer, StreamPartitionMsgOffset>
fetchLatestStreamOffset(Set<Intege
@Override
public StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria
offsetCriteria, long timeoutMillis) {
Preconditions.checkNotNull(offsetCriteria);
- long offset;
+ StreamPartitionMsgOffset offset =
+ fetchOffsetsForPartitions(List.of(_partition), offsetCriteria,
timeoutMillis).get(_partition);
+ if (offset == null) {
+ throw new TransientConsumerException(new RuntimeException(
+ "Failed to fetch offset for topic: " + _topic + " partition: " +
_partition));
+ }
+ return offset;
+ }
+
+ /**
+ * Fetches the offset matching {@code offsetCriteria} for the given
partitions in a single batched call to the
+ * stream. Kafka's {@code beginningOffsets}/{@code endOffsets}/{@code
offsetsForTimes} all accept a collection of
+ * partitions, so this issues one broker round-trip regardless of the number
of partitions (these calls do not
+ * require the consumer to be assigned to the partitions). A partition that
the stream does not return an offset
+ * for is omitted from the result map.
+ */
+ private Map<Integer, StreamPartitionMsgOffset>
fetchOffsetsForPartitions(Collection<Integer> partitionIds,
+ OffsetCriteria offsetCriteria, long timeoutMillis) {
+ Preconditions.checkNotNull(offsetCriteria);
+ if (partitionIds.isEmpty()) {
+ return Map.of();
+ }
+ List<TopicPartition> topicPartitions = new
ArrayList<>(partitionIds.size());
+ for (Integer partitionId : partitionIds) {
+ topicPartitions.add(new TopicPartition(_topic, partitionId));
+ }
+ Duration timeout = Duration.ofMillis(timeoutMillis);
try {
+ Map<TopicPartition, Long> topicPartitionToOffset;
if (offsetCriteria.isLargest()) {
- offset = _consumer.endOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
+ topicPartitionToOffset = _consumer.endOffsets(topicPartitions,
timeout);
} else if (offsetCriteria.isSmallest()) {
- offset =
- _consumer.beginningOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- } else if (offsetCriteria.isPeriod()) {
- OffsetAndTimestamp offsetAndTimestamp =
_consumer.offsetsForTimes(Map.of(_topicPartition,
- Clock.systemUTC().millis() -
TimeUtils.convertPeriodToMillis(offsetCriteria.getOffsetString())))
- .get(_topicPartition);
- if (offsetAndTimestamp == null) {
- offset = _consumer.endOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- LOGGER.warn(
- "initial offset type is period and its value evaluates to null
hence proceeding with offset {} for "
- + "topic {} partition {}", offset, _topicPartition.topic(),
_topicPartition.partition());
- } else {
- offset = offsetAndTimestamp.offset();
+ topicPartitionToOffset = _consumer.beginningOffsets(topicPartitions,
timeout);
+ } else if (offsetCriteria.isPeriod() || offsetCriteria.isTimestamp()) {
+ long timestampMillis = offsetCriteria.isPeriod()
+ ? Clock.systemUTC().millis() -
TimeUtils.convertPeriodToMillis(offsetCriteria.getOffsetString())
+ :
TimeUtils.convertTimestampToMillis(offsetCriteria.getOffsetString());
+ Map<TopicPartition, Long> timestampToSearch = new
HashMap<>(topicPartitions.size());
+ for (TopicPartition topicPartition : topicPartitions) {
+ timestampToSearch.put(topicPartition, timestampMillis);
}
- } else if (offsetCriteria.isTimestamp()) {
- OffsetAndTimestamp offsetAndTimestamp =
_consumer.offsetsForTimes(Map.of(_topicPartition,
-
TimeUtils.convertTimestampToMillis(offsetCriteria.getOffsetString()))).get(_topicPartition);
- if (offsetAndTimestamp == null) {
- offset = _consumer.endOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- LOGGER.warn(
- "initial offset type is timestamp and its value evaluates to
null hence proceeding with offset {} for "
- + "topic {} partition {}", offset, _topicPartition.topic(),
_topicPartition.partition());
- } else {
- offset = offsetAndTimestamp.offset();
+ Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes =
_consumer.offsetsForTimes(timestampToSearch, timeout);
Review Comment:
the new note cites `METADATA_FETCH_TIMEOUT_MS` (15s), but
`StatelessRealtimeSegmentWriter:425` also calls `computePartitionGroupMetadata`
with a 5s timeout and `List.of()` statuses, so every partition goes into one
`offsetsForTimes` there and the worst-case narrowing is 60s to 5s. that path
falls back to the config `numPartitions` on timeout, which is the stale divisor
the stream-count lookup exists to avoid, so worth naming that caller in the
note too.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -183,44 +189,83 @@ public Map<Integer, StreamPartitionMsgOffset>
fetchLatestStreamOffset(Set<Intege
@Override
public StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria
offsetCriteria, long timeoutMillis) {
Preconditions.checkNotNull(offsetCriteria);
- long offset;
+ // fetchOffsetsForPartitions throws if the stream returns no offset for
_partition, so the result is non-null.
+ return fetchOffsetsForPartitions(List.of(_partition), offsetCriteria,
timeoutMillis).get(_partition);
+ }
+
+ /**
+ * Fetches the offset matching {@code offsetCriteria} for the given
partitions in a single batched call to the
+ * stream. Kafka's {@code beginningOffsets}/{@code endOffsets}/{@code
offsetsForTimes} all accept a collection of
+ * partitions, so this issues one broker round-trip regardless of the number
of partitions (these calls do not
+ * require the consumer to be assigned to the partitions).
+ *
+ * @return an offset for every requested partition
+ * @throws TransientConsumerException if the stream returns no offset for a
requested partition (so the caller
+ * retries rather than treating the partition as absent)
+ */
+ private Map<Integer, StreamPartitionMsgOffset>
fetchOffsetsForPartitions(Collection<Integer> partitionIds,
+ OffsetCriteria offsetCriteria, long timeoutMillis) {
+ Preconditions.checkNotNull(offsetCriteria);
+ if (partitionIds.isEmpty()) {
+ return Map.of();
+ }
+ List<TopicPartition> topicPartitions = new
ArrayList<>(partitionIds.size());
+ for (Integer partitionId : partitionIds) {
+ topicPartitions.add(new TopicPartition(_topic, partitionId));
+ }
+ Duration timeout = Duration.ofMillis(timeoutMillis);
try {
+ Map<TopicPartition, Long> topicPartitionToOffset;
if (offsetCriteria.isLargest()) {
- offset = _consumer.endOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
+ topicPartitionToOffset = _consumer.endOffsets(topicPartitions,
timeout);
} else if (offsetCriteria.isSmallest()) {
- offset =
- _consumer.beginningOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- } else if (offsetCriteria.isPeriod()) {
- OffsetAndTimestamp offsetAndTimestamp =
_consumer.offsetsForTimes(Map.of(_topicPartition,
- Clock.systemUTC().millis() -
TimeUtils.convertPeriodToMillis(offsetCriteria.getOffsetString())))
- .get(_topicPartition);
- if (offsetAndTimestamp == null) {
- offset = _consumer.endOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- LOGGER.warn(
- "initial offset type is period and its value evaluates to null
hence proceeding with offset {} for "
- + "topic {} partition {}", offset, _topicPartition.topic(),
_topicPartition.partition());
- } else {
- offset = offsetAndTimestamp.offset();
+ topicPartitionToOffset = _consumer.beginningOffsets(topicPartitions,
timeout);
+ } else if (offsetCriteria.isPeriod() || offsetCriteria.isTimestamp()) {
+ long timestampMillis = offsetCriteria.isPeriod()
+ ? Clock.systemUTC().millis() -
TimeUtils.convertPeriodToMillis(offsetCriteria.getOffsetString())
+ :
TimeUtils.convertTimestampToMillis(offsetCriteria.getOffsetString());
+ Map<TopicPartition, Long> timestampToSearch = new
HashMap<>(topicPartitions.size());
+ for (TopicPartition topicPartition : topicPartitions) {
+ timestampToSearch.put(topicPartition, timestampMillis);
}
- } else if (offsetCriteria.isTimestamp()) {
- OffsetAndTimestamp offsetAndTimestamp =
_consumer.offsetsForTimes(Map.of(_topicPartition,
-
TimeUtils.convertTimestampToMillis(offsetCriteria.getOffsetString()))).get(_topicPartition);
- if (offsetAndTimestamp == null) {
- offset = _consumer.endOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- LOGGER.warn(
- "initial offset type is timestamp and its value evaluates to
null hence proceeding with offset {} for "
- + "topic {} partition {}", offset, _topicPartition.topic(),
_topicPartition.partition());
- } else {
- offset = offsetAndTimestamp.offset();
+ Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes =
_consumer.offsetsForTimes(timestampToSearch, timeout);
Review Comment:
the note in the description cites `METADATA_FETCH_TIMEOUT_MS` (15s), but
`StatelessRealtimeSegmentWriter:425` also calls `computePartitionGroupMetadata`
with a 5s timeout and `List.of()` statuses, so every partition goes into one
`offsetsForTimes` there and the worst-case narrowing is 60s to 5s. that path
falls back to the config `numPartitions` on timeout, which is the stale divisor
the stream-count lookup exists to avoid, so worth naming that caller in the
note too.
--
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]
