Added a custom SASL config option to the Topic Operator #5473

CHANGELOG.md

@@ -10,6 +10,7 @@
 * Added support for configuring the `externalIPs` field in node port type services.
 * The `UnidirectionalTopicOperator` feature gate moves to GA stage and is permanently enabled without the possibility to disable it.
   If the topics whose names start with `strimzi-store-topic` and `strimzi-topic-operator` still exist, you can delete them.
+* Added support for custom SASL config in standalone Topic Operator deployment to support alternate access controllers (i.e. `AWS_MSK_IAM`)
 
 ### Changes, deprecations and removals
 

documentation/modules/deploying/proc-deploy-topic-operator-standalone.adoc

@@ -141,6 +141,69 @@ env:
 <3> The keystore contains the private key for mTLS authentication.
 <4> The password for accessing the keystore.
 
+. If you need to configure custom SASL authentication, you can define the necessary authentication properties using the `STRIMZI_SASL_CUSTOM_CONFIG_JSON` environment variable for the standalone operator.
+For example, this configuration may be used for accessing a Kafka cluster in a cloud provider with a custom login module like the link:https://github.com/aws/aws-msk-iam-auth[Amazon MSK Library for AWS Identity and Access Management^] (`aws-msk_iam-auth`).
++
+The property `STRIMZI_ALTERABLE_TOPIC_CONFIG` defaults to `ALL`, allowing all `.spec.config` properties to be set in the `KafkaTopic` resource.  
+If this setting is not suitable for a managed Kafka service, do as follows:
++
+--
+* If only a subset of properties is configurable, list them as comma-separated values.
+* If no  properties are to be configured, use `NONE`, which is equivalent to an empty property list.
+--
++
+NOTE:  Only Kafka configuration properties starting with `sasl.` can be set with the `STRIMZI_SASL_CUSTOM_CONFIG_JSON` environment variable.
++
+.Example custom SASL configuration
+[source,shell,subs=+quotes]
+----
+# ....
+env:
+  - name: STRIMZI_SASL_ENABLED
+    value: "true"
+  - name: STRIMZI_SECURITY_PROTOCOL
+    value: SASL_SSL
+  - name: STRIMZI_SKIP_CLUSTER_CONFIG_REVIEW # <1>
+    value: "true"
+  - name: STRIMZI_ALTERABLE_TOPIC_CONFIG # <2>
+    value: compression.type, max.message.bytes, message.timestamp.difference.max.ms, message.timestamp.type, retention.bytes, retention.ms
+  - name: STRIMZI_SASL_CUSTOM_CONFIG_JSON # <3>
+    value: |
+      {
+        "sasl.mechanism": "AWS_MSK_IAM",
+        "sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
+        "sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
+      }
+  - name: STRIMZI_PUBLIC_CA
+    value: "true"
+  - name: STRIMZI_TRUSTSTORE_LOCATION
+    value: /etc/pki/java/cacerts
+  - name: STRIMZI_TRUSTSTORE_PASSWORD
+    value: changeit
+  - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS
+    value: my-kafka-cluster-.kafka-serverless.us-east-1.amazonaws.com:9098
+# ...
+----
+<1> Disables cluster configuration lookup for managed Kafka services that don't allow topic configuration changes.
+<2> Defines the topic configuration properties that can be updated based on the limitations set by managed Kafka services.
+<3> Specifies the SASL properties to be set in JSON format. Only properties starting with `sasl.` are allowed.
++
+.Example Dockerfile with external jars
+[source,shell,subs=+quotes]
+----
+FROM strimzi/operator:latest
+
+USER root
+
+RUN mkdir -p ${STRIMZI_HOME}/external-libs
+RUN chmod +rx ${STRIMZI_HOME}/external-libs
+
+COPY ./aws-msk-iam-auth-and-dependencies/* ${STRIMZI_HOME}/external-libs/
+ENV JAVA_CLASSPATH=${STRIMZI_HOME}/external-libs/*
+
+USER 1001
+----
+
 . Apply the changes to the `Deployment` configuration to deploy the Topic Operator.
 
 . Check the status of the deployment:

test/src/main/java/io/strimzi/test/executor/Exec.java

@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import java.util.Scanner;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -339,6 +340,11 @@ private void storeOutputsToFile() {
      * @return true.false
      */
     public static boolean isExecutableOnPath(String cmd) {
+        var osName = System.getProperty("os.name");
+        if (osName.toLowerCase(Locale.US).startsWith("windows")) {
+            cmd += ".exe";
+        }
+
         for (String dir : PATH_SPLITTER.split(System.getenv("PATH"))) {
             if (new File(dir, cmd).canExecute()) {
                 return true;

topic-operator/src/main/java/io/strimzi/operator/topic/BatchingTopicController.java

@@ -46,6 +46,7 @@
 import java.io.InterruptedIOException;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -104,13 +105,18 @@ public class BatchingTopicController {
         this.selector = Objects.requireNonNull(selector);
         this.useFinalizer = config.useFinalizer();
         this.admin = admin;
-        // Get the config of some broker and check whether auto topic creation is enabled
-        Optional<String> autoCreateValue = getClusterConfig(admin, AUTO_CREATE_TOPICS_ENABLE);
-        if (autoCreateValue.isPresent() ? "true".equals(autoCreateValue.get()) : false) {
-            LOGGER.warnOp(
-                    "It is recommended that " + AUTO_CREATE_TOPICS_ENABLE + " is set to 'false' " +
-                    "to avoid races between the operator and Kafka applications auto-creating topics");
+
+        var skipClusterConfigReview = config.skipClusterConfigReview();
+        if (!skipClusterConfigReview) {
+            // Get the config of some broker and check whether auto topic creation is enabled
+            Optional<String> autoCreateValue = getClusterConfig(admin, AUTO_CREATE_TOPICS_ENABLE);
+            if (autoCreateValue.filter("true"::equals).isPresent()) {
+                LOGGER.warnOp(
+                      "It is recommended that " + AUTO_CREATE_TOPICS_ENABLE + " is set to 'false' " +
+                            "to avoid races between the operator and Kafka applications auto-creating topics");
+            }
         }
+
         this.kubeClient = kubeClient;
         this.metrics = metrics;
         this.namespace = config.namespace();
@@ -581,6 +587,11 @@ private PartitionedByError<ReconcilableTopic, CurrentState> findDifferentRf(Part
      * @param reconcilableTopics Reconcilable topic.
      */
     private void warnTooLargeMinIsr(List<ReconcilableTopic> reconcilableTopics) {
+        if (config.skipClusterConfigReview()) {
+            // This method is for internal configurations. So skipping.
+            return;
+        }
+
         Optional<String> clusterMinIsr = getClusterConfig(admin, MIN_INSYNC_REPLICAS);
         for (ReconcilableTopic reconcilableTopic : reconcilableTopics) {
             var topicConfig = reconcilableTopic.kt().getSpec().getConfig();
@@ -709,7 +720,7 @@ private void accumulateResults(Map<ReconcilableTopic, Either<TopicOperatorExcept
         replicasChangeResults.errors().forEach(pair -> putResult(results, pair.getKey(), Either.ofLeft(pair.getValue())));
     }
 
-    private static List<Pair<ReconcilableTopic, Collection<AlterConfigOp>>> configChanges(Map<ReconcilableTopic, Either<TopicOperatorException, Object>> results, PartitionedByError<ReconcilableTopic, CurrentState> currentStatesOrError) {
+    private List<Pair<ReconcilableTopic, Collection<AlterConfigOp>>> configChanges(Map<ReconcilableTopic, Either<TopicOperatorException, Object>> results, PartitionedByError<ReconcilableTopic, CurrentState> currentStatesOrError) {
         // Determine config changes
         Map<Boolean, List<Pair<ReconcilableTopic, Collection<AlterConfigOp>>>> alterConfigs = currentStatesOrError.ok().map(pair -> {
             var reconcilableTopic = pair.getKey();
@@ -1083,7 +1094,7 @@ private static Either<TopicOperatorException, NewPartitions> buildNewPartitions(
         }
     }
 
-    private static Collection<AlterConfigOp> buildAlterConfigOps(Reconciliation reconciliation, KafkaTopic kt, Config configs) {
+    private Collection<AlterConfigOp> buildAlterConfigOps(Reconciliation reconciliation, KafkaTopic kt, Config configs) {
         Set<AlterConfigOp> alterConfigOps = new HashSet<>();
         if (hasConfig(kt)) {
             for (var specConfigEntry : kt.getSpec().getConfig().entrySet()) {
@@ -1109,6 +1120,9 @@ private static Collection<AlterConfigOp> buildAlterConfigOps(Reconciliation reco
                     new ConfigEntry(key, null),
                     AlterConfigOp.OpType.DELETE));
         }
+
+        skipNonAlterableConfigs(alterConfigOps);
+
         if (alterConfigOps.isEmpty()) {
             LOGGER.debugCr(reconciliation, "No config change");
         } else {
@@ -1117,24 +1131,89 @@ private static Collection<AlterConfigOp> buildAlterConfigOps(Reconciliation reco
         return alterConfigOps;
     }
 
+    /**
+     * <p>The Topic Operator {@code alterableTopicConfig} can be used to specify a comma separated list of Kafka
+     * topic configurations that can be altered by users through {@code .spec.config}. Keep in mind that if changes
+     * are applied directly in Kafka, the operator will try to revert them producing a warning.</p>
+     *
+     * <p>This is useful in standalone mode when you have a Kafka service that restricts alter operations
+     * to a subset of all the Kafka topic configurations.</p>
+     *
+     * <p>The default value is "all", which means no restrictions in changing {@code .spec.config}.
+     * The opposite is "none", which can be set to explicitly disable any change.</p>
+     *
+     * @param alterConfigOps Requested alter config operations.
+     */
+    private void skipNonAlterableConfigs(Set<AlterConfigOp> alterConfigOps) {
+        var alterableConfigs = config.alterableTopicConfig();
+        if (alterableConfigs != null && alterConfigOps != null && !alterableConfigs.isEmpty()) {
+            if (alterableConfigs.equalsIgnoreCase("none")) {
+                alterConfigOps.clear();
+            } else if (!alterableConfigs.equalsIgnoreCase("all")) {
+                var alterablePropertySet = Arrays.stream(alterableConfigs.replaceAll("\\s", "").split(","))
+                      .collect(Collectors.toSet());
+                alterConfigOps.removeIf(op -> !alterablePropertySet.contains(op.configEntry().name()));
+            }
+        }
+    }
+
     private static boolean hasConfig(KafkaTopic kt) {
         return kt.getSpec() != null
                 && kt.getSpec().getConfig() != null;
     }
 
     private void updateStatusForSuccess(ReconcilableTopic reconcilableTopic) {
+        List<Condition> conditions = new ArrayList<>();
+        conditions.add(new ConditionBuilder()
+              .withType(TopicOperatorUtil.isPaused(reconcilableTopic.kt()) ? "ReconciliationPaused" : "Ready")
+              .withStatus("True")
+              .withLastTransitionTime(StatusUtils.iso8601Now())
+              .build());
+
+        addNonAlterableConfigsWarning(reconcilableTopic, conditions);
+
         reconcilableTopic.kt().setStatus(
             new KafkaTopicStatusBuilder(reconcilableTopic.kt().getStatus())
-                .withConditions(List.of(new ConditionBuilder()
-                    .withType(TopicOperatorUtil.isPaused(reconcilableTopic.kt()) ? "ReconciliationPaused" : "Ready")
-                    .withStatus("True")
-                    .withLastTransitionTime(StatusUtils.iso8601Now())
-                    .build()))
+                .withConditions(conditions)
             .build());
         updateStatus(reconcilableTopic);
         metrics.successfulReconciliationsCounter(namespace).increment();
     }
-
+
+    private void addNonAlterableConfigsWarning(ReconcilableTopic reconcilableTopic,
+                                               List<Condition> conditions) {
+        var readOnlyConfigs = new ArrayList<>();
+        var alterableConfigs = config.alterableTopicConfig();
+
+        if (reconcilableTopic != null && reconcilableTopic.kt() != null
+              && hasConfig(reconcilableTopic.kt()) && alterableConfigs != null) {
+            if (alterableConfigs.equalsIgnoreCase("NONE")) {
+                reconcilableTopic.kt().getSpec().getConfig().forEach((key, value) -> readOnlyConfigs.add(key));
+            } else if (!alterableConfigs.equalsIgnoreCase("ALL") && !alterableConfigs.isBlank()) {
+                var alterablePropertySet = Arrays.stream(alterableConfigs.replaceAll("\\s", "").split(","))
+                      .collect(Collectors.toSet());
+                reconcilableTopic.kt().getSpec().getConfig().forEach((key, value) -> {
+                    if (!alterablePropertySet.contains(key)) {
+                        readOnlyConfigs.add(key);
+                    }
+                });
+            }
+        }
+
+        if (!readOnlyConfigs.isEmpty()) {
+            var properties = String.join(", ", readOnlyConfigs.toArray(new String[0]));
+            var message = "These .spec.config properties are not configurable: [" + properties + "]";
+            LOGGER.warnCr(reconcilableTopic.reconciliation(), message);
+            conditions.add(new ConditionBuilder()
+                  .withMessage(message)
+                  .withReason("NotConfigurable")
+                  .withStatus("True")
+                  .withType("Warning")
+                  .withLastTransitionTime(StatusUtils.iso8601Now())
+                  .build());
+        }
+    }
+
     private void updateStatusForException(ReconcilableTopic reconcilableTopic, Exception e) {
         String reason;
         if (e instanceof TopicOperatorException) {