Admin delete groups functionality (#1831)

* Add admin client functionality to delete consumer groups

* Corrected spelling

* Corrected line ending

* Resolved comments

* Fix file line ending

* Fix line ending

* Small test alignment and remove deletion of option pointer

* Fix copyright header

* Aligned test

* Avoid destroy pointer twice

* Fix comments

* Remove spaces

* Remove obsolete sleep

* Removed unused

* Change to plural

* Plural fixes

Co-authored-by: Matt Howlett <[email protected]>

Author: 3schwartz

src/Confluent.Kafka/Admin/DeleteGroupOptions.cs

@@ -0,0 +1,47 @@
+// Copyright 2022 Confluent Inc.
+//
+// Licensed 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.
+//
+// Refer to LICENSE for more information.
+
+using System;
+
+
+namespace Confluent.Kafka.Admin
+{
+    /// <summary>
+    ///     Options for the DeleteGroups method.
+    /// </summary>
+    public class DeleteGroupsOptions
+    {
+        /// <summary>
+        ///     The overall request timeout, including broker lookup, request 
+        ///     transmission, operation time on broker, and response. If set
+        ///     to null, the default request timeout for the AdminClient will
+        ///     be used.
+        /// 
+        ///     Default: null
+        /// </summary>
+        public TimeSpan? RequestTimeout { get; set; }
+
+        /// <summary>
+        ///     The broker's operation timeout - the maximum time to wait for
+        ///     DeleteRecordsAsync before returning a result to the application.
+        ///     If set to null, will return immediately upon triggering record
+        ///     deletion.
+        /// 
+        ///     Default: null
+        /// </summary>
+        public TimeSpan? OperationTimeout { get; set; }
+    }
+}

src/Confluent.Kafka/Admin/DeleteGroupReport.cs

@@ -0,0 +1,34 @@
+// Copyright 2022 Confluent Inc.
+//
+// Licensed 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.
+//
+// Refer to LICENSE for more information.
+
+namespace Confluent.Kafka.Admin
+{
+    /// <summary>
+    ///     The result of a request related to a group.
+    /// </summary>
+    public class DeleteGroupReport
+    {
+        /// <summary>
+        ///     The group.
+        /// </summary>
+        public string Group { get; set; }
+
+        /// <summary>
+        ///     The error (or success) of the group relevant for the request.
+        /// </summary>
+        public Error Error { get; set; }
+    }
+}

src/Confluent.Kafka/Admin/DeleteGroupsException.cs

@@ -0,0 +1,54 @@
+// Copyright 2022 Confluent Inc.
+//
+// Licensed 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.
+//
+// Refer to LICENSE for more information.
+
+using System;
+using System.Linq;
+using System.Collections.Generic;
+
+
+namespace Confluent.Kafka.Admin
+{
+    /// <summary>
+    ///     Represents an error that occured during a delete groups request.
+    /// </summary>
+    public class DeleteGroupsException : KafkaException
+    {
+        /// <summary>
+        ///     Initializes a new DeleteGroupsException.
+        /// </summary>
+        /// <param name="results">
+        ///     The result corresponding to all groups in the request 
+        ///     (whether or not they were in error). At least one of these
+        ///     results will be in error.
+        /// </param>
+        public DeleteGroupsException(List<DeleteGroupReport> results)
+            : base(new Error(ErrorCode.Local_Partial,
+                "An error occurred deleting groups: [" +
+                String.Join(", ", results.Where(r => r.Error.IsError).Select(r => r.Group)) +
+                "]: [" + String.Join(", ", results.Where(r => r.Error.IsError).Select(r => r.Error)) +
+                "]."))
+        {
+            Results = results;
+        }
+
+        /// <summary>
+        ///     The result corresponding to all groups in the request 
+        ///     (whether or not they were in error). At least one of these
+        ///     results will be in error.
+        /// </summary>
+        public List<DeleteGroupReport> Results { get; }
+    }
+}

src/Confluent.Kafka/AdminClient.cs

@@ -123,6 +123,20 @@ private List<DescribeConfigsReport> extractResultConfigs(IntPtr configResourcesP
             return result;
         }
 
+        private static List<DeleteGroupReport> extractDeleteGroupsReport(IntPtr eventPtr)
+        {
+            IntPtr groupsResultPtr = Librdkafka.DeleteGroups_result_groups(eventPtr, out UIntPtr resultCountPtr);
+            int groupsResultCount = (int)resultCountPtr;
+            IntPtr[] groupsResultPtrArr = new IntPtr[groupsResultCount];
+            Marshal.Copy(groupsResultPtr, groupsResultPtrArr, 0, groupsResultCount);
+
+            return groupsResultPtrArr.Select(groupResultPtr => new DeleteGroupReport
+            {
+                Group = PtrToStringUTF8(Librdkafka.group_result_name(groupResultPtr)),
+                Error = new Error(Librdkafka.group_result_error(groupResultPtr), false)
+            }).ToList();
+        }
+
         private List<CreateAclReport> extractCreateAclReports(IntPtr aclResultsPtr, int aclResultsCount)
         {
             IntPtr[] aclsResultsPtrArr = new IntPtr[aclResultsCount];
@@ -289,6 +303,32 @@ private Task StartPollTask(CancellationToken ct)
                                         }
                                         break;
 
+                                    case Librdkafka.EventType.DeleteGroups_Result:
+                                        {
+                                            if (errorCode != ErrorCode.NoError)
+                                            {
+                                                Task.Run(() =>
+                                                    ((TaskCompletionSource<List<DeleteGroupReport>>)adminClientResult).TrySetException(
+                                                        new KafkaException(kafkaHandle.CreatePossiblyFatalError(errorCode, errorStr))));
+                                                break;
+                                            }
+
+                                            var result = extractDeleteGroupsReport(eventPtr);
+                                            
+                                            if(result.Any(r => r.Error.IsError))
+                                            {
+                                                Task.Run(() =>
+                                                    ((TaskCompletionSource<List<DeleteGroupReport>>)adminClientResult).TrySetException(
+                                                        new DeleteGroupsException(result)));
+                                            }
+                                            else
+                                            {
+                                                Task.Run(() =>
+                                                    ((TaskCompletionSource<List<DeleteGroupReport>>)adminClientResult).TrySetResult(result));
+                                            }
+                                        }
+                                        break;
+
                                     case Librdkafka.EventType.CreatePartitions_Result:
                                         {
                                             if (errorCode != ErrorCode.NoError)
@@ -527,6 +567,7 @@ private Task StartPollTask(CancellationToken ct)
             { Librdkafka.EventType.AlterConfigs_Result, typeof(TaskCompletionSource<List<AlterConfigsReport>>) },
             { Librdkafka.EventType.CreatePartitions_Result, typeof(TaskCompletionSource<List<CreatePartitionsReport>>) },
             { Librdkafka.EventType.DeleteRecords_Result, typeof(TaskCompletionSource<List<DeleteRecordsResult>>) },
+            { Librdkafka.EventType.DeleteGroups_Result, typeof(TaskCompletionSource<List<DeleteGroupReport>>) },
             { Librdkafka.EventType.CreateAcls_Result, typeof(TaskCompletionSource<Null>) },
             { Librdkafka.EventType.DescribeAcls_Result, typeof(TaskCompletionSource<DescribeAclsResult>) },
             { Librdkafka.EventType.DeleteAcls_Result, typeof(TaskCompletionSource<List<DeleteAclsResult>>) },
@@ -603,6 +644,19 @@ public Task DeleteTopicsAsync(IEnumerable<string> topics, DeleteTopicsOptions op
             return completionSource.Task;
         }
 
+        /// <summary>
+        ///     Refer to <see cref="Confluent.Kafka.IAdminClient.DeleteGroupsAsync(IList{string}, DeleteGroupsOptions)" />
+        /// </summary>
+        public Task DeleteGroupsAsync(IList<string> groups, DeleteGroupsOptions options = null)
+        {
+            var completionSource = new TaskCompletionSource<List<DeleteGroupReport>>();
+            var gch = GCHandle.Alloc(completionSource);
+            Handle.LibrdkafkaHandle.DeleteGroups(
+                groups, options, resultQueue,
+                GCHandle.ToIntPtr(gch));
+            return completionSource.Task;
+        }
+
         /// <summary>
         ///     Refer to <see cref="Confluent.Kafka.IAdminClient.CreatePartitionsAsync(IEnumerable{PartitionsSpecification}, CreatePartitionsOptions)" />
         /// </summary>

src/Confluent.Kafka/IAdminClient.cs

@@ -102,6 +102,19 @@ public interface IAdminClient : IClient
         Task CreatePartitionsAsync(
             IEnumerable<PartitionsSpecification> partitionsSpecifications, CreatePartitionsOptions options = null);
 
+        /// <summary>
+        ///     Delete a set of groups.
+        /// </summary>
+        /// <param name="groups">
+        ///     The group names to delete.
+        /// </param>
+        /// <param name="options">
+        ///     The options to use when deleting groups.
+        /// </param>
+        /// <returns>
+        ///     The results of the delete group requests.
+        /// </returns>
+        Task DeleteGroupsAsync(IList<string> groups, DeleteGroupsOptions options = null);
 
         /// <summary>
         ///     Delete a set of topics. This operation is not

src/Confluent.Kafka/Impl/LibRdKafka.cs

@@ -59,9 +59,12 @@ internal enum AdminOp
             CreatePartitions=  3,
             AlterConfigs = 4,
             DescribeConfigs = 5,
+            DeleteRecords = 6,
+            DeleteGroups = 7,
+            DeleteConsumerGroupOffsets = 8,
             CreateAcls = 9,
             DescribeAcls = 10,
-            DeleteAcls = 11,
+            DeleteAcls = 11
         }
 
         public enum EventType : int
@@ -279,6 +282,12 @@ static bool SetDelegates(Type nativeMethodsClass)
             _DeleteTopics = (Action<IntPtr, IntPtr[], UIntPtr, IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_DeleteTopics").CreateDelegate(typeof(Action<IntPtr, IntPtr[], UIntPtr, IntPtr, IntPtr>));
             _DeleteTopics_result_topics = (_DeleteTopics_result_topics_delegate)methods.Single(m => m.Name == "rd_kafka_DeleteTopics_result_topics").CreateDelegate(typeof(_DeleteTopics_result_topics_delegate));
 
+            _DeleteGroup_new = (Func<string, IntPtr>)methods.Single(m => m.Name == "rd_kafka_DeleteGroup_new").CreateDelegate(typeof(Func<string, IntPtr>));
+            _DeleteGroup_destroy = (Action<IntPtr>)methods.Single(m => m.Name == "rd_kafka_DeleteGroup_destroy").CreateDelegate(typeof(Action<IntPtr>));
+
+            _DeleteGroups = (Action<IntPtr, IntPtr[], UIntPtr, IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_DeleteGroups").CreateDelegate(typeof(Action<IntPtr, IntPtr[], UIntPtr, IntPtr, IntPtr>));
+            _DeleteGroups_result_groups = (_DeleteGroups_result_groups_delegate)methods.Single(m => m.Name == "rd_kafka_DeleteGroups_result_groups").CreateDelegate(typeof(_DeleteGroups_result_groups_delegate));
+
             _DeleteRecords_new = (Func<IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_DeleteRecords_new").CreateDelegate(typeof(Func<IntPtr, IntPtr>));
             _DeleteRecords_destroy = (Action<IntPtr>)methods.Single(m => m.Name == "rd_kafka_DeleteRecords_destroy").CreateDelegate(typeof(Action<IntPtr>));
 
@@ -346,6 +355,9 @@ static bool SetDelegates(Type nativeMethodsClass)
             _topic_result_error_string = (Func<IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_topic_result_error_string").CreateDelegate(typeof(Func<IntPtr, IntPtr>));
             _topic_result_name = (Func<IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_topic_result_name").CreateDelegate(typeof(Func<IntPtr, IntPtr>));
 
+            _group_result_name = (Func<IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_group_result_name").CreateDelegate(typeof(Func<IntPtr, IntPtr>));
+            _group_result_error = (Func<IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_group_result_error").CreateDelegate(typeof(Func<IntPtr, IntPtr>));
+
             _destroy = (Action<IntPtr>)methods.Single(m => m.Name == "rd_kafka_destroy").CreateDelegate(typeof(Action<IntPtr>));
             _destroy_flags = (Action<IntPtr, IntPtr>)methods.Single(m => m.Name == "rd_kafka_destroy_flags").CreateDelegate(typeof(Action<IntPtr, IntPtr>));
 
@@ -1201,6 +1213,28 @@ out UIntPtr cntp
         ) => _DeleteTopics_result_topics(result, out cntp);
 
 
+        private static Func<string, IntPtr> _DeleteGroup_new;
+        internal static IntPtr DeleteGroup_new(
+            string group) => _DeleteGroup_new(group);
+
+        private static Action<IntPtr> _DeleteGroup_destroy;
+        internal static void DeleteGroup_destroy(IntPtr del_group) => _DeleteGroup_destroy(del_group);
+
+        private static Action<IntPtr, IntPtr[], UIntPtr, IntPtr, IntPtr> _DeleteGroups;
+        internal static void DeleteGroups(
+            IntPtr rk,
+            IntPtr[] del_groups,
+            UIntPtr del_groups_cnt,
+            IntPtr options,
+            IntPtr rkqu) => _DeleteGroups(rk, del_groups, del_groups_cnt, options, rkqu);
+
+        private delegate IntPtr _DeleteGroups_result_groups_delegate(IntPtr result, out UIntPtr cntp);
+        private static _DeleteGroups_result_groups_delegate _DeleteGroups_result_groups;
+        internal static IntPtr DeleteGroups_result_groups(
+            IntPtr result,
+            out UIntPtr cntp) => _DeleteGroups_result_groups(result, out cntp);
+
+
         private static Func<string, UIntPtr, StringBuilder, UIntPtr, IntPtr> _NewPartitions_new;
         internal static IntPtr NewPartitions_new(
                 string topic, 
@@ -1540,6 +1574,12 @@ out UIntPtr matchingAclsCntp
         private static Func<IntPtr, IntPtr> _topic_result_name;
         internal static IntPtr topic_result_name(IntPtr topicres) => _topic_result_name(topicres);
 
+        private static Func<IntPtr, IntPtr> _group_result_name;
+        internal static IntPtr group_result_name(IntPtr groupres) => _group_result_name(groupres);
+
+        private static Func<IntPtr, IntPtr> _group_result_error;
+        internal static IntPtr group_result_error(IntPtr groupres) => _group_result_error(groupres);
+
 
         //
         // Queues

src/Confluent.Kafka/Impl/NativeMethods/NativeMethods.cs

@@ -571,6 +571,30 @@ internal static extern IntPtr rd_kafka_DeleteTopics_result_topics(
         );
 
 
+        [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
+        internal static extern /* rd_kafka_DeleteGroup_t * */ IntPtr rd_kafka_DeleteGroup_new(
+                [MarshalAs(UnmanagedType.LPStr)] string group
+        );
+
+        [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
+        internal static extern void rd_kafka_DeleteGroup_destroy(
+                /* rd_kafka_DeleteGroup_t * */ IntPtr del_group);
+
+        [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
+        internal static extern void rd_kafka_DeleteGroups(
+                /* rd_kafka_t * */ IntPtr rk,
+                /* rd_kafka_DeleteGroup_t ** */ IntPtr[] del_groups,
+                UIntPtr del_group_cnt,
+                /* rd_kafka_AdminOptions_t * */ IntPtr options,
+                /* rd_kafka_queue_t * */ IntPtr rkqu);
+
+        [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
+        internal static extern IntPtr rd_kafka_DeleteGroups_result_groups(
+                /* rd_kafka_DeleteGroups_result_t * */ IntPtr result,
+                /* size_t * */ out UIntPtr cntp
+        );
+
+
         [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
         internal static extern /* rd_kafka_DeleteRecords_t * */ IntPtr rd_kafka_DeleteRecords_new(
                 /* rd_kafka_topic_partition_list_t * */ IntPtr offsets
@@ -862,6 +886,13 @@ internal static extern IntPtr rd_kafka_DeleteAcls_result_response_matching_acls(
         internal static extern IntPtr rd_kafka_topic_result_name(IntPtr topicres);
 
 
+        [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
+        internal static extern IntPtr rd_kafka_group_result_name(IntPtr groupres);
+
+        [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
+        internal static extern IntPtr rd_kafka_group_result_error(IntPtr groupres);
+
+
         //
         // Queues
         //