Incremental-alter-configs-implementation(KIP-339) (#2005)

requires broker version >= 2.3.0

---------

Co-authored-by: Emanuele Sabellico <[email protected]>
Co-authored-by: Anchit Jain <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -1,5 +1,11 @@
 # 2.2.0
 
+## Enhancements
+
+- References librdkafka.redist 2.2.0. Refer to the [librdkafka v2.2.0 release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.2.0) for more information.
+- [KIP-339](https://cwiki.apache.org/confluence/display/KAFKA/KIP-339%3A+Create+a+new+IncrementalAlterConfigs+API)
+  IncrementalAlterConfigs API (#2005).
+
 ## Fixes
 
 - Fix backwards compatability of TopicPartitionOffset constructor. ([drinehimer](https://github.com/drinehimer), #2066)

examples/AdminClient/Program.cs

@@ -306,7 +306,8 @@ static async Task AlterConsumerGroupOffsetsAsync(string bootstrapServers, string
 
             var group = commandArgs[0];
             var tpoes = new List<TopicPartitionOffset>();
-            for (int i = 1; i + 2 < commandArgs.Length; i += 3) {
+            for (int i = 1; i + 2 < commandArgs.Length; i += 3)
+            {
                 try
                 {
                     var topic = commandArgs[i];
@@ -330,7 +331,8 @@ static async Task AlterConsumerGroupOffsetsAsync(string bootstrapServers, string
                 {
                     var results = await adminClient.AlterConsumerGroupOffsetsAsync(input);
                     Console.WriteLine("Successfully altered offsets:");
-                    foreach(var groupResult in results) {
+                    foreach(var groupResult in results)
+                    {
                         Console.WriteLine(groupResult);
                     }
 
@@ -363,7 +365,8 @@ static async Task ListConsumerGroupOffsetsAsync(string bootstrapServers, string[
 
             var group = commandArgs[0];
             var tpes = new List<TopicPartition>();
-            for (int i = 1; i + 1 < commandArgs.Length; i += 2) {
+            for (int i = 1; i + 1 < commandArgs.Length; i += 2)
+            {
                 try
                 {
                     var topic = commandArgs[i];
@@ -391,7 +394,8 @@ static async Task ListConsumerGroupOffsetsAsync(string bootstrapServers, string[
                 {
                     var result = await adminClient.ListConsumerGroupOffsetsAsync(input);
                     Console.WriteLine("Successfully listed offsets:");
-                    foreach(var groupResult in result) {
+                    foreach(var groupResult in result)
+                    {
                         Console.WriteLine(groupResult);
                     }
                 }
@@ -412,7 +416,8 @@ static async Task ListConsumerGroupOffsetsAsync(string bootstrapServers, string[
             }
         }
 
-        static async Task ListConsumerGroupsAsync(string bootstrapServers, string[] commandArgs) {
+        static async Task ListConsumerGroupsAsync(string bootstrapServers, string[] commandArgs)
+        {
             var timeout = TimeSpan.FromSeconds(30);
             var statesList = new List<ConsumerGroupState>();
             try
@@ -423,7 +428,8 @@ static async Task ListConsumerGroupsAsync(string bootstrapServers, string[] comm
                 }
                 if (commandArgs.Length > 1)
                 {
-                    for (int i = 1; i < commandArgs.Length; i++) {
+                    for (int i = 1; i < commandArgs.Length; i++)
+                    {
                         statesList.Add(Enum.Parse<ConsumerGroupState>(commandArgs[i]));
                     }
                 }
@@ -458,7 +464,8 @@ static async Task ListConsumerGroupsAsync(string bootstrapServers, string[] comm
         }
 
 
-        static async Task DescribeConsumerGroupsAsync(string bootstrapServers, string[] commandArgs) {
+        static async Task DescribeConsumerGroupsAsync(string bootstrapServers, string[] commandArgs)
+        {
             if (commandArgs.Length < 1)
             {
                 Console.WriteLine("usage: .. <bootstrapServers> describe-consumer-groups <group1> [<group2 ... <groupN>]");
@@ -501,6 +508,96 @@ static async Task DescribeConsumerGroupsAsync(string bootstrapServers, string[]
                 }
             }
         }
+        
+        static async Task IncrementalAlterConfigsAsync(string bootstrapServers, string[] commandArgs)
+        {
+            var timeout = TimeSpan.FromSeconds(30);
+            var configResourceList = new Dictionary<ConfigResource, List<ConfigEntry>>();
+            try
+            {
+                if (commandArgs.Length > 0)
+                {
+                    timeout = TimeSpan.FromSeconds(Int32.Parse(commandArgs[0]));
+                }
+                if (((commandArgs.Length - 1) % 3) != 0)
+                {
+                    throw new ArgumentException("invalid arguments length");
+                }
+                
+                for (int i = 1; i < commandArgs.Length; i+=3)
+                {
+                    var resourceType = Enum.Parse<ResourceType>(commandArgs[i]);
+                    var resourceName = commandArgs[i + 1];
+                    var configs = commandArgs[i + 2];
+                    var configList = new List<ConfigEntry>();
+                    foreach (var config in configs.Split(";"))
+                    {
+                        var nameOpValue = config.Split("=");
+                        if (nameOpValue.Length != 2)
+                        {
+                            throw new ArgumentException($"invalid alteration name \"{config}\"");
+                        }
+                        
+                        var name = nameOpValue[0];
+                        var opValue = nameOpValue[1].Split(":");
+                        if (opValue.Length != 2)
+                        {
+                            throw new ArgumentException($"invalid alteration value \"{nameOpValue[1]}\"");
+                        }
+                        
+                        var op = Enum.Parse<AlterConfigOpType>(opValue[0]);
+                        var value = opValue[1];
+                        configList.Add(new ConfigEntry
+                        {
+                            Name = name,
+                            Value = value,
+                            IncrementalOperation = op
+                        });
+                    }
+                    var resource = new ConfigResource
+                    {
+                        Name = resourceName,
+                        Type = resourceType
+                    };
+                    configResourceList[resource] = configList;
+                }
+            }
+            catch (Exception  e) when (
+                e is ArgumentException ||
+                e is FormatException
+            )
+            {
+                Console.WriteLine($"error: {e.Message}");
+                Console.WriteLine("usage: .. <bootstrapServers> incremental-alter-configs [<timeout_seconds> <resource-type1> <resource-name1> <config-name1=op-type1:config-value1;config-name1=op-type1:config-value1> ...]");
+                Environment.ExitCode = 1;
+                return;
+            }
+            
+            using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
+            {
+                try
+                {
+                    var alterResultList = await adminClient.IncrementalAlterConfigsAsync(configResourceList, new IncrementalAlterConfigsOptions() { RequestTimeout = timeout });
+                    foreach (var alterResult in alterResultList)
+                    {
+                        Console.WriteLine($"Resource {alterResult.ConfigResource} altered correctly");
+                    }
+                }
+                catch (IncrementalAlterConfigsException e)
+                {
+                    foreach (var alterResult in e.Results)
+                    {
+                        Console.WriteLine($"Resource {alterResult.ConfigResource} had error: {alterResult.Error}");
+                    }
+                    Environment.ExitCode = 1;
+                }
+                catch (Exception e)
+                {
+                    Console.WriteLine($"An error occurred altering configs incrementally: {e.Message}");
+                    Environment.ExitCode = 1;
+                }
+            }
+        }
 
         public static async Task Main(string[] args)
         {
@@ -511,7 +608,8 @@ public static async Task Main(string[] args)
                         "list-groups", "metadata", "library-version", "create-topic", "create-acls",
                         "describe-acls", "delete-acls",
                         "list-consumer-groups", "describe-consumer-groups",
-                        "list-consumer-group-offsets", "alter-consumer-group-offsets"
+                        "list-consumer-group-offsets", "alter-consumer-group-offsets",
+                        "incremental-alter-configs"
                     }) +
                     " ..");
                 Environment.ExitCode = 1;
@@ -557,6 +655,9 @@ public static async Task Main(string[] args)
                 case "describe-consumer-groups":
                     await DescribeConsumerGroupsAsync(bootstrapServers, commandArgs);
                     break;
+                case "incremental-alter-configs":
+                    await IncrementalAlterConfigsAsync(bootstrapServers, commandArgs);
+                    break;
                 default:
                     Console.WriteLine($"unknown command: {command}");
                     break;

src/Confluent.Kafka/Admin/AlterConfigOpType.cs

@@ -0,0 +1,45 @@
+// Copyright 2018 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>
+    ///     Enum of allowed AlterConfigOpType.
+    /// </summary>
+    public enum AlterConfigOpType : int
+    {
+        /// <summary>
+        ///     Sets/overwrites the configuration value.
+        /// </summary>
+        Set = 0,
+
+        /// <summary>
+        ///     Sets the configuration value to default or NULL.
+        /// </summary>
+        Delete = 1,
+
+        /// <summary>
+        ///     Appends the value to existing configuration values(only for list type values).
+        /// </summary>
+        Append = 2,
+
+        /// <summary>
+        ///     Subtracts the value from existing configuration values(only for list type values).
+        /// </summary>
+        Subtract = 3,
+    };
+}

src/Confluent.Kafka/Admin/ConfigEntry.cs

@@ -31,5 +31,10 @@ public class ConfigEntry
         ///     The config value.
         /// </summary>
         public string Value { get; set; }
+
+        /// <summary>
+        ///     Incremental operation to perform.
+        /// </summary>
+        public AlterConfigOpType IncrementalOperation { get; set; }
     }
 }

src/Confluent.Kafka/Admin/IncrementalAlterConfigsException.cs

@@ -0,0 +1,54 @@
+// Copyright 2023 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 an incremental alter configs request.
+    /// </summary>
+    public class IncrementalAlterConfigsException : KafkaException
+    {
+        /// <summary>
+        ///     Initializes a new instance of IncrementalAlterConfigsException.
+        /// </summary>
+        /// <param name="results">
+        ///     The result corresponding to all ConfigResources in the request 
+        ///     (whether or not they were in error). At least one of these
+        ///     results will be in error.
+        /// </param>
+        public IncrementalAlterConfigsException(List<IncrementalAlterConfigsReport> results)
+            : base(new Error(ErrorCode.Local_Partial,
+                "An error occurred incremental altering the following resources: [" +
+                String.Join(", ", results.Where(r => r.Error.IsError).Select(r => r.ConfigResource)) +
+                "]: [" + String.Join(", ", results.Where(r => r.Error.IsError).Select(r => r.Error)) + 
+                "]."))
+        {
+            Results = results;
+        }
+        
+        /// <summary>
+        ///     The result corresponding to all ConfigResources in the request 
+        ///     (whether or not they were in error). At least one of these
+        ///     results will be in error.
+        /// </summary>
+        public List<IncrementalAlterConfigsReport> Results { get; }
+    }
+}

src/Confluent.Kafka/Admin/IncrementalAlterConfigsOptions.cs

@@ -0,0 +1,45 @@
+// Copyright 2023 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 IncrementalAlterConfigs method.
+    /// </summary>
+    public class IncrementalAlterConfigsOptions
+    {
+        /// <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>
+        ///     If true, the request should be validated only without altering
+        ///     the configs.
+        /// 
+        ///     Default: false
+        /// </summary>
+        public bool ValidateOnly { get; set; } = false;
+    }
+}

src/Confluent.Kafka/Admin/IncrementalAlterConfigsReport.cs

@@ -0,0 +1,36 @@
+// Copyright 2023 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 incremental alter config request for a specific resource,
+    ///     when an error occurred.
+    /// </summary>
+    public class IncrementalAlterConfigsReport
+    {
+        /// <summary>
+        ///     The resource the result corresponds to.
+        /// </summary>
+        public ConfigResource ConfigResource;
+
+        /// <summary>
+        ///     The error (or success) of the incremental alter config request.
+        /// </summary>
+        public Error Error { get; set; }
+    }
+}