aDD SASL class callback config for producer and consumer

goto · Sep 2, 2024 · 1f6d09c · 1f6d09c
1 parent 328c44a
commit 1f6d09c
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 0 deletions.
diff --git a/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/SinkOrchestrator.java b/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/SinkOrchestrator.java
@@ -13,6 +13,7 @@
 import org.apache.flink.connector.kafka.sink.KafkaSink;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase;
 import org.apache.flink.types.Row;
+import org.apache.commons.lang.StringUtils;
 
 import com.gotocompany.dagger.common.configuration.Configuration;
 import com.gotocompany.dagger.common.core.StencilClientOrchestrator;
@@ -106,6 +107,12 @@ protected Properties getProducerProperties(Configuration configuration) {
             kafkaProducerConfigs.setProperty(Constants.SINK_KAFKA_COMPRESSION_TYPE_KEY, Constants.SINK_KAFKA_COMPRESSION_TYPE_DEFAULT);
             kafkaProducerConfigs.setProperty(Constants.SINK_KAFKA_MAX_REQUEST_SIZE_KEY, Constants.SINK_KAFKA_MAX_REQUEST_SIZE_DEFAULT);
         }
+        if (StringUtils.isNotEmpty(configuration.getString(Constants.SINK_KAFKA_SASL_LOGIN_CALLBACK_HANDLER_CLASS_KEY, StringUtils.EMPTY))) {
+            kafkaProducerConfigs.setProperty(Constants.SASL_LOGIN_CALLBACK_HANDLER_CLASS, configuration.getString(Constants.SINK_KAFKA_SASL_LOGIN_CALLBACK_HANDLER_CLASS_KEY, StringUtils.EMPTY));
+        }
+        if (StringUtils.isNotEmpty(configuration.getString(Constants.SINK_KAFKA_SASL_JAAS_CONFIG_KEY, StringUtils.EMPTY))) {
+            kafkaProducerConfigs.setProperty(Constants.SASL_JAAS_CONFIG, configuration.getString(Constants.SINK_KAFKA_SASL_JAAS_CONFIG_KEY, StringUtils.EMPTY));
+        }
         String lingerMs = configuration.getString(Constants.SINK_KAFKA_LINGER_MS_KEY, Constants.SINK_KAFKA_LINGER_MS_DEFAULT);
         validateLingerMs(lingerMs);
         kafkaProducerConfigs.setProperty(Constants.SINK_KAFKA_LINGER_MS_CONFIG_KEY, lingerMs);

diff --git a/dagger-core/src/main/java/com/gotocompany/dagger/core/source/config/StreamConfig.java b/dagger-core/src/main/java/com/gotocompany/dagger/core/source/config/StreamConfig.java
@@ -34,6 +34,7 @@
 import static com.gotocompany.dagger.common.core.Constants.STREAM_INPUT_SCHEMA_PROTO_CLASS;
 import static com.gotocompany.dagger.common.core.Constants.STREAM_INPUT_SCHEMA_TABLE;
 import static com.gotocompany.dagger.core.utils.Constants.*;
+import static com.gotocompany.dagger.core.utils.Constants.SOURCE_KAFKA_CONSUMER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS_CONFIG_KEY;
 import static com.gotocompany.dagger.core.utils.Constants.STREAM_SOURCE_PARQUET_FILE_DATE_RANGE_KEY;
 
 public class StreamConfig {
@@ -120,6 +121,10 @@ public class StreamConfig {
     @Getter
     private String saslJaasConfig;
 
+    @SerializedName(SOURCE_KAFKA_CONSUMER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS_CONFIG_KEY)
+    @Getter
+    private String saslLoginCallbackHandlerClassConfig;
+
     @SerializedName(STREAM_INPUT_STREAM_NAME_KEY)
     @Getter
     private String kafkaName;

diff --git a/dagger-core/src/main/java/com/gotocompany/dagger/core/utils/Constants.java b/dagger-core/src/main/java/com/gotocompany/dagger/core/utils/Constants.java
@@ -82,6 +82,11 @@ public class Constants {
     public static final String SINK_KAFKA_MAX_REQUEST_SIZE_KEY = "max.request.size";
     public static final String SINK_KAFKA_MAX_REQUEST_SIZE_DEFAULT = "20971520";
     public static final String SINK_KAFKA_LINGER_MS_DEFAULT = "0";
+    public static final String SINK_KAFKA_SASL_LOGIN_CALLBACK_HANDLER_CLASS_KEY= "SINK_KAFKA_SASL_LOGIN_CALLBACK_HANDLER_CLASS";
+    public static final String SINK_KAFKA_SASL_JAAS_CONFIG_KEY = "SINK_KAFKA_SASL_JAAS_CONFIG";
+
+    public static final String SASL_LOGIN_CALLBACK_HANDLER_CLASS= "sasl.login.callback.handler.class";
+    public static final String SASL_JAAS_CONFIG = "sasl.jaas.config";
 
     public static final String ES_TYPE = "ES";
     public static final String HTTP_TYPE = "HTTP";
@@ -129,6 +134,7 @@ public class Constants {
     public static final String SOURCE_KAFKA_CONSUMER_CONFIG_SECURITY_PROTOCOL_KEY = "SOURCE_KAFKA_CONSUMER_CONFIG_SECURITY_PROTOCOL";
     public static final String SOURCE_KAFKA_CONSUMER_CONFIG_SASL_MECHANISM_KEY = "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_MECHANISM";
     public static final String SOURCE_KAFKA_CONSUMER_CONFIG_SASL_JAAS_CONFIG_KEY = "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_JAAS_CONFIG";
+    public static final String SOURCE_KAFKA_CONSUMER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS_CONFIG_KEY = "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS";
 
     public static final String SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEY_PASSWORD_KEY = "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEY_PASSWORD";
     public static final String SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEYSTORE_LOCATION_KEY = "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEYSTORE_LOCATION";