gerrit.onap Code Review - policy/common.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2022-2023 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  * ============LICENSE_END=========================================================
  17  */
  18
  19 package org.onap.policy.common.endpoints.event.comm.bus.internal;
  20
  21 import java.net.MalformedURLException;
  22 import java.util.Map;
  23 import org.onap.policy.common.endpoints.event.comm.Topic;
  24 import org.onap.policy.common.endpoints.event.comm.bus.KafkaTopicSource;
  25
  26 /**
  27  * This topic source implementation specializes in reading messages over a Kafka Bus topic source and
  28  * notifying its listeners.
  29  */
  30 public class SingleThreadedKafkaTopicSource extends SingleThreadedBusTopicSource implements KafkaTopicSource {
  31
  32     protected Map<String, String> additionalProps = null;
  33
  34     /**
  35      * Constructor.
  36      *
  37      * @param busTopicParams Parameters object containing all the required inputs
  38      * @throws IllegalArgumentException An invalid parameter passed in
  39      */
  40     public SingleThreadedKafkaTopicSource(BusTopicParams busTopicParams) {
  41         super(busTopicParams);
  42         this.additionalProps = busTopicParams.getAdditionalProps();
  43         try {
  44             this.init();
  45         } catch (Exception e) {
  46             throw new IllegalArgumentException("ERROR during init in kafka-source: cannot create topic " + topic, e);
  47         }
  48     }
  49
  50     /**
  51      * Initialize the Cambria client.
  52      */
  53     @Override
  54     public void init() throws MalformedURLException {
  55         BusTopicParams.TopicParamsBuilder builder = BusTopicParams.builder()
  56                 .servers(this.servers)
  57                 .topic(this.effectiveTopic)
  58                 .fetchTimeout(this.fetchTimeout)
  59                 .consumerGroup(this.consumerGroup)
  60                 .useHttps(this.useHttps)
  61                 .allowTracing(this.allowTracing);
  62
  63         this.consumer = new BusConsumer.KafkaConsumerWrapper(builder
  64                         .additionalProps(this.additionalProps)
  65                         .build());
  66     }
  67
  68     @Override
  69     public CommInfrastructure getTopicCommInfrastructure() {
  70         return Topic.CommInfrastructure.KAFKA;
  71     }
  72
  73     @Override
  74     public String toString() {
  75         return "SingleThreadedKafkaTopicSource [getTopicCommInfrastructure()=" + getTopicCommInfrastructure()
  76             + ", toString()=" + super.toString() + "]";
  77     }
  78
  79 }