adithyachakilam commented on code in PR #17735:
URL: https://github.com/apache/druid/pull/17735#discussion_r1981330155
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -380,6 +382,108 @@ public LagStats computeLagStats()
return computeLags(partitionRecordLag);
}
+ /**
+ * This method is similar to updatePartitionLagFromStream
+ * but also determines time lag. Once this method has been
+ * tested, we can remove the older one.
+ */
+ private void updatePartitionTimeAndRecordLagFromStream()
+ {
+ final Map<KafkaTopicPartition, Long> highestCurrentOffsets =
getHighestCurrentOffsets();
+
+ getRecordSupplierLock().lock();
+ try {
+ Set<KafkaTopicPartition> partitionIds;
+ try {
+ partitionIds =
recordSupplier.getPartitionIds(getIoConfig().getStream());
+ }
+ catch (Exception e) {
+ log.warn("Could not fetch partitions for topic/stream [%s]",
getIoConfig().getStream());
+ throw new StreamException(e);
+ }
+
+ final Set<StreamPartition<KafkaTopicPartition>> partitions = partitionIds
+ .stream()
+ .map(e -> new StreamPartition<>(getIoConfig().getStream(), e))
+ .collect(Collectors.toSet());
+
+ final Set<KafkaTopicPartition> yetToReadPartitions = new HashSet<>();
+ for (Map.Entry<KafkaTopicPartition, Long> entry :
highestCurrentOffsets.entrySet()) {
+ if (partitionIds.contains(entry.getKey())) {
+ if (highestCurrentOffsets.get(entry.getKey()) == null ||
highestCurrentOffsets.get(entry.getKey()) == 0) {
+ yetToReadPartitions.add(entry.getKey());
+ continue;
+ }
+
+ recordSupplier.seek(new StreamPartition<>(getIoConfig().getStream(),
entry.getKey()), entry.getValue() - 1);
+ }
+ }
+
+ final Map<KafkaTopicPartition, Long> lastIngestedTimestamps =
getTimestampPerPartitionAtCurrentOffset(partitions);
+ // TODO: this might give wierd values for lag when the tasks are yet to
start processing
+ yetToReadPartitions.forEach(p -> lastIngestedTimestamps.put(p, 0L));
+
+ recordSupplier.seekToLatest(partitions);
+ // this method isn't actually computing the lag, just fetching the
latests offsets from the stream. This is
+ // because we currently only have record lag for kafka, which can be
lazily computed by subtracting the highest
+ // task offsets from the latest offsets from the stream when it is needed
+ latestSequenceFromStream =
recordSupplier.getLatestSequenceNumbers(partitions);
+
+ for (Map.Entry<KafkaTopicPartition, Long> entry :
latestSequenceFromStream.entrySet()) {
+ // if there are no messages .getEndOffset would return 0, but if there
are n msgs it would return n+1
+ // and hence we need to seek to n - 2 to get the nth msg in the next
poll.
+ if (entry.getValue() != 0) {
+ recordSupplier.seek(new StreamPartition<>(getIoConfig().getStream(),
entry.getKey()), entry.getValue() - 2);
+ }
+ }
+
+ partitionToTimeLag = getTimestampPerPartitionAtCurrentOffset(partitions)
+ .entrySet().stream().filter(e ->
lastIngestedTimestamps.containsKey(e.getKey()))
+ .collect(
+ Collectors.toMap(
+ Entry::getKey,
+ e -> e.getValue() - lastIngestedTimestamps.get(e.getKey())
+ )
+ );
Review Comment:
I don't this thats necessary. The poll that we do should make them
seektolatest.
--
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]
