aboutsummaryrefslogtreecommitdiffstats
path: root/common/src/main
diff options
context:
space:
mode:
Diffstat (limited to 'common/src/main')
-rw-r--r--common/src/main/java/org/onap/so/client/kafka/KafkaClient.java21
-rw-r--r--common/src/main/java/org/onap/so/client/kafka/KafkaConsumerImpl.java104
-rw-r--r--common/src/main/resources/kafka/default-consumer.properties6
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