[AdminClient] Allow AdminClient to be authenticated (#1608)

Set token refresh mechanism

Co-authored-by: Jerome Viveret <[email protected]>

Author: jerive

src/Confluent.Kafka/AdminClient.cs

@@ -506,6 +506,7 @@ internal AdminClient(AdminClientBuilder builder)
             if (builder.LogHandler != null) { producerBuilder.SetLogHandler((_, logMessage) => builder.LogHandler(this, logMessage)); }
             if (builder.ErrorHandler != null) { producerBuilder.SetErrorHandler((_, error) => builder.ErrorHandler(this, error)); }
             if (builder.StatisticsHandler != null) { producerBuilder.SetStatisticsHandler((_, stats) => builder.StatisticsHandler(this, stats)); }
+            if (builder.OAuthBearerTokenRefreshHandler != null) { producerBuilder.SetOAuthBearerTokenRefreshHandler(builder.OAuthBearerTokenRefreshHandler); }
             this.ownedClient = producerBuilder.Build();
 
             this.handle = new Handle

src/Confluent.Kafka/AdminClientBuilder.cs

@@ -45,6 +45,11 @@ public class AdminClientBuilder
         /// </summary>
         internal protected Action<IAdminClient, string> StatisticsHandler { get; set; }
 
+        /// <summary>
+        ///     The configured OAuthBearer Token Refresh handler.
+        /// </summary>
+        public Action<IProducer<Null, Null>, string> OAuthBearerTokenRefreshHandler { get; set; }
+
         /// <summary>
         ///     Initialize a new <see cref="AdminClientBuilder" /> instance.
         /// </summary>
@@ -129,6 +134,39 @@ public AdminClientBuilder SetLogHandler(Action<IAdminClient, LogMessage> logHand
             return this;
         }
 
+        /// <summary>
+        ///     Set SASL/OAUTHBEARER token refresh callback in provided
+        ///     conf object. The SASL/OAUTHBEARER token refresh callback
+        ///     is triggered via <see cref="IAdminClient"/>'s admin methods
+        ///     (or any of its overloads) whenever OAUTHBEARER is the SASL
+        ///     mechanism and a token needs to be retrieved, typically
+        ///     based on the configuration defined in
+        ///     sasl.oauthbearer.config. The callback should invoke
+        ///     <see cref="ClientExtensions.OAuthBearerSetToken"/>
+        ///     or <see cref="ClientExtensions.OAuthBearerSetTokenFailure"/>
+        ///     to indicate success or failure, respectively.
+        ///
+        ///     An unsecured JWT refresh handler is provided by librdkafka
+        ///     for development and testing purposes, it is enabled by
+        ///     setting the enable.sasl.oauthbearer.unsecure.jwt property
+        ///     to true and is mutually exclusive to using a refresh callback.
+        /// </summary>
+        /// <param name="oAuthBearerTokenRefreshHandler">
+        ///     the callback to set; callback function arguments:
+        ///     IProducer - instance of the admin client's inner producer instance
+        ///     which should be used to set token or token failure string - Value of configuration
+        ///     property sasl.oauthbearer.config
+        /// </param>
+        public AdminClientBuilder SetOAuthBearerTokenRefreshHandler(Action<IProducer<Null, Null>, string> oAuthBearerTokenRefreshHandler)
+        {
+            if (this.OAuthBearerTokenRefreshHandler != null)
+            {
+                throw new InvalidOperationException("OAuthBearer token refresh handler may not be specified more than once.");
+            }
+            this.OAuthBearerTokenRefreshHandler = oAuthBearerTokenRefreshHandler;
+            return this;
+        }
+
         /// <summary>
         ///     Build the <see cref="AdminClient" /> instance.
         /// </summary>