diff --git a/core/src/main/scala/kafka/admin/AdminClient.scala b/core/src/main/scala/kafka/admin/AdminClient.scala index 22a8abb3043..11795575c28 100644 --- a/core/src/main/scala/kafka/admin/AdminClient.scala +++ b/core/src/main/scala/kafka/admin/AdminClient.scala @@ -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) } diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala index 6300d76cce2..a9cd6d3a685 100755 --- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala +++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala @@ -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)) {