Apply client options as defined on the cluster

Author: Crim

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtil.java

@@ -31,6 +31,8 @@
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.sourcelab.kafka.webview.ui.manager.kafka.config.ClusterConfig;
 
 import java.util.ArrayList;
@@ -46,6 +48,8 @@
  * Utility class to DRY out common Kafka client configuration options that apply to multiple client types.
  */
 public class KafkaClientConfigUtil {
+    private static final Logger logger = LoggerFactory.getLogger(KafkaClientConfigUtil.class);
+
     /**
      * Path on filesystem where keystores are persisted.
      */
@@ -116,9 +120,38 @@ private Map<String, Object> applyCommonSettings(
         // Optionally configure SASL
         applySaslSettings(clusterConfig, config);
 
+        // Apply cluster client properties if defined.
+        // Note: This should always be applied last.
+        applyClusterClientProperties(clusterConfig, config);
+
         return config;
     }
 
+    /**
+     * If client properties are defined on the cluster, they get applied last ontop of everything else.
+     * @param clusterConfig configuration properties.
+     * @param config config to be applied to.
+     */
+    private void applyClusterClientProperties(final ClusterConfig clusterConfig, final Map<String, Object> config) {
+        if (clusterConfig.getClusterClientProperties().isEmpty()) {
+            return;
+        }
+
+        for (final Map.Entry<String, String> entry : clusterConfig.getClusterClientProperties().entrySet()) {
+            if (config.containsKey(entry.getKey())) {
+                // Log a warning as behavior may be altered in a way that causes Kafka WebView to no longer function.
+                logger.warn(
+                    "Client property defined on the cluster replaced property '{}'. "
+                    + "The original value of '{}' was replaced with with '{}'. "
+                    + "Overriding of configuration properties in this way may cause Kafka Webview to not function correctly.",
+                    entry.getKey(), config.get(entry.getKey()), entry.getValue()
+                );
+            }
+            // Set value.
+            config.put(entry.getKey(), entry.getValue());
+        }
+    }
+
     /**
      * If SSL is configured for this cluster, apply the settings.
      * @param clusterConfig Cluster configuration definition to source values from.

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/config/ClusterConfig.java

@@ -24,13 +24,18 @@
 
 package org.sourcelab.kafka.webview.ui.manager.kafka.config;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.sourcelab.kafka.webview.ui.manager.encryption.SecretManager;
 import org.sourcelab.kafka.webview.ui.manager.sasl.SaslProperties;
 import org.sourcelab.kafka.webview.ui.manager.sasl.SaslUtility;
 import org.sourcelab.kafka.webview.ui.model.Cluster;
 
+import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -58,6 +63,11 @@ public class ClusterConfig {
     private final String saslMechanism;
     private final String saslJaas;
 
+    /**
+     * Client Properties defined on the Cluster configuration.
+     */
+    private final Map<String, String> clusterClientProperties;
+
     /**
      * Private constructor for connecting to SSL brokers.
      */
@@ -72,7 +82,9 @@ private ClusterConfig(
         final String saslPlaintextUsername,
         final String saslPlaintextPassword,
         final String saslMechanism,
-        final String saslJaas) {
+        final String saslJaas,
+        final Map<String, String> clusterClientProperties
+    ) {
 
         this.brokerHosts = brokerHosts;
 
@@ -89,6 +101,11 @@ private ClusterConfig(
         this.saslPlaintextPassword = saslPlaintextPassword;
         this.saslMechanism = saslMechanism;
         this.saslJaas = saslJaas;
+
+        // Shallow copy the cluster client properties.
+        this.clusterClientProperties = Collections.unmodifiableMap(
+            new HashMap<>(clusterClientProperties)
+        );
     }
 
     public Set<String> getBrokerHosts() {
@@ -139,6 +156,10 @@ public String getSaslJaas() {
         return saslJaas;
     }
 
+    public Map<String, String> getClusterClientProperties() {
+        return clusterClientProperties;
+    }
+
     @Override
     public String toString() {
         return "ClusterConfig{"
@@ -196,6 +217,19 @@ public static Builder newBuilder(final Cluster cluster, final SecretManager secr
             builder.withUseSasl(false);
         }
 
+        // If we have defined cluster client options, decode and set them.
+        if (cluster.getOptionParameters() != null) {
+            final ObjectMapper objectMapper = new ObjectMapper();
+            Map<String, String> customOptions;
+            try {
+                customOptions = objectMapper.readValue(cluster.getOptionParameters(), Map.class);
+            } catch (final IOException e) {
+                // Fail safe?
+                customOptions = new HashMap<>();
+            }
+            builder.withClusterClientConfig(customOptions);
+        }
+
         return builder;
     }
 
@@ -223,6 +257,12 @@ public static final class Builder {
         private String saslMechanism;
         private String saslJaas;
 
+        /**
+         * Override properties defined from the cluster.option_parameters field.
+         * These should be applied LAST ontop of all the other config options.
+         */
+        private Map<String, String> clusterOverrideProperties = new HashMap<>();
+
         private Builder() {
         }
 
@@ -323,6 +363,22 @@ public Builder withSaslJaas(final String saslJaas) {
             return this;
         }
 
+        /**
+         * Declare Override Properties defined on the cluster.
+         */
+        public Builder withClusterClientConfig(final Map<String, String> clusterClientConfig) {
+            this.clusterOverrideProperties.putAll(clusterClientConfig);
+            return this;
+        }
+
+        /**
+         * Declare Override Properties defined on the cluster.
+         */
+        public Builder withClusterClientConfig(final String key, final String value) {
+            this.clusterOverrideProperties.put(key, value);
+            return this;
+        }
+
         /**
          * Create ClusterConfig instance from builder values.
          */
@@ -340,7 +396,9 @@ public ClusterConfig build() {
                 saslPlaintextUsername,
                 saslPlaintextPassword,
                 saslMechanism,
-                saslJaas
+                saslJaas,
+                // Cluster client properties
+                clusterOverrideProperties
             );
         }
     }

kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtilTest.java

@@ -31,6 +31,7 @@
 import org.junit.Test;
 import org.sourcelab.kafka.webview.ui.manager.kafka.config.ClusterConfig;
 
+import java.util.HashMap;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
@@ -81,6 +82,43 @@ public void testApplyCommonSettings_noSsl_noSasl() {
         validateNoKey(config, CommonClientConfigs.SECURITY_PROTOCOL_CONFIG);
     }
 
+    /**
+     * Basic smoke test, without SSL or SASL options, but has cluster properties.
+     */
+    @Test
+    public void testApplyCommonSettings_noSsl_noSasl_withClientProperties() {
+        final Map<String, String> customProperties = new HashMap<>();
+        customProperties.put("key3", "value3");
+        customProperties.put("key4", "value4");
+
+        final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
+            .withBrokerHosts(expectedBrokerHosts)
+            .withUseSsl(false)
+            .withUseSasl(false)
+            // Use both setters
+            .withClusterClientConfig("key1", "value1")
+            .withClusterClientConfig("key2", "value2")
+            .withClusterClientConfig(customProperties)
+            // Build it.
+            .build();
+
+        final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId);
+
+        // Validate
+        validateDefaultKeys(config);
+        validateNoSsl(config);
+        validateNoSasl(config);
+
+        // Validate this is not set.
+        validateNoKey(config, CommonClientConfigs.SECURITY_PROTOCOL_CONFIG);
+
+        // Validate custom properties applied
+        validateKey(config, "key1", "value1");
+        validateKey(config, "key2", "value2");
+        validateKey(config, "key3", "value3");
+        validateKey(config, "key4", "value4");
+    }
+
     /**
      * Basic smoke test with SSL and TrustStore, without SASL options.
      */