feat: add oauth

feat: add oauth

src/KafkaFlow.Abstractions/Configuration/SaslMechanism.cs

@@ -14,4 +14,7 @@ public enum SaslMechanism
 
     /// <summary>SCRAM-SHA-512</summary>
     ScramSha512,
+
+    /// <summary>OAUTH-BEARER</summary>
+    OAuthBearer,
 }

src/KafkaFlow.Abstractions/Configuration/SaslOauthbearerMethod.cs

@@ -0,0 +1,12 @@
+namespace KafkaFlow.Configuration
+{
+    /// <summary>SaslOauthbearerMethod enum values</summary>
+    public enum SaslOauthbearerMethod
+    {
+        /// <summary>Default</summary>
+        Default,
+
+        /// <summary>Oidc</summary>
+        Oidc,
+    }
+}

src/KafkaFlow.Abstractions/Configuration/SecurityInformation.cs

@@ -209,4 +209,44 @@ public class SecurityInformation
     ///     importance: low
     /// </summary>
     public bool? EnableSaslOauthbearerUnsecureJwt { get; set; }
-}
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerMethod
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public SaslOauthbearerMethod? SaslOauthbearerMethod { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerClientId
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerClientId { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerClientSecret
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerClientSecret { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerTokenEndpointUrl
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerTokenEndpointUrl { get; set; }
+
+    /// <summary>
+    ///     Gets or sets the SaslOauthbearerScope
+    ///
+    ///     default: false
+    ///     importance: low
+    /// </summary>
+    public string SaslOauthbearerScope { get; set; }
+}

src/KafkaFlow/Extensions/ConfigurationExtensions.cs

@@ -2,6 +2,7 @@
 using Confluent.Kafka;
 using KafkaFlow.Configuration;
 using SaslMechanism = KafkaFlow.Configuration.SaslMechanism;
+using SaslOauthbearerMethod = KafkaFlow.Configuration.SaslOauthbearerMethod;
 using SecurityProtocol = KafkaFlow.Configuration.SecurityProtocol;
 using SslEndpointIdentificationAlgorithm = KafkaFlow.Configuration.SslEndpointIdentificationAlgorithm;
 
@@ -15,6 +16,14 @@ internal static class ConfigurationExtensions
         SaslMechanism.Plain => Confluent.Kafka.SaslMechanism.Plain,
         SaslMechanism.ScramSha256 => Confluent.Kafka.SaslMechanism.ScramSha256,
         SaslMechanism.ScramSha512 => Confluent.Kafka.SaslMechanism.ScramSha512,
+        SaslMechanism.OAuthBearer => Confluent.Kafka.SaslMechanism.OAuthBearer,
+        _ => throw new ArgumentOutOfRangeException()
+    };
+
+    public static Confluent.Kafka.SaslOauthbearerMethod ToConfluent(this SaslOauthbearerMethod method) => method switch
+    {
+        SaslOauthbearerMethod.Default => Confluent.Kafka.SaslOauthbearerMethod.Default,
+        SaslOauthbearerMethod.Oidc => Confluent.Kafka.SaslOauthbearerMethod.Oidc,
         _ => throw new ArgumentOutOfRangeException()
     };
 
@@ -70,5 +79,10 @@ public static void ReadSecurityInformationFrom(this ClientConfig config, Cluster
         config.SaslMechanism = securityInformation.SaslMechanism?.ToConfluent();
         config.SaslUsername = securityInformation.SaslUsername;
         config.SaslPassword = securityInformation.SaslPassword;
+        config.SaslOauthbearerMethod = securityInformation.SaslOauthbearerMethod?.ToConfluent();
+        config.SaslOauthbearerClientId = securityInformation.SaslOauthbearerClientId;
+        config.SaslOauthbearerClientSecret = securityInformation.SaslOauthbearerClientSecret;
+        config.SaslOauthbearerTokenEndpointUrl = securityInformation.SaslOauthbearerTokenEndpointUrl;
+        config.SaslOauthbearerScope = securityInformation.SaslOauthbearerScope;
     }
-}
+}