Parse additional config at the end of props setup to allow rewrite configurations (#55)

michalxo · web-flow · commit 3b801e4f444f · 2021-03-16T08:57:01.000+01:00
* UPD: Various client updates

Main goal of my changes is to get rid of sasl.login.callback.handler.class, as it was used there hardwired.
Using this additional configuration, sasl.mechanism can be overridden as well (hardwired to OAUTHBEARER now)
Minor updates:
  - some clean up code
  - simplify statements
  - added readme docs for this new env variable

Signed-off-by: Michal Tóth &lt;mtoth@redhat.com&gt;

* UPD: Remove custom SASL_CALLBACK_CLASS env var - pass it via additional Config

Signed-off-by: Michal Tóth &lt;mtoth@redhat.com&gt;
diff --git a/README.md b/README.md
@@ -63,11 +63,11 @@ Producer
 * `USER_CRT` - the user's certificate
 * `USER_KEY` - the user's private key
 * `LOG_LEVEL` - logging level  
-* `PRODUCER_ACKS` = acknowledgement level
-* `HEADERS` = custom headers list separated by commas of `key1=value1, key2=value2`
-* `ADDITIONAL_CONFIG` = additional configuration for a producer application. The form is `key=value` records separated by new line character
-* `BLOCKING_PRODUCER` = if it's set, the producer will block another message until ack will be received
-* `MESSAGES_PER_TRANSACTION` = how many messages will be part of one transaction. Transaction config could be set via `ADDITIONAL_CONFIG` variable. Default is 10.
+* `PRODUCER_ACKS` - acknowledgement level
+* `HEADERS` - custom headers list separated by commas of `key1=value1, key2=value2`
+* `BLOCKING_PRODUCER` - if it's set, the producer will block another message until ack will be received
+* `MESSAGES_PER_TRANSACTION` - how many messages will be part of one transaction. Transaction config could be set via `ADDITIONAL_CONFIG` variable. Default is 10.
+* `ADDITIONAL_CONFIG` - additional configuration for a producer application. Notice, that you can also override any previously set variable by setting this. The form is `key=value` records separated by new line character
 
 Consumer  
 * `BOOTSTRAP_SERVERS` - comma-separated host and port pairs that is a list of Kafka broker addresses. The form of pair is `host:port`, e.g. `my-cluster-kafka-bootstrap:9092` 
@@ -78,7 +78,7 @@ Consumer
 * `USER_CRT` - the user's certificate
 * `USER_KEY` - the user's private key
 * `LOG_LEVEL` - logging level  
-* `ADDITIONAL_CONFIG` = additional configuration for a consumer application. The form is `key=value` records separated by new line character
+* `ADDITIONAL_CONFIG` - additional configuration for a consumer application. Notice, that you can also override any previously set variable by setting this. The form is `key=value` records separated by new line character
 
 Streams  
 * `BOOTSTRAP_SERVERS` - comma-separated host and port pairs that is a list of Kafka broker addresses. The form of pair is `host:port`, e.g. `my-cluster-kafka-bootstrap:9092`
@@ -90,7 +90,7 @@ Streams
 * `USER_CRT` - the user's certificate
 * `USER_KEY` - the user's private key
 * `LOG_LEVEL` - logging level
-* `ADDITIONAL_CONFIG` = additional configuration for a streams application. The form is `key=value` records separated by new line character
+* `ADDITIONAL_CONFIG` - additional configuration for a streams application. Notice, that you can also override any previously set variable by setting this. The form is `key=value` records separated by new line character.
 
 ### Tracing
 
diff --git a/java/kafka/consumer/src/main/java/KafkaConsumerConfig.java b/java/kafka/consumer/src/main/java/KafkaConsumerConfig.java
@@ -34,8 +34,14 @@ public class KafkaConsumerConfig {
     private final String oauthRefreshToken;
     private final String oauthTokenEndpointUri;
     private final String additionalConfig;
+    private final String saslLoginCallbackClass = "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler";
+    ;
 
-    public KafkaConsumerConfig(String bootstrapServers, String topic, String groupId, String clientRack, Long messageCount, String trustStorePassword, String trustStorePath, String keyStorePassword, String keyStorePath, String oauthClientId, String oauthClientSecret, String oauthAccessToken, String oauthRefreshToken, String oauthTokenEndpointUri, String additionalConfig) {
+
+    public KafkaConsumerConfig(String bootstrapServers, String topic, String groupId, String clientRack, Long messageCount,
+                               String trustStorePassword, String trustStorePath, String keyStorePassword, String keyStorePath,
+                               String oauthClientId, String oauthClientSecret, String oauthAccessToken, String oauthRefreshToken,
+                               String oauthTokenEndpointUri, String additionalConfig) {
         this.bootstrapServers = bootstrapServers;
         this.topic = topic;
         this.groupId = groupId;
@@ -51,26 +57,29 @@ public KafkaConsumerConfig(String bootstrapServers, String topic, String groupId
         this.oauthRefreshToken = oauthRefreshToken;
         this.oauthTokenEndpointUri = oauthTokenEndpointUri;
         this.additionalConfig = additionalConfig;
+        ;
     }
 
     public static KafkaConsumerConfig fromEnv() {
         String bootstrapServers = System.getenv("BOOTSTRAP_SERVERS");
         String topic = System.getenv("TOPIC");
         String groupId = System.getenv("GROUP_ID");
-        String clientRack = System.getenv("CLIENT_RACK") == null ? null : System.getenv("CLIENT_RACK");
+        String clientRack = System.getenv("CLIENT_RACK");
         Long messageCount = System.getenv("MESSAGE_COUNT") == null ? DEFAULT_MESSAGES_COUNT : Long.valueOf(System.getenv("MESSAGE_COUNT"));
-        String trustStorePassword = System.getenv("TRUSTSTORE_PASSWORD") == null ? null : System.getenv("TRUSTSTORE_PASSWORD");
-        String trustStorePath = System.getenv("TRUSTSTORE_PATH") == null ? null : System.getenv("TRUSTSTORE_PATH");
-        String keyStorePassword = System.getenv("KEYSTORE_PASSWORD") == null ? null : System.getenv("KEYSTORE_PASSWORD");
-        String keyStorePath = System.getenv("KEYSTORE_PATH") == null ? null : System.getenv("KEYSTORE_PATH");
+        String trustStorePassword = System.getenv("TRUSTSTORE_PASSWORD");
+        String trustStorePath = System.getenv("TRUSTSTORE_PATH");
+        String keyStorePassword = System.getenv("KEYSTORE_PASSWORD");
+        String keyStorePath = System.getenv("KEYSTORE_PATH");
         String oauthClientId = System.getenv("OAUTH_CLIENT_ID");
         String oauthClientSecret = System.getenv("OAUTH_CLIENT_SECRET");
         String oauthAccessToken = System.getenv("OAUTH_ACCESS_TOKEN");
         String oauthRefreshToken = System.getenv("OAUTH_REFRESH_TOKEN");
         String oauthTokenEndpointUri = System.getenv("OAUTH_TOKEN_ENDPOINT_URI");
         String additionalConfig = System.getenv().getOrDefault("ADDITIONAL_CONFIG", "");
 
-        return new KafkaConsumerConfig(bootstrapServers, topic, groupId, clientRack, messageCount, trustStorePassword, trustStorePath, keyStorePassword, keyStorePath, oauthClientId, oauthClientSecret, oauthAccessToken, oauthRefreshToken, oauthTokenEndpointUri, additionalConfig);
+        return new KafkaConsumerConfig(bootstrapServers, topic, groupId, clientRack, messageCount, trustStorePassword, trustStorePath,
+                keyStorePassword, keyStorePath, oauthClientId, oauthClientSecret, oauthAccessToken, oauthRefreshToken, oauthTokenEndpointUri,
+                additionalConfig);
     }
 
     public static Properties createProperties(KafkaConsumerConfig config) {
@@ -85,20 +94,6 @@ public static Properties createProperties(KafkaConsumerConfig config) {
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
 
-        if (!config.getAdditionalConfig().isEmpty()) {
-            StringTokenizer tok = new StringTokenizer(config.getAdditionalConfig(), ", \t\n\r");
-            while (tok.hasMoreTokens()) {
-                String record = tok.nextToken();
-                int endIndex = record.indexOf('=');
-                if (endIndex == -1) {
-                    throw new RuntimeException("Failed to parse Map from String");
-                }
-                String key = record.substring(0, endIndex);
-                String value = record.substring(endIndex + 1);
-                props.put(key.trim(), value.trim());
-            }
-        }
-
         if (config.getTrustStorePassword() != null && config.getTrustStorePath() != null)   {
             log.info("Configuring truststore");
             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
@@ -115,15 +110,36 @@ public static Properties createProperties(KafkaConsumerConfig config) {
             props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, config.getKeyStorePath());
         }
 
+        Properties additionalProps = new Properties();
+        if (!config.getAdditionalConfig().isEmpty()) {
+            StringTokenizer tok = new StringTokenizer(config.getAdditionalConfig(), System.lineSeparator());
+            while (tok.hasMoreTokens()) {
+                String record = tok.nextToken();
+                int endIndex = record.indexOf('=');
+                if (endIndex == -1) {
+                    throw new RuntimeException("Failed to parse Map from String");
+                }
+                String key = record.substring(0, endIndex);
+                String value = record.substring(endIndex + 1);
+                additionalProps.put(key.trim(), value.trim());
+            }
+        }
+
         if ((config.getOauthAccessToken() != null)
                 || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthRefreshToken() != null)
                 || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthClientSecret() != null))    {
+            log.info("Configuring OAuth");
             props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL".equals(props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)) ? "SASL_SSL" : "SASL_PLAINTEXT");
             props.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");
-            props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler");
+            if (!(additionalProps.containsKey(SaslConfigs.SASL_MECHANISM) && additionalProps.getProperty(SaslConfigs.SASL_MECHANISM).equals("PLAIN"))) {
+                props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, config.saslLoginCallbackClass);
+            }
         }
 
+        // override properties with defined additional properties
+        props.putAll(additionalProps);
+
         return props;
     }
 
diff --git a/java/kafka/pom.xml b/java/kafka/pom.xml
@@ -21,7 +21,7 @@
         <kafka.version>2.7.0</kafka.version>
         <opentracing-kafka.version>0.1.11</opentracing-kafka.version>
         <jaeger.version>1.1.0</jaeger.version>
-        <strimzi-oauth-callback.version>0.6.1</strimzi-oauth-callback.version>
+        <strimzi-oauth-callback.version>0.7.1</strimzi-oauth-callback.version>
     </properties>
 
     <dependencyManagement>
diff --git a/java/kafka/producer/src/main/java/KafkaProducerConfig.java b/java/kafka/producer/src/main/java/KafkaProducerConfig.java
@@ -35,8 +35,12 @@ public class KafkaProducerConfig {
     private final String oauthRefreshToken;
     private final String oauthTokenEndpointUri;
     private final String additionalConfig;
+    private final String saslLoginCallbackClass = "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler";
 
-    public KafkaProducerConfig(String bootstrapServers, String topic, int delay, Long messageCount, String message, String trustStorePassword, String trustStorePath, String keyStorePassword, String keyStorePath, String oauthClientId, String oauthClientSecret, String oauthAccessToken, String oauthRefreshToken, String oauthTokenEndpointUri, String acks, String additionalConfig, String headers) {
+    public KafkaProducerConfig(String bootstrapServers, String topic, int delay, Long messageCount, String message,
+                               String trustStorePassword, String trustStorePath, String keyStorePassword, String keyStorePath,
+                               String oauthClientId, String oauthClientSecret, String oauthAccessToken, String oauthRefreshToken,
+                               String oauthTokenEndpointUri, String acks, String additionalConfig, String headers) {
         this.bootstrapServers = bootstrapServers;
         this.topic = topic;
         this.delay = delay;
@@ -59,13 +63,13 @@ public KafkaProducerConfig(String bootstrapServers, String topic, int delay, Lon
     public static KafkaProducerConfig fromEnv() {
         String bootstrapServers = System.getenv("BOOTSTRAP_SERVERS");
         String topic = System.getenv("TOPIC");
-        int delay = Integer.valueOf(System.getenv("DELAY_MS"));
+        int delay = Integer.parseInt(System.getenv("DELAY_MS"));
         Long messageCount = System.getenv("MESSAGE_COUNT") == null ? DEFAULT_MESSAGES_COUNT : Long.valueOf(System.getenv("MESSAGE_COUNT"));
         String message = System.getenv("MESSAGE") == null ? DEFAULT_MESSAGE : System.getenv("MESSAGE");
-        String trustStorePassword = System.getenv("TRUSTSTORE_PASSWORD") == null ? null : System.getenv("TRUSTSTORE_PASSWORD");
-        String trustStorePath = System.getenv("TRUSTSTORE_PATH") == null ? null : System.getenv("TRUSTSTORE_PATH");
-        String keyStorePassword = System.getenv("KEYSTORE_PASSWORD") == null ? null : System.getenv("KEYSTORE_PASSWORD");
-        String keyStorePath = System.getenv("KEYSTORE_PATH") == null ? null : System.getenv("KEYSTORE_PATH");
+        String trustStorePassword = System.getenv("TRUSTSTORE_PASSWORD");
+        String trustStorePath = System.getenv("TRUSTSTORE_PATH");
+        String keyStorePassword = System.getenv("KEYSTORE_PASSWORD");
+        String keyStorePath = System.getenv("KEYSTORE_PATH");
         String oauthClientId = System.getenv("OAUTH_CLIENT_ID");
         String oauthClientSecret = System.getenv("OAUTH_CLIENT_SECRET");
         String oauthAccessToken = System.getenv("OAUTH_ACCESS_TOKEN");
@@ -75,7 +79,9 @@ public static KafkaProducerConfig fromEnv() {
         String headers = System.getenv("HEADERS");
         String additionalConfig = System.getenv().getOrDefault("ADDITIONAL_CONFIG", "");
 
-        return new KafkaProducerConfig(bootstrapServers, topic, delay, messageCount, message, trustStorePassword, trustStorePath, keyStorePassword, keyStorePath, oauthClientId, oauthClientSecret, oauthAccessToken, oauthRefreshToken, oauthTokenEndpointUri, acks, additionalConfig, headers);
+        return new KafkaProducerConfig(bootstrapServers, topic, delay, messageCount, message, trustStorePassword, trustStorePath,
+                keyStorePassword, keyStorePath, oauthClientId, oauthClientSecret, oauthAccessToken, oauthRefreshToken,
+                oauthTokenEndpointUri, acks, additionalConfig, headers);
     }
 
     public static Properties createProperties(KafkaProducerConfig config) {
@@ -85,20 +91,6 @@ public static Properties createProperties(KafkaProducerConfig config) {
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
 
-        if (!config.getAdditionalConfig().isEmpty()) {
-            StringTokenizer tok = new StringTokenizer(config.getAdditionalConfig(), ", \t\n\r");
-            while (tok.hasMoreTokens()) {
-                String record = tok.nextToken();
-                int endIndex = record.indexOf('=');
-                if (endIndex == -1) {
-                    throw new RuntimeException("Failed to parse Map from String");
-                }
-                String key = record.substring(0, endIndex);
-                String value = record.substring(endIndex + 1);
-                props.put(key.trim(), value.trim());
-            }
-        }
-
         if (config.getTrustStorePassword() != null && config.getTrustStorePath() != null)   {
             log.info("Configuring truststore");
             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
@@ -115,15 +107,36 @@ public static Properties createProperties(KafkaProducerConfig config) {
             props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, config.getKeyStorePath());
         }
 
+        Properties additionalProps = new Properties();
+        if (!config.getAdditionalConfig().isEmpty()) {
+            StringTokenizer tok = new StringTokenizer(config.getAdditionalConfig(), System.lineSeparator());
+            while (tok.hasMoreTokens()) {
+                String record = tok.nextToken();
+                int endIndex = record.indexOf('=');
+                if (endIndex == -1) {
+                    throw new RuntimeException("Failed to parse Map from String");
+                }
+                String key = record.substring(0, endIndex);
+                String value = record.substring(endIndex + 1);
+                additionalProps.put(key.trim(), value.trim());
+            }
+        }
+
         if ((config.getOauthAccessToken() != null)
                 || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthRefreshToken() != null)
                 || (config.getOauthTokenEndpointUri() != null && config.getOauthClientId() != null && config.getOauthClientSecret() != null))    {
+            log.info("Configuring OAuth");
             props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL".equals(props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)) ? "SASL_SSL" : "SASL_PLAINTEXT");
             props.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");
-            props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler");
+            if (!(additionalProps.containsKey(SaslConfigs.SASL_MECHANISM) && additionalProps.getProperty(SaslConfigs.SASL_MECHANISM).equals("PLAIN"))) {
+                props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, config.saslLoginCallbackClass);
+            }
         }
 
+        // override properties with defined additional properties
+        props.putAll(additionalProps);
+
         return props;
     }
     public String getBootstrapServers() {
diff --git a/java/kafka/streams/src/main/java/KafkaStreamsConfig.java b/java/kafka/streams/src/main/java/KafkaStreamsConfig.java
