KAFKA-3158; ConsumerGroupCommand should tell whether group is actually dead

This patch fix differentiates between when a consumer group is rebalancing or dead and reports the appropriate error message.

Author: Ishita Mandhan <imandha@us.ibm.com>

Reviewers: Vahid Hashemian <vahidhashemian@us.ibm.com>, Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk>

Closes #1429 from imandhan/KAFKA-3158

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

@@ -143,21 +143,21 @@ class AdminClient(val time: Time,
                              clientHost: String,
                              assignment: List[TopicPartition])
 
-  def describeConsumerGroup(groupId: String): List[ConsumerSummary] = {
+  def describeConsumerGroup(groupId: String): Option[List[ConsumerSummary]] = {
     val group = describeGroup(groupId)
     if (group.state == "Dead")
-      return List.empty[ConsumerSummary]
+      return None
 
     if (group.protocolType != ConsumerProtocol.PROTOCOL_TYPE)
       throw new IllegalArgumentException(s"Group ${groupId} with protocol type '${group.protocolType}' is not a valid consumer group")
 
     if (group.state == "Stable") {
-      group.members.map { member =>
+      Some(group.members.map { member =>
         val assignment = ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(member.assignment))
         new ConsumerSummary(member.memberId, member.clientId, member.clientHost, assignment.partitions().asScala.toList)
-      }
+      })
     } else {
-      List.empty
+      Some(List.empty)
     }
   }
 

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

@@ -312,9 +312,11 @@ object ConsumerGroupCommand {
     }
 
     protected def describeGroup(group: String) {
-      val consumerSummaries = adminClient.describeConsumerGroup(group)
+      adminClient.describeConsumerGroup(group) match {
+        case None => println(s"Consumer group `${group}` does not exist.")
+        case Some(consumerSummaries) =>
           if (consumerSummaries.isEmpty)
-        println(s"Consumer group `${group}` does not exist or is rebalancing.")
+            println(s"Consumer group `${group}` is rebalancing.")
           else {
             val consumer = getConsumer()
             printDescribeHeader()
@@ -330,6 +332,7 @@ object ConsumerGroupCommand {
             }
           }
       }
+    }
 
     protected def getLogEndOffset(topic: String, partition: Int): LogEndOffsetResult = {
       val consumer = getConsumer()

core/src/test/scala/integration/kafka/api/AdminClientTest.scala

@@ -106,7 +106,7 @@ class AdminClientTest extends IntegrationTestHarness with Logging {
 
     val consumerSummaries = client.describeConsumerGroup(groupId)
     assertEquals(1, consumerSummaries.size)
-    assertEquals(Set(tp, tp2), consumerSummaries.head.assignment.toSet)
+    assertEquals(Some(Set(tp, tp2)), consumerSummaries.map(_.head.assignment.toSet))
   }
 
   @Test