Added support for using oauth token for Kafka

[nonrtric/plt/ranpm.git] / influxlogger / src / main / java / org / oran / pmlog / KafkaTopicListener.java

diff --git a/influxlogger/src/main/java/org/oran/pmlog/KafkaTopicListener.java b/influxlogger/src/main/java/org/oran/pmlog/KafkaTopicListener.java
index 390ccae..1617f1d 100644 (file)
--- a/influxlogger/src/main/java/org/oran/pmlog/KafkaTopicListener.java
+++ b/influxlogger/src/main/java/org/oran/pmlog/KafkaTopicListener.java
@@ -89,6 +89,7 @@ public class KafkaTopicListener {
         consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
 
         consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId + "_" + applicationConfig.getKafkaGroupId());
+        this.applicationConfig.addKafkaSecurityProps(consumerProps);
 
         return ReceiverOptions.<byte[], byte[]>create(consumerProps)
                 .subscription(Collections.singleton(this.applicationConfig.getKafkaInputTopic()));