catalog-be/src/main/java/org/openecomp/sdc/be/components/kafka/SdcKafkaConsumer.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * SDC
   4  * ================================================================================
   5  * Copyright (C) 2022 Nordix Foundation. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20 package org.openecomp.sdc.be.components.kafka;
  21
  22 import com.google.common.annotations.VisibleForTesting;
  23 import java.time.Duration;
  24 import java.util.ArrayList;
  25 import java.util.Collections;
  26 import java.util.List;
  27 import java.util.Properties;
  28 import org.apache.kafka.clients.consumer.ConsumerRecord;
  29 import org.apache.kafka.clients.consumer.ConsumerRecords;
  30 import org.apache.kafka.clients.consumer.KafkaConsumer;
  31 import org.apache.kafka.common.KafkaException;
  32 import org.openecomp.sdc.be.config.DistributionEngineConfiguration;
  33 import org.openecomp.sdc.common.log.wrappers.Logger;
  34
  35 /**
  36  * Utility class that provides a KafkaConsumer to communicate with a kafka cluster
  37  */
  38 public class SdcKafkaConsumer {
  39
  40     private static final Logger log = Logger.getLogger(SdcKafkaConsumer.class.getName());
  41     private final DistributionEngineConfiguration deConfiguration;
  42     private final KafkaConsumer<String, String> kafkaConsumer;
  43
  44     /**
  45      * Constructor setting up the KafkaConsumer from a predefined set of configurations
  46      */
  47     public SdcKafkaConsumer(DistributionEngineConfiguration deConfiguration){
  48         log.info("Create SdcKafkaConsumer via constructor");
  49         KafkaCommonConfig kafkaCommonConfig = new KafkaCommonConfig(deConfiguration);
  50         Properties properties = kafkaCommonConfig.getConsumerProperties();
  51         this.deConfiguration = deConfiguration;
  52         kafkaConsumer = new KafkaConsumer<>(properties);
  53     }
  54
  55     /**
  56      *
  57      * @param kafkaConsumer a kafkaConsumer to use within the class
  58      * @param deConfiguration - Configuration to pass into the class
  59      */
  60     @VisibleForTesting
  61     SdcKafkaConsumer(KafkaConsumer kafkaConsumer, DistributionEngineConfiguration deConfiguration){
  62         this.deConfiguration = deConfiguration;
  63         this.kafkaConsumer = kafkaConsumer;
  64     }
  65
  66     /**
  67      *
  68      * @param topic Topic in which to subscribe
  69      */
  70     public void subscribe(String topic) throws KafkaException {
  71         if (!kafkaConsumer.subscription().contains(topic)) {
  72             kafkaConsumer.subscribe(Collections.singleton(topic));
  73         }
  74     }
  75
  76     /**
  77      *
  78      * @return The list of messages for a specified topic, returned from the poll
  79      */
  80     public List<String> poll(String topicName) throws KafkaException {
  81         log.info("SdcKafkaConsumer - polling for messages from Topic: {}", topicName);
  82         List<String> msgs = new ArrayList<>();
  83         ConsumerRecords<String, String> consumerRecordsForSpecificTopic = kafkaConsumer.poll(Duration.ofSeconds(deConfiguration.getDistributionStatusTopic().getPollingIntervalSec()));
  84         for(ConsumerRecord<String, String> rec : consumerRecordsForSpecificTopic){
  85             msgs.add(rec.value());
  86         }
  87         return msgs;
  88     }
  89 }