GWphua commented on code in PR #18234:
URL: https://github.com/apache/druid/pull/18234#discussion_r2221401883
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -265,12 +266,18 @@ protected Map<KafkaTopicPartition, Long>
getPartitionRecordLag()
return null;
}
- if
(!latestSequenceFromStream.keySet().equals(highestCurrentOffsets.keySet())) {
- log.warn(
- "Kafka partitions[%s] do not match task partitions[%s]",
- latestSequenceFromStream.keySet(),
- highestCurrentOffsets.keySet()
+ Set<KafkaTopicPartition> kafkaPartitions =
latestSequenceFromStream.keySet();
+ Set<KafkaTopicPartition> taskPartitions = highestCurrentOffsets.keySet();
+ if (!kafkaPartitions.equals(taskPartitions)) {
+ List<KafkaTopicPartition> missingTaskPartitions = new ArrayList<>(
+ Sets.difference(kafkaPartitions, taskPartitions)
);
+ List<KafkaTopicPartition> missingKafkaPartitions = new ArrayList<>(
+ Sets.difference(taskPartitions, kafkaPartitions)
+ );
+
+ log.warn("Mismatched kafka and task partitions: Missing Task Partitions
%s, Missing Kafka Partitions %s",
Review Comment:
Hi @cecemei I am thinking how to handle the `JsonProcessingException`. Since
this is an unchecked exception, it must be caught or the compiler wont be happy
about it. Do you have any suggestions on this?
I am hesitating from adding exception to the signature
`getPartitionRecordLag()` since it
1. Inherits from the `SeekableStreamSupervisor`, we will also need to change
method signature in that class.
2. We will need to handle the `JsonProcessingException` at places that call
it (and possible change method signature there too)
3. It is unlikely to throw the `JsonProcessingException` exception in the
first place, given the presence of toString() methods.
Currently, I use the DruidException.defensive(), whose JavaDocs state it
being "an exception that should never actually be triggered". I feel this may
be better than using an unreachable log, or totally ignore the exception, which
may confuse future readers.
--
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]
