Extended MirrordKafkaClientConfig CRD with SASL OAUTH token provider

changelog.d/+kafka-sasl-oauth-token-provider.added.md

@@ -0,0 +1 @@
+Extended `MirrordKafkaClientConfig` CRD with a kind of a SASL OAUTH token provider to use.

mirrord/operator/src/crd/kafka.rs

@@ -11,7 +11,8 @@ use serde::{Deserialize, Serialize};
     kind = "MirrordKafkaClientConfig",
     namespaced,
     printcolumn = r#"{"name":"PARENT", "type":"string", "description":"Name of parent configuration.", "jsonPath":".spec.parent"}"#,
-    printcolumn = r#"{"name":"SECRET", "type":"string", "description":"Name of Secret to load from.", "jsonPath":".spec.loadFromSecret"}"#
+    printcolumn = r#"{"name":"SECRET", "type":"string", "description":"Name of Secret to load from.", "jsonPath":".spec.loadFromSecret"}"#,
+    printcolumn = r#"{"name":"AUTHENTICATION_EXTRA", "type":"string", "description":"Additional authentication config.", "jsonPath":".spec.authenticationExtra"}"#
 )]
 #[serde(rename_all = "camelCase")]
 pub struct MirrordKafkaClientConfigSpec {
@@ -33,6 +34,9 @@ pub struct MirrordKafkaClientConfigSpec {
     ///
     /// Example value: `default/my-secret`
     pub load_from_secret: Option<String>,
+
+    /// Additional authentication configuration.
+    pub authentication_extra: Option<MirrordKafkaClientAuthExtra>,
 }
 
 /// Property to use when creating operator's Kafka client.
@@ -47,6 +51,29 @@ pub struct MirrordKafkaClientProperty {
     pub value: Option<String>,
 }
 
+/// Additional authentication configuration for Kafka splitting.
+#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct MirrordKafkaClientAuthExtra {
+    /// Authentication kind.
+    ///
+    /// Right now the only supported value is `MSK_IAM`, which enables IAM/OAUTHBEARER
+    /// authentication with Amazon Managed Streaming for Apache Kafka.
+    ///
+    /// When this is set to `MSK_IAM`, additional properties are merged into the configuration:
+    /// 1. `sasl.mechanism=OAUTHBEARER`
+    /// 2. `security.protocol=SASL_SSL`
+    pub kind: String,
+    /// AWS region of the MSK cluster.
+    ///
+    /// Required when `kind` is `MSK_IAM`.
+    pub aws_region: Option<String>,
+}
+
+impl MirrordKafkaClientAuthExtra {
+    pub const MSK_IAM: &'static str = "MSK_IAM";
+}
+
 /// Defines splittable Kafka topics consumed by some workload living in the same namespace.
 ///
 /// # Concurrent splitting