IgnacioAcunaF commented on a change in pull request #10858:
URL: https://github.com/apache/kafka/pull/10858#discussion_r652153471
##########
File path: core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
##########
@@ -560,14 +560,16 @@ object ConsumerGroupCommand extends Logging {
val groupOffsets = TreeMap[String, (Option[String],
Option[Seq[PartitionAssignmentState]])]() ++ (for ((groupId, consumerGroup) <-
consumerGroups) yield {
val state = consumerGroup.state
val committedOffsets = getCommittedOffsets(groupId)
+ // The admin client returns `null` as a value to indicate that there
is not committed offset for a partition. The following getPartitionOffset
function seeks to avoid NullPointerException by filtering out those null values.
+ def getPartitionOffset(tp: TopicPartition): Option[Long] =
committedOffsets.get(tp).filter(_ != null).map(_.offset)
var assignedTopicPartitions = ListBuffer[TopicPartition]()
val rowsWithConsumer =
consumerGroup.members.asScala.filter(!_.assignment.topicPartitions.isEmpty).toSeq
.sortWith(_.assignment.topicPartitions.size >
_.assignment.topicPartitions.size).flatMap { consumerSummary =>
val topicPartitions =
consumerSummary.assignment.topicPartitions.asScala
assignedTopicPartitions = assignedTopicPartitions ++ topicPartitions
val partitionOffsets =
consumerSummary.assignment.topicPartitions.asScala
.map { topicPartition =>
- topicPartition ->
committedOffsets.get(topicPartition).map(_.offset)
+ topicPartition -> getPartitionOffset(topicPartition)
Review comment:
Not directly. As `collectConsumerAssignment` expects for
getPartitionOffset parameter to be a `TopicPartition => Option[Long]`, and as
defined getPartitionOffset returns a Option[Long] for the specified
TopicPartition.
It can be modified to accept a sequence of TopicPartitions though, to
something like this:
```
def getPartitionOffsets(tp: Seq[TopicPartition]): TopicPartition =>
Option[Long] = tp.map { topicPartition => topicPartition ->
committedOffsets.get(topicPartition).filter(_ != null).map(_.offset)}.toMap
```
That way can be pass directly to collectConsumerAssignment, by changing:
```
val partitionOffsets = consumerSummary.assignment.topicPartitions.asScala
.map { topicPartition =>
topicPartition -> getPartitionOffset(topicPartition)
}.toMap
collectConsumerAssignment(groupId, Option(consumerGroup.coordinator),
topicPartitions.toList,
partitionOffsets, Some(s"${consumerSummary.consumerId}"),
Some(s"${consumerSummary.host}"),
Some(s"${consumerSummary.clientId}"))
```
To:
```
collectConsumerAssignment(groupId, Option(consumerGroup.coordinator),
topicPartitions.toList,
getPartitionOffsets(consumerSummary.assignment.topicPartitions.asScala.toSeq),
Some(s"${consumerSummary.consumerId}"), Some(s"${consumerSummary.host}"),
Some(s"${consumerSummary.clientId}"))
```
@dajac Do you think is a good approach?
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
