fix(analytics): Support sasl authentication to kafka (datahub-project#2675)

Author: thomasplarsson

datahub-frontend/app/react/controllers/TrackingController.java

@@ -6,6 +6,7 @@
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import react.auth.Authenticator;
@@ -17,10 +18,16 @@
 import play.mvc.Security;
 import react.graphql.PlayQueryContext;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.Properties;
 
 public class TrackingController extends Controller {
 
+    private static final List<String> KAFKA_SSL_PROTOCOLS = Collections.unmodifiableList(
+            Arrays.asList(SecurityProtocol.SSL.name(),SecurityProtocol.SASL_SSL.name()));
+
     private final Boolean _isEnabled;
     private final Config _config;
     private final KafkaProducer<String, String> _producer;
@@ -81,7 +88,7 @@ private KafkaProducer createKafkaProducer() {
 
         final String securityProtocolConfig = "analytics.kafka.security.protocol";
         if (_config.hasPath(securityProtocolConfig)
-            && _config.getString(securityProtocolConfig).equals(SecurityProtocol.SSL)) {
+                && KAFKA_SSL_PROTOCOLS.contains(_config.getString(securityProtocolConfig))) {
             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, _config.getString(securityProtocolConfig));
             props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, _config.getString("analytics.kafka.ssl.key.password"));
 
@@ -95,6 +102,11 @@ private KafkaProducer createKafkaProducer() {
 
             props.put(SslConfigs.SSL_PROTOCOL_CONFIG, _config.getString("analytics.kafka.ssl.protocol"));
             props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, _config.getString("analytics.kafka.ssl.endpoint.identification.algorithm"));
+
+            if (_config.getString(securityProtocolConfig).equals(SecurityProtocol.SASL_SSL.name())) {
+                props.put(SaslConfigs.SASL_MECHANISM, _config.getString("analytics.kafka.sasl.mechanism"));
+                props.put(SaslConfigs.SASL_JAAS_CONFIG, _config.getString("analytics.kafka.sasl.jaas.config"));
+            }
         }
 
         return new KafkaProducer(props);

datahub-frontend/conf/application.conf

@@ -154,6 +154,9 @@ analytics.kafka.ssl.truststore.location = ${?KAFKA_PROPERTIES_SSL_TRUSTSTORE_LOC
 analytics.kafka.ssl.truststore.password = ${?KAFKA_PROPERTIES_SSL_TRUSTSTORE_PASSWORD}
 analytics.kafka.ssl.protocol = ${?KAFKA_PROPERTIES_SSL_PROTOCOL}
 analytics.kafka.ssl.endpoint.identification.algorithm = ${?KAFKA_PROPERTIES_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM}
+# If analytics.kafka.security.protocol is set to "SASL_SSL", both of these need to be set.
+analytics.kafka.sasl.mechanism = ${?KAFKA_PROPERTIES_SASL_MECHANISM}
+analytics.kafka.sasl.jaas.config = ${?KAFKA_PROPERTIES_SASL_JAAS_CONFIG}
 
 # Required Elastic Client Configuration
 analytics.elastic.host =  ${ELASTIC_CLIENT_HOST}