KIP-229: DeleteGroups API

clients/src/main/java/org/apache/kafka/common/requests/DeleteGroupsRequest.java

@@ -93,7 +93,7 @@ protected Struct toStruct() {
     @Override
     public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         Errors error = Errors.forException(e);
-        Map<String, Errors> groupErrors = new HashMap<>();
+        Map<String, Errors> groupErrors = new HashMap<>(groups.size());
         for (String group : groups)
             groupErrors.put(group, error);
 
@@ -102,7 +102,7 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
                 return new DeleteGroupsResponse(throttleTimeMs, groupErrors);
             default:
                 throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
-                    version(), this.getClass().getSimpleName(), ApiKeys.DELETE_GROUPS.latestVersion()));
+                    version(), ApiKeys.DELETE_GROUPS.name, ApiKeys.DELETE_GROUPS.latestVersion()));
         }
     }
 

clients/src/main/java/org/apache/kafka/common/requests/DeleteGroupsResponse.java

@@ -52,7 +52,9 @@ public static Schema[] schemaVersions() {
     /**
      * Possible error codes:
      *
+     * COORDINATOR_LOAD_IN_PROGRESS (14)
      * COORDINATOR_NOT_AVAILABLE(15)
+     * NOT_COORDINATOR (16)
      * INVALID_GROUP_ID(24)
      * GROUP_AUTHORIZATION_FAILED(30)
      * NON_EMPTY_GROUP(68)

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

@@ -370,23 +370,38 @@ class AdminClient(val time: Time,
   }
 
   def deleteConsumerGroups(groups: List[String]): Map[String, Errors] = {
-    var errors: Map[String, Errors] = Map()
-    val groupsPerCoordinator = groups.map { group =>
+
+    def coordinatorLookup(group: String): Either[Node, Errors] = {
       try {
-        (group, findCoordinator(group))
+        Left(findCoordinator(group))
       } catch {
         case e: Throwable =>
-          errors += (group -> {
-            if (e.isInstanceOf[TimeoutException])
-              Errors.COORDINATOR_NOT_AVAILABLE
-            else
-              Errors.forException(e)
-          })
-          (group, null)
+          if (e.isInstanceOf[TimeoutException])
+            Right(Errors.COORDINATOR_NOT_AVAILABLE)
+          else
+            Right(Errors.forException(e))
       }
-    }.groupBy(_._2).map {
-      case (coordinator, groupCoordinator) => (coordinator, groupCoordinator.unzip._1.toArray)
-    }.filter(_._1 != null)
+    }
+
+    val groupCoordinator = groups.map(group => (group -> coordinatorLookup(group)))
+
+    var errors: Map[String, Errors] = Map()
+    var groupsPerCoordinator: Map[Node, List[String]] = Map()
+
+    groups.foreach { group =>
+      coordinatorLookup(group) match {
+        case Right(error) =>
+          errors += (group -> error)
+        case Left(coordinator) =>
+          groupsPerCoordinator.get(coordinator) match {
+            case Some(gList: List[String]) =>
+              val gListNew = group :: gList
+              groupsPerCoordinator += (coordinator -> gListNew)
+            case None =>
+              groupsPerCoordinator += (coordinator -> List(group))
+          }
+      }
+    }
 
     groupsPerCoordinator.foreach { case (coordinator, groups) =>
       val responseBody = send(coordinator, ApiKeys.DELETE_GROUPS, new DeleteGroupsRequest.Builder(groups.toSet.asJava))

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

@@ -429,9 +429,10 @@ object AdminUtils extends Logging with AdminUtilities {
    * @param topic Topic of the consumer group information we wish to delete
    */
   @deprecated("This method has been deprecated and will be removed in a future release.", "0.11.0.0")
-  def deleteAllConsumerGroupInfoForTopicInZK(zkUtils: ZkUtils, topic: String) {
+  def deleteAllConsumerGroupInfoForTopicInZK(zkUtils: ZkUtils, topic: String): Set[String] = {
     val groups = zkUtils.getAllConsumerGroupsForTopic(topic)
     groups.foreach(group => deleteConsumerGroupInfoForTopicInZK(zkUtils, group, topic))
+    groups
   }
 
   def topicExists(zkUtils: ZkUtils, topic: String): Boolean =

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

@@ -362,11 +362,11 @@ object ConsumerGroupCommand extends Logging {
 
     def deleteGroups(): Map[String, Errors] = {
       if (opts.options.has(opts.groupOpt) && opts.options.has(opts.topicOpt))
-        deleteForTopic()
+        deleteGroupsInfoForTopic()
       else if (opts.options.has(opts.groupOpt))
-        deleteForGroup()
+        deleteGroupsInfo()
       else if (opts.options.has(opts.topicOpt))
-        deleteAllForTopic()
+        deleteAllGroupsInfoForTopic()
 
       Map()
     }
@@ -474,45 +474,57 @@ object ConsumerGroupCommand extends Logging {
       }.toMap
     }
 
-    private def deleteForGroup() {
+    private def deleteGroupsInfo(): Map[String, Errors] = {
       val groups = opts.options.valuesOf(opts.groupOpt)
-      groups.asScala.foreach { group =>
+      groups.asScala.map { group =>
         try {
-          if (AdminUtils.deleteConsumerGroupInZK(zkUtils, group))
+          if (AdminUtils.deleteConsumerGroupInZK(zkUtils, group)) {
             println(s"Deleted all consumer group information for group '$group' in zookeeper.")
-          else
+            (group -> Errors.NONE)
+          }
+          else {
             printError(s"Delete for group '$group' failed because its consumers are still active.")
+            (group -> Errors.NON_EMPTY_GROUP)
+          }
         }
         catch {
           case e: ZkNoNodeException =>
             printError(s"Delete for group '$group' failed because group does not exist.", Some(e))
+            (group -> Errors.forException(e))
         }
-      }
+      }.toMap
     }
 
-    private def deleteForTopic() {
+    private def deleteGroupsInfoForTopic(): Map[String, Errors] = {
       val groups = opts.options.valuesOf(opts.groupOpt)
       val topic = opts.options.valueOf(opts.topicOpt)
       Topic.validate(topic)
-      groups.asScala.foreach { group =>
+      groups.asScala.map { group =>
         try {
-          if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkUtils, group, topic))
+          if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkUtils, group, topic)) {
             println(s"Deleted consumer group information for group '$group' topic '$topic' in zookeeper.")
-          else
+            (group -> Errors.NONE)
+          }
+          else {
             printError(s"Delete for group '$group' topic '$topic' failed because its consumers are still active.")
+            (group -> Errors.NON_EMPTY_GROUP)
+          }
         }
         catch {
           case e: ZkNoNodeException =>
             printError(s"Delete for group '$group' topic '$topic' failed because group does not exist.", Some(e))
+            (group -> Errors.forException(e))
         }
-      }
+      }.toMap
     }
 
