KIP-229: DeleteGroups API

clients/src/main/java/org/apache/kafka/common/errors/GroupIdNotFoundException.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class GroupIdNotFoundException extends ApiException {
+    private final String groupId;
+
+    public GroupIdNotFoundException(String groupId) {
+        super("The group id " + groupId + " was not found");
+        this.groupId = groupId;
+    }
+
+    public String groupId() {
+        return groupId;
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/errors/GroupNotEmptyException.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class GroupNotEmptyException extends ApiException {
+    private final String groupId;
+
+    public GroupNotEmptyException(String groupId) {
+        super("The group " + groupId + " is not empty");
+        this.groupId = groupId;
+    }
+
+    public String groupId() {
+        return groupId;
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -43,6 +43,8 @@
 import org.apache.kafka.common.requests.CreateTopicsResponse;
 import org.apache.kafka.common.requests.DeleteAclsRequest;
 import org.apache.kafka.common.requests.DeleteAclsResponse;
+import org.apache.kafka.common.requests.DeleteGroupsRequest;
+import org.apache.kafka.common.requests.DeleteGroupsResponse;
 import org.apache.kafka.common.requests.DeleteRecordsRequest;
 import org.apache.kafka.common.requests.DeleteRecordsResponse;
 import org.apache.kafka.common.requests.DeleteTopicsRequest;
@@ -183,7 +185,8 @@ public Struct parseResponse(short version, ByteBuffer buffer) {
     CREATE_DELEGATION_TOKEN(38, "CreateDelegationToken", CreateDelegationTokenRequest.schemaVersions(), CreateDelegationTokenResponse.schemaVersions()),
     RENEW_DELEGATION_TOKEN(39, "RenewDelegationToken", RenewDelegationTokenRequest.schemaVersions(), RenewDelegationTokenResponse.schemaVersions()),
     EXPIRE_DELEGATION_TOKEN(40, "ExpireDelegationToken", ExpireDelegationTokenRequest.schemaVersions(), ExpireDelegationTokenResponse.schemaVersions()),
-    DESCRIBE_DELEGATION_TOKEN(41, "DescribeDelegationToken", DescribeDelegationTokenRequest.schemaVersions(), DescribeDelegationTokenResponse.schemaVersions());
+    DESCRIBE_DELEGATION_TOKEN(41, "DescribeDelegationToken", DescribeDelegationTokenRequest.schemaVersions(), DescribeDelegationTokenResponse.schemaVersions()),
+    DELETE_GROUPS(42, "DeleteGroups", DeleteGroupsRequest.schemaVersions(), DeleteGroupsResponse.schemaVersions());
 
     private static final ApiKeys[] ID_TO_TYPE;
     private static final int MIN_API_KEY = 0;

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -17,21 +17,22 @@
 package org.apache.kafka.common.protocol;
 
 import org.apache.kafka.common.errors.ApiException;
-import org.apache.kafka.common.errors.InvalidPrincipalTypeException;
-import org.apache.kafka.common.errors.SaslAuthenticationException;
 import org.apache.kafka.common.errors.BrokerNotAvailableException;
 import org.apache.kafka.common.errors.ClusterAuthorizationException;
 import org.apache.kafka.common.errors.ConcurrentTransactionsException;
 import org.apache.kafka.common.errors.ControllerMovedException;
 import org.apache.kafka.common.errors.CoordinatorLoadInProgressException;
 import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
 import org.apache.kafka.common.errors.CorruptRecordException;
-import org.apache.kafka.common.errors.LogDirNotFoundException;
 import org.apache.kafka.common.errors.DuplicateSequenceException;
+import org.apache.kafka.common.errors.DelegationTokenAuthorizationException;
 import org.apache.kafka.common.errors.DelegationTokenDisabledException;
+import org.apache.kafka.common.errors.DelegationTokenExpiredException;
 import org.apache.kafka.common.errors.DelegationTokenNotFoundException;
 import org.apache.kafka.common.errors.DelegationTokenOwnerMismatchException;
 import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.apache.kafka.common.errors.IllegalGenerationException;
 import org.apache.kafka.common.errors.IllegalSaslStateException;
 import org.apache.kafka.common.errors.InconsistentGroupProtocolException;
@@ -41,6 +42,7 @@
 import org.apache.kafka.common.errors.InvalidGroupIdException;
 import org.apache.kafka.common.errors.InvalidPartitionsException;
 import org.apache.kafka.common.errors.InvalidPidMappingException;
+import org.apache.kafka.common.errors.InvalidPrincipalTypeException;
 import org.apache.kafka.common.errors.InvalidReplicaAssignmentException;
 import org.apache.kafka.common.errors.InvalidReplicationFactorException;
 import org.apache.kafka.common.errors.InvalidRequestException;
@@ -52,6 +54,7 @@
 import org.apache.kafka.common.errors.InvalidTxnTimeoutException;
 import org.apache.kafka.common.errors.KafkaStorageException;
 import org.apache.kafka.common.errors.LeaderNotAvailableException;
+import org.apache.kafka.common.errors.LogDirNotFoundException;
 import org.apache.kafka.common.errors.NetworkException;
 import org.apache.kafka.common.errors.NotControllerException;
 import org.apache.kafka.common.errors.NotCoordinatorException;
@@ -70,10 +73,9 @@
 import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.apache.kafka.common.errors.ReplicaNotAvailableException;
 import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.SaslAuthenticationException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
 import org.apache.kafka.common.errors.TimeoutException;
-import org.apache.kafka.common.errors.DelegationTokenAuthorizationException;
-import org.apache.kafka.common.errors.DelegationTokenExpiredException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.errors.TopicExistsException;
 import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
@@ -594,6 +596,18 @@ public ApiException build(String message) {
         public ApiException build(String message) {
             return new InvalidPrincipalTypeException(message);
         }
+    }),
+    NON_EMPTY_GROUP(68, "The group is not empty", new ApiExceptionBuilder() {
+        @Override
+        public ApiException build(String message) {
+            return new GroupNotEmptyException(message);
+        }
+    }),
+    GROUP_ID_NOT_FOUND(69, "The group id does not exist", new ApiExceptionBuilder() {
+        @Override
+        public ApiException build(String message) {
+            return new GroupIdNotFoundException(message);
+        }
     });
 
     private interface ApiExceptionBuilder {

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

@@ -222,6 +222,8 @@ public static AbstractRequest parseRequest(ApiKeys apiKey, short apiVersion, Str
                 return new ExpireDelegationTokenRequest(struct, apiVersion);
             case DESCRIBE_DELEGATION_TOKEN:
                 return new DescribeDelegationTokenRequest(struct, apiVersion);
+            case DELETE_GROUPS:
+                return new DeleteGroupsRequest(struct, apiVersion);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
                         "code should be updated to do so.", apiKey));

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

@@ -154,6 +154,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Struct struct) {
                 return new ExpireDelegationTokenResponse(struct);
             case DESCRIBE_DELEGATION_TOKEN:
                 return new DescribeDelegationTokenResponse(struct);
+            case DELETE_GROUPS:
+                return new DeleteGroupsResponse(struct);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                         "code should be updated to do so.", apiKey));

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

@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.ArrayOf;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.utils.Utils;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.kafka.common.protocol.types.Type.STRING;
+
+public class DeleteGroupsRequest extends AbstractRequest {
+    private static final String GROUPS_KEY_NAME = "groups";
+
+    /* DeleteGroups api */
+    private static final Schema DELETE_GROUPS_REQUEST_V0 = new Schema(
+            new Field(GROUPS_KEY_NAME, new ArrayOf(STRING), "An array of groups to be deleted."));
+
+    public static Schema[] schemaVersions() {
+        return new Schema[]{DELETE_GROUPS_REQUEST_V0};
+    }
+
+    private final Set<String> groups;
+
+    public static class Builder extends AbstractRequest.Builder<DeleteGroupsRequest> {
+        private final Set<String> groups;
+
+        public Builder(Set<String> groups) {
+            super(ApiKeys.DELETE_GROUPS);
+            this.groups = groups;
+        }
+
+        @Override
+        public DeleteGroupsRequest build(short version) {
+            return new DeleteGroupsRequest(groups, version);
+        }
+
+        @Override
+        public String toString() {
+            StringBuilder bld = new StringBuilder();
+            bld.append("(type=DeleteGroupsRequest").
+                append(", groups=(").append(Utils.join(groups, ", ")).append(")").
+                append(")");
+            return bld.toString();
+        }
+    }
+
+    private DeleteGroupsRequest(Set<String> groups, short version) {
+        super(version);
+        this.groups = groups;
+    }
+
+    public DeleteGroupsRequest(Struct struct, short version) {
+        super(version);
+        Object[] groupsArray = struct.getArray(GROUPS_KEY_NAME);
+        Set<String> groups = new HashSet<>(groupsArray.length);
+        for (Object group : groupsArray)
+            groups.add((String) group);
+
+        this.groups = groups;
+    }
+
+    @Override
+    protected Struct toStruct() {
+        Struct struct = new Struct(ApiKeys.DELETE_GROUPS.requestSchema(version()));
+        struct.set(GROUPS_KEY_NAME, groups.toArray());
+        return struct;
+    }
+
+    @Override
+    public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        Errors error = Errors.forException(e);
+        Map<String, Errors> groupErrors = new HashMap<>(groups.size());
+        for (String group : groups)
+            groupErrors.put(group, error);
+
+        switch (version()) {
+            case 0:
+                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(), ApiKeys.DELETE_GROUPS.name, ApiKeys.DELETE_GROUPS.latestVersion()));
+        }
+    }
+
+    public Set<String> groups() {
+        return groups;
+    }
+
+    public static DeleteGroupsRequest parse(ByteBuffer buffer, short version) {
+        return new DeleteGroupsRequest(ApiKeys.DELETE_GROUPS.parseRequest(version, buffer), version);
+    }
+
+}