Merge pull request #2281 from mgotz/kafka-ssl

Additional options for kafka clients

Author: shroffk

app/alarm/datasource/src/main/java/org/phoebus/pv/alarm/AlarmContext.java

@@ -34,7 +34,7 @@ private static synchronized void initializeAlarmClient(String config)
         if (!alarmModels.containsKey(config))
         {
             logger.log(Level.CONFIG, "Creating a alarm client for config : + " + config + " in the alarm datasource");
-            AlarmClient model = new AlarmClient(AlarmSystem.server, config);
+            AlarmClient model = new AlarmClient(AlarmSystem.server, config, AlarmSystem.kafka_properties);
             model.addListener(new AlarmClientDatasourceListener(config));
             model.start();
             alarmModels.put(config, model);

app/alarm/model/src/main/java/org/phoebus/applications/alarm/AlarmSystem.java

@@ -63,6 +63,9 @@ public class AlarmSystem
     /** Kafka Server host:port */
     @Preference public static String server;
 
+    /** Kafka settings file */
+    @Preference public static String kafka_properties;
+
     /** Name of alarm tree root
      *
      *  <p>Default name from preferences.

app/alarm/model/src/main/java/org/phoebus/applications/alarm/client/AlarmClient.java

@@ -94,8 +94,9 @@ public class AlarmClient
 
     /** @param server Kafka Server host:port
      *  @param config_name Name of alarm tree root
+     *  @param kafka_properties_file File to load additional kafka properties from
      */
-    public AlarmClient(final String server, final String config_name)
+    public AlarmClient(final String server, final String config_name, final String kafka_properties_file)
     {
         Objects.requireNonNull(server);
         Objects.requireNonNull(config_name);
@@ -105,8 +106,8 @@ public AlarmClient(final String server, final String config_name)
 
         root = new AlarmClientNode(null, config_name);
         final List<String> topics = List.of(config_topic);
-        consumer = KafkaHelper.connectConsumer(server, topics, topics);
-        producer = KafkaHelper.connectProducer(server);
+        consumer = KafkaHelper.connectConsumer(server, topics, topics, kafka_properties_file);
+        producer = KafkaHelper.connectProducer(server, kafka_properties_file);
 
         thread = new Thread(this::run, "AlarmClientModel " + config_name);
         thread.setDaemon(true);

app/alarm/model/src/main/java/org/phoebus/applications/alarm/client/KafkaHelper.java

@@ -9,10 +9,13 @@
 
 import static org.phoebus.applications.alarm.AlarmSystem.logger;
 
+import java.io.FileInputStream;
+import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
 import java.util.Properties;
 import java.util.UUID;
+import java.util.logging.Level;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -51,24 +54,28 @@ public class KafkaHelper
      *  @param kafka_servers Servers to read
      *  @param topics Topics to which to subscribe
      *  @param from_beginning Topics to read from the beginning
+     *  @param properties_file File name to load additional settings for the kafka consumer
      *  @return {@link Consumer}
      */
-    public static Consumer<String, String> connectConsumer(final String kafka_servers, final List<String> topics, final List<String> from_beginning)
+    public static Consumer<String, String> connectConsumer(final String kafka_servers, final List<String> topics, final List<String> from_beginning, final String properties_file)
     {
-        final Properties props = new Properties();
-        props.put("bootstrap.servers", kafka_servers);
-        // API requires for Consumer to be in a group.
-        // Each alarm client must receive all updates,
-        // cannot balance updates across a group
-        // --> Use unique group for each client
-        final String group_id = "Alarm-" + UUID.randomUUID();
-        props.put("group.id", group_id);
+        Properties kafka_props = loadPropsFromFile(properties_file);
+        kafka_props.put("bootstrap.servers", kafka_servers);
 
-        logger.fine(group_id + " subscribes to " + kafka_servers + " for " + topics);
+        if (!kafka_props.containsKey("group.id")){
+            // API requires for Consumer to be in a group.
+            // Each alarm client must receive all updates,
+            // cannot balance updates across a group
+            // --> Use unique group for each client
+            final String group_id = "Alarm-" + UUID.randomUUID();
+            kafka_props.put("group.id", group_id);
+        }
+
+        logger.fine(kafka_props.getProperty("group.id") + " subscribes to " + kafka_servers + " for " + topics);
 
         // Read key, value as string
         final Deserializer<String> deserializer = new StringDeserializer();
-        final Consumer<String, String> consumer = new KafkaConsumer<>(props, deserializer, deserializer);
+        final Consumer<String, String> consumer = new KafkaConsumer<>(kafka_props, deserializer, deserializer);
 
         // Rewind whenever assigned to partition
         final ConsumerRebalanceListener crl = new ConsumerRebalanceListener()
@@ -101,19 +108,20 @@ public void onPartitionsRevoked(final Collection<TopicPartition> parts)
 
     /** Create producer for alarm information
      *  @param kafka_servers Kafka servers
+     *  @param properties_file File name to load additional settings for the kafka producer
      *  @return {@link Producer}
      */
-    public static Producer<String, String> connectProducer(final String kafka_servers)
+    public static Producer<String, String> connectProducer(final String kafka_servers, final String properties_file)
     {
-        final Properties props = new Properties();
-        props.put("bootstrap.servers", kafka_servers);
+        Properties kafka_props = loadPropsFromFile(properties_file);
+        kafka_props.put("bootstrap.servers", kafka_servers);
         // Collect messages for 20ms until sending them out as a batch
-        props.put("linger.ms", 20);
+        kafka_props.put("linger.ms", 20);
 
         // Write String key, value
         final Serializer<String> serializer = new StringSerializer();
 
-        final Producer<String, String> producer = new KafkaProducer<>(props, serializer, serializer);
+        final Producer<String, String> producer = new KafkaProducer<>(kafka_props, serializer, serializer);
 
         return producer;
     }
@@ -123,22 +131,43 @@ public static Producer<String, String> connectProducer(final String kafka_server
      * @param kafka_servers - Sever to connect to.
      * @param topics List of topics to aggregate.
      * @param aggregate_topic - Name of topic to aggregate to.
+     * @param kafka_props File name to load additional settings for the kafka stream
      * @return aggregate_stream - KafkaStreams
      * @author Evan Smith
      */
-    public static KafkaStreams aggregateTopics(String kafka_servers, List<String> topics, String aggregate_topic)
+    public static KafkaStreams aggregateTopics(String kafka_servers, List<String> topics, String aggregate_topic, final String properties_file)
     {
-        final Properties props = new Properties();
-        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "Stream-To-Long-Term");
-        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka_servers);
-        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
-        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+        Properties kafka_props = loadPropsFromFile(properties_file);
+        kafka_props.put(StreamsConfig.APPLICATION_ID_CONFIG, "Stream-To-Long-Term");
+        kafka_props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka_servers);
+        kafka_props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+        kafka_props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
 
         final StreamsBuilder builder = new StreamsBuilder();
 
         // Aggregate the topics by mapping the topic key value pairs one to one into the aggregate topic.
         builder.<String, String>stream(topics).mapValues(pair -> pair).to(aggregate_topic);
 
-        return new KafkaStreams(builder.build(), props);
+        return new KafkaStreams(builder.build(), kafka_props);
+    }
+
+
+    /**
+     * Load properties from the given file path. Path may be blank or null
+     * resulting in a properties object without entries.
+     * @param filePath Full path to properties file
+     * @return properties - the properties loaded from file
+     */
+    static public Properties loadPropsFromFile(String filePath) {
+        logger.fine("loading file from path: " + filePath);
+        Properties properties = new Properties();
+        if(filePath != null && !filePath.isBlank()){
+            try(FileInputStream file = new FileInputStream(filePath);){
+                properties.load(file);
+            } catch(IOException e) {
+                logger.log(Level.SEVERE, "failed to load kafka properties", e);
+            }
+        }
+        return properties;
     }
 }