-    private def deleteAllForTopic() {
+    private def deleteAllGroupsInfoForTopic(): Map[String, Errors] = {
       val topic = opts.options.valueOf(opts.topicOpt)
       Topic.validate(topic)
-      AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkUtils, topic)
+      val deletedGroups = AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkUtils, topic)
       println(s"Deleted consumer group information for all inactive consumer groups for topic '$topic' in zookeeper.")
+      deletedGroups.map((_, Errors.NONE)).toMap
+
     }
 
     private def getZkConsumer(brokerId: Int): Option[SimpleConsumer] = {
@@ -807,7 +819,7 @@ object ConsumerGroupCommand extends Logging {
       val groupsToDelete = opts.options.valuesOf(opts.groupOpt).asScala.toList
       val result = adminClient.deleteConsumerGroups(groupsToDelete)
       val successfullyDeleted = result.filter {
-        case (group, error) => error == Errors.NONE
+        case (_, error) => error == Errors.NONE
       }.keySet
 
       if (successfullyDeleted.size == result.size)

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

@@ -543,7 +543,7 @@ class GroupCoordinator(val brokerId: Int,
   }
 
   def handleDeletedPartitions(topicPartitions: Seq[TopicPartition]) {
-    groupManager.cleanupGroupMetadata(Some(topicPartitions))
+    groupManager.cleanupGroupMetadata(Some(topicPartitions), groupManager.currentGroups, time.milliseconds())
   }
 
   private def validateGroup(groupId: String): Option[Errors] = {

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -192,19 +192,22 @@ class GroupMetadataManager(brokerId: Int,
     var result: Map[String, Errors] = Map()
 
     groupIds.foreach { groupId =>
-      if (!groupMetadataCache.contains(groupId))
-        result += (groupId -> Errors.GROUP_ID_NOT_FOUND)
-      else {
-        val group = groupMetadataCache.get(groupId)
-        if (!group.is(Empty))
-          result += (groupId -> Errors.NON_EMPTY_GROUP)
-        else
-          eligibleGroups ++= Seq(group)
+      getGroup(groupId) match {
+        case None =>
+          result += (groupId -> Errors.GROUP_ID_NOT_FOUND)
+        case Some(group) =>
+          group.inLock {
+            if (!group.is(Empty))
+              result += (groupId -> Errors.NON_EMPTY_GROUP)
+            else {
+              eligibleGroups :+= group
+              cleanupGroupMetadata(None, Seq(group), Long.MaxValue)
+            }
+          }
       }
     }
 
     if (eligibleGroups.nonEmpty) {
-      cleanupGroupMetadata(None, eligibleGroups, Long.MaxValue)
       result ++= eligibleGroups.map(_.groupId -> Errors.NONE).toMap
       info(s"The following groups were deleted: ${eligibleGroups.map(_.groupId).mkString(", ")}")
     }
@@ -734,12 +737,12 @@ class GroupMetadataManager(brokerId: Int,
 
   // visible for testing
   private[group] def cleanupGroupMetadata(): Unit = {
-    cleanupGroupMetadata(None)
+    cleanupGroupMetadata(None, groupMetadataCache.values, time.milliseconds())
   }
 
   def cleanupGroupMetadata(deletedTopicPartitions: Option[Seq[TopicPartition]],
-                           groups: Iterable[GroupMetadata] = groupMetadataCache.values,
-                           startMs: Long = time.milliseconds()) {
+                           groups: Iterable[GroupMetadata],
+                           startMs: Long) {
     var offsetsRemoved = 0
 
     groups.foreach { group =>

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -1224,17 +1224,14 @@ class KafkaApis(val requestChannel: RequestChannel,
   def handleDeleteGroupsRequest(request: RequestChannel.Request): Unit = {
     val deleteGroupsRequest = request.body[DeleteGroupsRequest]
     var groups = deleteGroupsRequest.groups.asScala.toSet
-    var unauthorizedGroupsDeletionResult: Map[String, Errors] = Map()
 
-    groups.foreach { group =>
-      if (!authorize(request.session, Delete, new Resource(Group, group))) {
-        unauthorizedGroupsDeletionResult += (group -> Errors.GROUP_AUTHORIZATION_FAILED)
-        groups -= group
-      }
+    val (authorizedGroups, unauthorizedGroups) = groups.partition { group =>
+      authorize(request.session, Delete, new Resource(Group, group))
     }
 
-    val authorizedGroupsDeletionResult = groupCoordinator.handleDeleteGroups(groups)
-    val groupDeletionResult = authorizedGroupsDeletionResult._2 ++ unauthorizedGroupsDeletionResult
+    val groupDeletionResult = groupCoordinator.handleDeleteGroups(authorizedGroups)._2 ++
+      unauthorizedGroups.map((_ -> Errors.GROUP_AUTHORIZATION_FAILED))
+
     sendResponseMaybeThrottle(request, requestThrottleMs =>
       new DeleteGroupsResponse(requestThrottleMs, groupDeletionResult.asJava))
   }

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

@@ -1319,6 +1319,50 @@ class GroupCoordinatorTest extends JUnitSuite {
     assert(result.size == 1 && result.contains(groupId) && result.get(groupId).contains(Errors.NONE))
   }
 
+  @Test
+  def testDeleteEmptyGroupWithStoredOffsets() {
+    val memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID
+
+    val joinGroupResult = joinGroup(groupId, memberId, protocolType, protocols)
+    val assignedMemberId = joinGroupResult.memberId
+    val joinGroupError = joinGroupResult.error
+    assertEquals(Errors.NONE, joinGroupError)
+
+    EasyMock.reset(replicaManager)
+    val syncGroupResult = syncGroupLeader(groupId, joinGroupResult.generationId, assignedMemberId, Map(assignedMemberId -> Array[Byte]()))
+    val syncGroupError = syncGroupResult._2
+    assertEquals(Errors.NONE, syncGroupError)
+
+    EasyMock.reset(replicaManager)
+    val tp = new TopicPartition("topic", 0)
+    val offset = OffsetAndMetadata(0)
+    val commitOffsetResult = commitOffsets(groupId, assignedMemberId, joinGroupResult.generationId, immutable.Map(tp -> offset))
+    assertEquals(Errors.NONE, commitOffsetResult(tp))
+
+    val describeGroupResult = groupCoordinator.handleDescribeGroup(groupId)
+    assertEquals(Stable.toString, describeGroupResult._2.state)
+    assertEquals(assignedMemberId, describeGroupResult._2.members.head.memberId)
+
+    EasyMock.reset(replicaManager)
+    val leaveGroupResult = leaveGroup(groupId, assignedMemberId)
+    assertEquals(Errors.NONE, leaveGroupResult)
+
+    val groupTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupPartitionId)
+    val partition = EasyMock.niceMock(classOf[Partition])
+
+    EasyMock.reset(replicaManager)
+    EasyMock.expect(replicaManager.getMagic(EasyMock.anyObject())).andStubReturn(Some(RecordBatch.CURRENT_MAGIC_VALUE))
+    EasyMock.expect(replicaManager.getPartition(groupTopicPartition)).andStubReturn(Some(partition))
+    EasyMock.expect(replicaManager.nonOfflinePartition(groupTopicPartition)).andStubReturn(Some(partition))
+    EasyMock.replay(replicaManager, partition)
+
+    val (error, result) = groupCoordinator.handleDeleteGroups(Set(groupId).toSet)
+    assertEquals(Errors.NONE, error)
+    assert(result.size == 1 && result.contains(groupId) && result.get(groupId).contains(Errors.NONE))
+
+    assertEquals(Dead.toString, groupCoordinator.handleDescribeGroup(groupId)._2.state)
+  }
+
   @Test
   def shouldDelayInitialRebalanceByGroupInitialRebalanceDelayOnEmptyGroup() {
     val firstJoinFuture = sendJoinGroup(groupId, JoinGroupRequest.UNKNOWN_MEMBER_ID, protocolType, protocols)