Browse Source

KAFKA-4349; Handle 'PreparingRebalance' and 'AwaitingSync' states in consumer group describe

The edge case where consumer group state is `PreparingRebalance` or `AwaitingSync` will be separately handled as the group assignment is not yet determined.

Author: Vahid Hashemian <vahidhashemian@us.ibm.com>

Reviewers: Jason Gustafson <jason@confluent.io>

Closes #2070 from vahidhashemian/KAFKA-4349
pull/2070/merge
Vahid Hashemian 8 years ago committed by Jason Gustafson
parent
commit
34e9cc5dfa
  1. 10
      core/src/main/scala/kafka/admin/AdminClient.scala
  2. 31
      core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

10
core/src/main/scala/kafka/admin/AdminClient.scala

@ -149,9 +149,15 @@ class AdminClient(val time: Time, @@ -149,9 +149,15 @@ class AdminClient(val time: Time,
Errors.forCode(metadata.errorCode()).maybeThrow()
val consumers = metadata.members.map { consumer =>
val assignment = ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(Utils.readBytes(consumer.memberAssignment)))
ConsumerSummary(consumer.memberId, consumer.clientId, consumer.clientHost, assignment.partitions.toList)
ConsumerSummary(consumer.memberId, consumer.clientId, consumer.clientHost, metadata.state match {
case "Stable" =>
val assignment = ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(Utils.readBytes(consumer.memberAssignment)))
assignment.partitions.toList
case _ =>
List()
})
}.toList
ConsumerGroupSummary(metadata.state, metadata.protocol, Some(consumers), coordinator)
}

31
core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

@ -73,20 +73,23 @@ object ConsumerGroupCommand extends Logging { @@ -73,20 +73,23 @@ object ConsumerGroupCommand extends Logging {
case None =>
printError(s"The consumer group '$groupId' does not exist.")
case Some(assignments) =>
if (assignments.isEmpty)
state match {
case Some("Dead") =>
printError(s"Consumer group '$groupId' does not exist.")
case Some("Empty") =>
printError(s"Consumer group '$groupId' has no active members.")
case Some(_) =>
printError(s"Consumer group '$groupId' is rebalancing.")
case None =>
// the control should never reach here
throw new KafkaException("Expected a valid consumer group state, but none found.")
}
else
printAssignment(assignments, !opts.useOldConsumer)
state match {
case Some("Dead") =>
printError(s"Consumer group '$groupId' does not exist.")
case Some("Empty") =>
printError(s"Consumer group '$groupId' has no active members.")
case Some("PreparingRebalance") | Some("AwaitingSync") =>
System.err.println(s"Warning: Consumer group '$groupId' is rebalancing.")
printAssignment(assignments, !opts.useOldConsumer)
case Some("Stable") =>
printAssignment(assignments, !opts.useOldConsumer)
case Some(other) =>
// the control should never reach here
throw new KafkaException(s"Expected a valid consumer group state, but found '$other'.")
case None =>
// the control should never reach here
throw new KafkaException("Expected a valid consumer group state, but none found.")
}
}
}
else if (opts.options.has(opts.deleteOpt)) {

Loading…
Cancel
Save