sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/kafka/VESMsgKafkaConsumer.java

   1 package org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.kafka;
   2
   3 import java.time.Duration;
   4 import java.util.ArrayList;
   5 import java.util.Collections;
   6 import java.util.List;
   7 import java.util.Properties;
   8 import org.apache.kafka.clients.CommonClientConfigs;
   9 import org.apache.kafka.clients.consumer.ConsumerConfig;
  10 import org.apache.kafka.clients.consumer.ConsumerRecord;
  11 import org.apache.kafka.clients.consumer.ConsumerRecords;
  12 import org.apache.kafka.clients.consumer.KafkaConsumer;
  13 import org.apache.kafka.common.config.SaslConfigs;
  14 import org.apache.kafka.common.errors.InvalidGroupIdException;
  15 import org.slf4j.Logger;
  16 import org.slf4j.LoggerFactory;
  17
  18 /**
  19  * Utility class that provides a KafkaConsumer to communicate with a kafka cluster
  20  */
  21 public class VESMsgKafkaConsumer {
  22
  23     private static final Logger log = LoggerFactory.getLogger(VESMsgKafkaConsumer.class);
  24     final KafkaConsumer<String, String> consumer;
  25     private final int pollTimeout;
  26     private String topicName;
  27     private static final String DESERIALIZER_CLASS = "org.apache.kafka.common.serialization.StringDeserializer";
  28
  29     /**
  30      *
  31      * @param consumerProperties
  32      * @param configuration The config provided to the client
  33      */
  34     public VESMsgKafkaConsumer(Properties strimziKafkaProperties, Properties consumerProperties) {
  35         Properties props = new Properties();
  36         props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, strimziKafkaProperties.getProperty("bootstrap.servers"));
  37         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, strimziKafkaProperties.getProperty("security.protocol"));
  38         props.put(SaslConfigs.SASL_MECHANISM, strimziKafkaProperties.getProperty("sasl.mechanism"));
  39         props.put(SaslConfigs.SASL_JAAS_CONFIG, strimziKafkaProperties.getProperty("sasl.jaas.config"));
  40         props.put(ConsumerConfig.GROUP_ID_CONFIG,
  41                 consumerProperties.getProperty("consumerGroup") + "-" + consumerProperties.getProperty("topic"));
  42         props.put(ConsumerConfig.CLIENT_ID_CONFIG,
  43                 consumerProperties.getProperty("topic") + "-" + consumerProperties.getProperty("consumerID"));
  44         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  45         props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
  46         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, DESERIALIZER_CLASS);
  47         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DESERIALIZER_CLASS);
  48         consumer = new KafkaConsumer<>(props);
  49         pollTimeout = Integer.parseInt(consumerProperties.getProperty("timeout"));
  50     }
  51
  52     /**
  53      *
  54      * @param topic The kafka topic to subscribe to
  55      */
  56     public void subscribe(String topic) {
  57         try {
  58             consumer.subscribe(Collections.singleton(topic));
  59             this.topicName = topic;
  60         } catch (InvalidGroupIdException e) {
  61             log.error("Invalid Group {}", e.getMessage());
  62         }
  63     }
  64
  65     /**
  66      *
  67      * @return The list of records returned from the poll
  68      */
  69     public List<String> poll() {
  70         List<String> msgs = new ArrayList<>();
  71         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(pollTimeout));
  72         for (ConsumerRecord<String, String> rec : records) {
  73             msgs.add(rec.value());
  74         }
  75         return msgs;
  76     }
  77
  78     public String getTopicName() {
  79         return topicName;
  80     }
  81
  82     public void stop() {
  83         consumer.unsubscribe();
  84         consumer.close();
  85     }
  86 }