app/alarm/model/src/main/java/org/phoebus/applications/alarm/talk/TalkClient.java

@@ -52,7 +52,7 @@ public TalkClient(final String server, final String config_name)
         Objects.requireNonNull(config_name);
 
         final List<String> topics = List.of(config_name + AlarmSystem.TALK_TOPIC_SUFFIX);
-        consumer = KafkaHelper.connectConsumer(server, topics, Collections.emptyList());
+        consumer = KafkaHelper.connectConsumer(server, topics, Collections.emptyList(), AlarmSystem.kafka_properties);
 
         thread = new Thread(this::run, "TalkClient");
         thread.setDaemon(true);

app/alarm/model/src/main/resources/alarm_preferences.properties

@@ -5,6 +5,9 @@
 # Kafka Server host:port
 server=localhost:9092
 
+# A file to configure the properites of kafka clients
+kafka_properties=
+
 # Name of alarm tree root
 config_name=Accelerator
 

app/alarm/model/src/test/java/org/phoebus/applications/alarm/AlarmClientModelDemo.java

@@ -24,7 +24,7 @@ public class AlarmClientModelDemo
     @Test
     public void testClientModel() throws Exception
     {
-        final AlarmClient client = new AlarmClient(AlarmDemoSettings.SERVERS, AlarmDemoSettings.ROOT);
+        final AlarmClient client = new AlarmClient(AlarmDemoSettings.SERVERS, AlarmDemoSettings.ROOT, AlarmDemoSettings.KAFKA_PROPERTIES_FILE);
         client.start();
         TimeUnit.SECONDS.sleep(4);
 

app/alarm/model/src/test/java/org/phoebus/applications/alarm/AlarmDemoSettings.java

@@ -15,4 +15,5 @@ public class AlarmDemoSettings
 {
     public static final String SERVERS = "localhost:9092";
     public static final String ROOT = "Accelerator";
+    public static final String KAFKA_PROPERTIES_FILE = "";
 }

app/alarm/model/src/test/java/org/phoebus/applications/alarm/AlarmModelSnapshotDemo.java

@@ -24,7 +24,7 @@ public class AlarmModelSnapshotDemo
 	public void testAlarmModelWriter() throws Exception
 	{
 	    // Get alarm configuration
-	    final AlarmClient client = new AlarmClient(AlarmDemoSettings.SERVERS, AlarmDemoSettings.ROOT);
+	    final AlarmClient client = new AlarmClient(AlarmDemoSettings.SERVERS, AlarmDemoSettings.ROOT, AlarmDemoSettings.KAFKA_PROPERTIES_FILE);
 
 	    System.out.println("Wait for stable configuration, i.e. no changes for 4 seconds...");
 

app/alarm/model/src/test/java/org/phoebus/applications/alarm/EnabledDeserializerTest.java

@@ -1,4 +1,4 @@
-package src.test.java.org.phoebus.applications.alarm;
+package org.phoebus.applications.alarm;
 
 import org.junit.Test;
 import java.io.IOException;