feat: propagate sasl.client.callback.handler.class for MSK IAM producers

api/src/main/java/io/github/microcks/operator/api/base/v1alpha1/KafkaAuthenticationSpec.java

@@ -30,7 +30,7 @@
 @JsonIgnoreProperties(ignoreUnknown = true)
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @JsonPropertyOrder({ "type", "truststoreType", "truststoreSecretRef", "keystoreType", "keystoreSecretRef",
-      "saslMechanism", "saslJaasConfig" })
+      "saslMechanism", "saslJaasConfig", "saslLoginCallbackHandlerClass", "saslClientCallbackHandlerClass" })
 @Buildable(editableEnabled = false, builderPackage = "io.fabric8.kubernetes.api.builder")
 public class KafkaAuthenticationSpec {
 
@@ -55,6 +55,12 @@ public class KafkaAuthenticationSpec {
    @JsonPropertyDescription("Additional JAAS config for SASL_TLS authentication type")
    private String saslJaasConfig;
 
+   @JsonPropertyDescription("SASL Login Callback Handler class for SASL_TLS authentication type")
+   private String saslLoginCallbackHandlerClass;
+
+   @JsonPropertyDescription("SASL Client Callback Handler class for SASL_TLS authentication type")
+   private String saslClientCallbackHandlerClass;
+
    public KafkaAuthenticationType getType() {
       return type;
    }
@@ -110,4 +116,20 @@ public String getSaslJaasConfig() {
    public void setSaslJaasConfig(String saslJaasConfig) {
       this.saslJaasConfig = saslJaasConfig;
    }
+
+   public String getSaslLoginCallbackHandlerClass() {
+      return saslLoginCallbackHandlerClass;
+   }
+
+   public void setSaslLoginCallbackHandlerClass(String saslLoginCallbackHandlerClass) {
+      this.saslLoginCallbackHandlerClass = saslLoginCallbackHandlerClass;
+   }
+
+   public String getSaslClientCallbackHandlerClass() {
+      return saslClientCallbackHandlerClass;
+   }
+
+   public void setSaslClientCallbackHandlerClass(String saslClientCallbackHandlerClass) {
+      this.saslClientCallbackHandlerClass = saslClientCallbackHandlerClass;
+   }
 }

documentation/microcks-cr.md

@@ -570,6 +570,7 @@ Here are below the configuration properties of the Kafka support feature:
 | `async.kafka.authentication` | `saslMechanism`                 | **Optional**. For SASL authentication, you'll have to specify an additional authentication mechanism such as `SCRAM-SHA-512` or `OAUTHBEARER`                                                                          |
 | `async.kafka.authentication` | `saslJaasConfig`                | **Optional**. For SASL authentication, you'll have to specify a JAAS configuration line with login module, username and password.                                                                                      |
 | `async.kafka.authentication` | `saslLoginCallbackHandlerClass` | **Optional**. For SASL authentication, you may want to provide a Login Callback Handler implementations. This implementation may be provided by extending the main and `async-minion` images and adding your own libs. |
+| `async.kafka.authentication` | `saslClientCallbackHandlerClass`| **Optional**. For SASL authentication, you may want to provide a Client Callback Handler implementations (e.g. for Amazon MSK IAM). This implementation may be provided by extending the main and `async-minion` images and adding your own libs. |
 
 #### MQTT feature details
 

operator/src/main/resources/templates/AsyncMinionConfigMapDependentResource/application.properties

@@ -104,6 +104,20 @@ keycloak.auth.url=http://{name}-keycloak.{namespace}.svc.{spec.clusterDomain}:80
 %kube.mp.messaging.incoming.microcks-asyncapi-triggers.sasl.mechanism={spec.features.async.kafka.authentication.saslMechanism}
 %kube.mp.messaging.incoming.microcks-asyncapi-triggers.sasl.jaas.config={spec.features.async.kafka.authentication.saslJaasConfig}
 {/if}
+{#if spec.features.async.kafka.authentication.saslLoginCallbackHandlerClass != null}
+%kube.kafka.sasl.login.callback.handler.class={spec.features.async.kafka.authentication.saslLoginCallbackHandlerClass}
+%kube.mp.messaging.incoming.microcks-services-updates.sasl.login.callback.handler.class={spec.features.async.kafka.authentication.saslLoginCallbackHandlerClass}
+{#if minorVersion >= 14}
+%kube.mp.messaging.incoming.microcks-asyncapi-triggers.sasl.login.callback.handler.class={spec.features.async.kafka.authentication.saslLoginCallbackHandlerClass}
+{/if}
+{/if}
+{#if spec.features.async.kafka.authentication.saslClientCallbackHandlerClass != null}
+%kube.kafka.sasl.client.callback.handler.class={spec.features.async.kafka.authentication.saslClientCallbackHandlerClass}
+%kube.mp.messaging.incoming.microcks-services-updates.sasl.client.callback.handler.class={spec.features.async.kafka.authentication.saslClientCallbackHandlerClass}
+{#if minorVersion >= 14}
+%kube.mp.messaging.incoming.microcks-asyncapi-triggers.sasl.client.callback.handler.class={spec.features.async.kafka.authentication.saslClientCallbackHandlerClass}
+{/if}
+{/if}
 {/if}
 {/if}
 

operator/src/test/java/io/github/microcks/operator/base/resources/AsyncMinionConfigMapDependentResourceTest.java

@@ -22,6 +22,10 @@
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import io.github.microcks.operator.api.base.v1alpha1.AsyncFeatureSpec;
+import io.github.microcks.operator.api.base.v1alpha1.KafkaAuthenticationSpec;
+import io.github.microcks.operator.api.base.v1alpha1.KafkaAuthenticationType;
+import io.github.microcks.operator.api.base.v1alpha1.KafkaSpec;
 import io.quarkus.test.junit.QuarkusTest;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -70,5 +74,34 @@ void testAsyncApiTriggersPropertiesRenderedFor1_15_0() throws Exception {
       Assertions.assertTrue(applicationProperties.contains(ASYNCAPI_TRIGGERS_BOOTSTRAP),
             "microcks-asyncapi-triggers bootstrap.servers property must be present for Microcks 1.15.0");
    }
+
+   @Test
+   void testSaslCallbackHandlerPropertiesRendered() throws Exception {
+      Microcks microcks = buildMicrocks("1.14.0");
+
+      KafkaAuthenticationSpec authSpec = new KafkaAuthenticationSpec();
+      authSpec.setType(KafkaAuthenticationType.SASL_SSL);
+      authSpec.setSaslClientCallbackHandlerClass("software.amazon.msk.auth.iam.IAMClientCallbackHandler");
+      authSpec.setSaslLoginCallbackHandlerClass("com.example.MyLoginHandler");
+
+      KafkaSpec kafkaSpec = new KafkaSpec();
+      kafkaSpec.setInstall(false);
+      kafkaSpec.setUrl("my-cluster:9092");
+      kafkaSpec.setAuthentication(authSpec);
+
+      AsyncFeatureSpec asyncSpec = new AsyncFeatureSpec();
+      asyncSpec.setEnabled(true);
+      asyncSpec.setKafka(kafkaSpec);
+
+      microcks.getSpec().getFeatures().setAsync(asyncSpec);
+
+      ConfigMap configMap = new AsyncMinionConfigMapDependentResource().desired(microcks, null);
+      String applicationProperties = configMap.getData().get("application.properties");
+
+      Assertions.assertTrue(applicationProperties.contains("kafka.sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler"),
+            "sasl client callback handler class property must be present");
+      Assertions.assertTrue(applicationProperties.contains("kafka.sasl.login.callback.handler.class=com.example.MyLoginHandler"),
+            "sasl login callback handler class property must be present");
+   }
 }