diff options
Diffstat (limited to 'common/src/main')
3 files changed, 131 insertions, 0 deletions
diff --git a/common/src/main/java/org/onap/so/client/kafka/KafkaClient.java b/common/src/main/java/org/onap/so/client/kafka/KafkaClient.java new file mode 100644 index 0000000000..2c695255e0 --- /dev/null +++ b/common/src/main/java/org/onap/so/client/kafka/KafkaClient.java @@ -0,0 +1,21 @@ +package org.onap.so.client.kafka; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; +import java.io.IOException; +import java.util.Properties; + +public class KafkaClient { + protected static Logger logger = LoggerFactory.getLogger(KafkaClient.class); + protected final Properties properties; + + public KafkaClient(String filepath) throws IOException { + Resource resource = new ClassPathResource(filepath); + this.properties = new Properties(); + properties.load(resource.getInputStream()); + + } + +} diff --git a/common/src/main/java/org/onap/so/client/kafka/KafkaConsumerImpl.java b/common/src/main/java/org/onap/so/client/kafka/KafkaConsumerImpl.java new file mode 100644 index 0000000000..69dd16acf8 --- /dev/null +++ b/common/src/main/java/org/onap/so/client/kafka/KafkaConsumerImpl.java @@ -0,0 +1,104 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP - SO + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.so.client.kafka; + +import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.config.SaslConfigs; +import org.apache.kafka.common.security.auth.SecurityProtocol; +import org.apache.kafka.common.security.scram.internals.ScramMechanism; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Properties; + +public class KafkaConsumerImpl extends KafkaClient { + + protected static Logger logger = LoggerFactory.getLogger(KafkaConsumerImpl.class); + private static final String kafkaBootstrapServers = "kafkaBootstrapServers"; + private Consumer<String, String> consumer; + + public KafkaConsumerImpl(String bootstrapServers) throws Exception { + super("kafka/default-consumer.properties"); + setProperties(bootstrapServers); + } + + + public List<String> get(String topic, String consumerGroup, String consumerId) { + logger.info("consuming message from kafka topic : " + topic); + this.properties.put("group.id", consumerGroup); + this.properties.put("client.id", consumerId); + if (consumer == null) { + consumer = getKafkaConsumer(properties); + consumer.subscribe(Arrays.asList(topic)); + } + ArrayList<String> msgs = new ArrayList<>(); + ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); + for (ConsumerRecord<String, String> rec : records) { + msgs.add(rec.value()); + } + logger.info("<<<<<<<<<<<<<<<<<<<<<<<<<<<<READING THE CONSUMED MESSAGES<<<<<<<<<<<<<<<<<<<<<<<<<<<"); + msgs.forEach(msg -> logger.info("MESSAGE CONSUMED FROM KAFKA : <<<<<" + msg + ">>>>>")); + return msgs; + } + + private void setProperties(String bootstrapServers) throws Exception { + if (bootstrapServers == null) { + logger.error("Environment Variable " + kafkaBootstrapServers + " is missing"); + throw new Exception("Environment Variable " + kafkaBootstrapServers + " is missing"); + } else { + this.properties.put("bootstrap.servers", bootstrapServers); + } + + if (System.getenv("JAAS_CONFIG") == null) { + logger.info("Not using any authentication for kafka interaction"); + } else { + logger.info("Using {} authentication provided for kafka interaction", + ScramMechanism.SCRAM_SHA_512.mechanismName()); + this.properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name); + this.properties.put(SaslConfigs.SASL_MECHANISM, ScramMechanism.SCRAM_SHA_512.mechanismName()); + this.properties.put(SaslConfigs.SASL_JAAS_CONFIG, System.getenv("JAAS_CONFIG")); + } + } + + public static KafkaConsumer<String, String> getKafkaConsumer(Properties properties) { + return new KafkaConsumer<>(properties); + } + + public void setConsumer(Consumer<String, String> kafkaConsumer) { + this.consumer = kafkaConsumer; + } + + public void close() { + if (consumer != null) { + logger.info("Closing the Kafka Consumer"); + consumer.close(); + consumer = null; + } + } + +} diff --git a/common/src/main/resources/kafka/default-consumer.properties b/common/src/main/resources/kafka/default-consumer.properties new file mode 100644 index 0000000000..a7edf58b6b --- /dev/null +++ b/common/src/main/resources/kafka/default-consumer.properties @@ -0,0 +1,6 @@ +key.deserializer=org.apache.kafka.common.serialization.StringDeserializer +value.deserializer=org.apache.kafka.common.serialization.StringDeserializer +max.poll.interval.ms=300000 +heartbeat.interval.ms=60000 +session.timeout.ms=240000 +max.poll.records=1000 |