Merge pull request #33 from b3rnh8rd/develop

added saslClientCallbackHandlerClass Property to enable AWS MSK IAM Auth
patschuh · Feb 6, 2023 · 5874745 · 5874745
2 parents 80c06dc + 717ceb9
commit 5874745
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 2 deletions.
diff --git a/build.gradle b/build.gradle
@@ -52,6 +52,7 @@ dependencies {
     implementation 'com.squareup.okhttp3:okhttp:4.9.3'
     implementation 'com.flipkart.zjsonpatch:zjsonpatch:0.4.12'
     implementation 'tech.allegro.schema.json2avro:converter:0.2.15'
+    implementation 'software.amazon.msk:aws-msk-iam-auth:1.1.5'
 
     testImplementation 'junit:junit:4.13.2'
     testImplementation 'org.springframework.kafka:spring-kafka-test:2.4.13.RELEASE'

diff --git a/src/main/java/at/esque/kafka/cluster/ClusterConfig.java b/src/main/java/at/esque/kafka/cluster/ClusterConfig.java
@@ -23,6 +23,7 @@ public class ClusterConfig {
     private StringProperty trustStorePassword = new SimpleStringProperty();
     private StringProperty saslSecurityProtocol = new SimpleStringProperty();
     private StringProperty saslMechanism = new SimpleStringProperty();
+    private StringProperty saslClientCallbackHandlerClass = new SimpleStringProperty();
     private StringProperty saslJaasConfig = new SimpleStringProperty();
     private StringProperty kafkaConnectUrl = new SimpleStringProperty();
     private StringProperty kafkaConnectBasicAuthUser = new SimpleStringProperty();
@@ -52,6 +53,7 @@ public void update(ClusterConfig existingConfig) {
             this.setTrustStorePassword(existingConfig.getTrustStorePassword());
             this.setSaslSecurityProtocol(existingConfig.getSaslSecurityProtocol());
             this.setSaslMechanism(existingConfig.getSaslMechanism());
+            this.setSaslClientCallbackHandlerClass(existingConfig.getSaslClientCallbackHandlerClass());
             this.setSaslJaasConfig(existingConfig.getSaslJaasConfig());
             this.setkafkaConnectUrl(existingConfig.getkafkaConnectUrl());
             this.setkafkaConnectBasicAuthUser(existingConfig.getkafkaConnectBasicAuthUser());
@@ -204,6 +206,19 @@ public void setSaslMechanism(String saslMechanism) {
         this.saslMechanism.set(saslMechanism);
     }
 
+    @JsonProperty("saslClientCallbackHandlerClass")
+    public String getSaslClientCallbackHandlerClass() {
+        return saslClientCallbackHandlerClass.get();
+    }
+
+    public StringProperty saslClientCallbackHandlerClassProperty() {
+        return saslClientCallbackHandlerClass;
+    }
+
+    public void setSaslClientCallbackHandlerClass(String saslClientCallbackHandlerClass) {
+        this.saslClientCallbackHandlerClass.set(saslClientCallbackHandlerClass);
+    }
+
     @JsonProperty("saslJaasConfig")
     public String getSaslJaasConfig() {
         return saslJaasConfig.get();

diff --git a/src/main/java/at/esque/kafka/dialogs/ClusterConfigDialog.java b/src/main/java/at/esque/kafka/dialogs/ClusterConfigDialog.java
@@ -29,6 +29,7 @@ public class ClusterConfigDialog {
     public static final String LABEL_SASL_MECHANISM = "SASL Mechanism";
     public static final String LABEL_SASL_JAAS_CONFIG = "SASL JAAS Config";
     public static final String LABEL_SASL_SECURITY_PROTOCOL = "SASL Security Protocol";
+    public static final String LABEL_SASL_CLIENT_CALLBACK_HANDLER_CLASS = "SASL Client Callback Handler Class";
     public static final String LABEL_KAFKA_CONNECT_URL = "Kafka Connect URL";
     public static final String LABEL_KAFKA_CONNECT_BASIC_AUTH_USER = "Kafka Connect Basic Auth User";
     public static final String LABEL_KAFKA_CONNECT_BASIC_AUTH_PASSWORD = "Kafka Connect Basic Auth Password";
@@ -163,7 +164,14 @@ public static Optional<ClusterConfig> show(ClusterConfig existingConfig) {
                                 .tooltip(LABEL_SASL_JAAS_CONFIG)
                                 .placeholder(LABEL_SASL_JAAS_CONFIG)
                                 .format(new  NullFormatStringConverter())
-                                .bind(copy.saslJaasConfigProperty())
+                                .bind(copy.saslJaasConfigProperty()),
+                        Field.ofStringType(copy.getSaslClientCallbackHandlerClass()==null?"":copy.getSaslClientCallbackHandlerClass())
+                                .label(LABEL_SASL_CLIENT_CALLBACK_HANDLER_CLASS)
+                                .tooltip(LABEL_SASL_CLIENT_CALLBACK_HANDLER_CLASS)
+                                .placeholder(LABEL_SASL_CLIENT_CALLBACK_HANDLER_CLASS)
+                                .valueDescription(String.format("Is used f.e. %s=AWS_MSK_IAM, %s=software.amazon.msk.auth.iam.IAMClientCallbackHandler", LABEL_SASL_MECHANISM,LABEL_SASL_CLIENT_CALLBACK_HANDLER_CLASS))
+                                .format(new  NullFormatStringConverter())
+                                .bind(copy.saslClientCallbackHandlerClassProperty())
                 )
         ).title(isCreatingNew ? "Add new Kafka Cluster" : "Change Kafka Cluster")
                 .binding(BindingMode.CONTINUOUS);

diff --git a/src/main/java/at/esque/kafka/handlers/ConfigHandler.java b/src/main/java/at/esque/kafka/handlers/ConfigHandler.java
@@ -304,11 +304,14 @@ public Map<String, String> getSaslProperties(ClusterConfig config) {
             props.put(SaslConfigs.SASL_MECHANISM, config.getSaslMechanism());
         }
 
-
         if (StringUtils.isNotEmpty(config.getSaslJaasConfig())) {
             props.put(SaslConfigs.SASL_JAAS_CONFIG, config.getSaslJaasConfig());
         }
 
+        if (StringUtils.isNotEmpty(config.getSaslClientCallbackHandlerClass())) {
+            props.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, config.getSaslClientCallbackHandlerClass());
+        }
+
         return props;
     }