From 38f5b4b9dc667c52561867d4e36f940109f3e3a5 Mon Sep 17 00:00:00 2001
From: "david.mcweeney" <david.mcweeney@est.tech>
Date: Fri, 6 Jan 2023 11:15:15 +0000
Subject: [KAFKA] Adding new client code
Signed-off-by: david.mcweeney <david.mcweeney@est.tech>
Change-Id: I38b930b1b5f4233f961d51bbab4b1828d034e67a
Issue-ID: DMAAP-1847
---
.../org/onap/dmaap/kafka/OnapKafkaClientTest.java | 126 +++++++++++++++++++++
.../org/onap/dmaap/kafka/TestConfiguration.java | 110 ++++++++++++++++++
2 files changed, 236 insertions(+)
create mode 100644 kafkaClient/src/test/java/org/onap/dmaap/kafka/OnapKafkaClientTest.java
create mode 100644 kafkaClient/src/test/java/org/onap/dmaap/kafka/TestConfiguration.java
(limited to 'kafkaClient/src/test/java/org/onap')
diff --git a/kafkaClient/src/test/java/org/onap/dmaap/kafka/OnapKafkaClientTest.java b/kafkaClient/src/test/java/org/onap/dmaap/kafka/OnapKafkaClientTest.java
new file mode 100644
index 0000000..9708f3b
--- /dev/null
+++ b/kafkaClient/src/test/java/org/onap/dmaap/kafka/OnapKafkaClientTest.java
@@ -0,0 +1,126 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * dmaap-kafka-client
+ * ================================================================================
+ * Copyright (C) 2023 Nordix Foundation. 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.dmaap.kafka;
+
+import com.salesforce.kafka.test.KafkaTestCluster;
+import com.salesforce.kafka.test.KafkaTestUtils;
+import com.salesforce.kafka.test.listeners.BrokerListener;
+import com.salesforce.kafka.test.listeners.SaslPlainListener;
+import io.github.netmikey.logunit.api.LogCapturer;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class OnapKafkaClientTest {
+
+ @RegisterExtension
+ LogCapturer producerLogs = LogCapturer.create().captureForType(OnapKafkaProducer.class);
+
+ @RegisterExtension
+ LogCapturer clientLogs = LogCapturer.create().captureForType(OnapKafkaClient.class);
+
+ private static final Logger logger = LoggerFactory.getLogger(OnapKafkaClientTest.class);
+
+ private static TestConfiguration configuration = new TestConfiguration("application.properties");
+ private static final List<String> consumerTopics = configuration.getConsumerTopics();
+ private static KafkaTestCluster kafkaTestCluster = null;
+
+ @BeforeAll
+ static void before() throws Exception {
+ startKafkaService();
+ KafkaTestUtils utils = new KafkaTestUtils(kafkaTestCluster);
+ for (String topic: consumerTopics) {
+ utils.createTopic(topic, 1, (short) 1);
+ }
+ configuration.setBootstrapServers(Collections.singletonList(kafkaTestCluster.getKafkaConnectString()));
+ }
+
+ @AfterAll
+ static void after() throws Exception {
+ kafkaTestCluster.close();
+ kafkaTestCluster.stop();
+ }
+
+ @Test
+ void whenProducingCorrectRecordsArePresent() {
+ OnapKafkaClient handler = new OnapKafkaClient(configuration);
+ Assertions.assertEquals(handler.fetchFromTopic(consumerTopics.get(0)).size(), 0);
+ handler.publishToTopic(consumerTopics.get(0), "blahblahblahblah");
+ handler.publishToTopic(consumerTopics.get(1), "iaerugfoiaeurgfoaiuerf");
+ List<String> eventsFrom1 = handler.fetchFromTopic(consumerTopics.get(0));
+ Assertions.assertEquals(1, eventsFrom1.size());
+ handler.fetchFromTopic(consumerTopics.get(0));
+ List<String> events2 = handler.fetchFromTopic(consumerTopics.get(1));
+ Assertions.assertEquals( 0, events2.size());
+ }
+
+ @Test
+ void whenConsumingFromInvalidTopicEmptyListIsReturned() {
+ OnapKafkaClient handler = new OnapKafkaClient(configuration);
+ List<String> events = handler.fetchFromTopic("invalidTopic");
+ Assertions.assertEquals(0, events.size());
+ }
+
+ @Test
+ void whenPublishingToInvalidTopicExceptionIsLogged() {
+ OnapKafkaClient handler = new OnapKafkaClient(configuration);
+ RecordMetadata metadata = handler.publishToTopic("invalid.topic", "blahblahblahblah");
+ producerLogs.assertContains("Failed the send data");
+ Assertions.assertNull(metadata);
+ }
+
+ @Test
+ void whenSubscribingToInvalidTopicExceptionIsLogged() {
+ configuration = new TestConfiguration("invalid-application.properties");
+ OnapKafkaClient handler = new OnapKafkaClient(configuration);
+ handler.fetchFromTopic("bvserbatb");
+ clientLogs.assertContains("Consumer has not been initialised");
+ configuration.setConsumerTopics(consumerTopics);
+ }
+
+
+ private static void startKafkaService() throws Exception {
+ final BrokerListener listener = new SaslPlainListener()
+ .withUsername("kafkaclient")
+ .withPassword("client-secret");
+ final Properties brokerProperties = new Properties();
+ brokerProperties.setProperty("auto.create.topics.enable", "false");
+ kafkaTestCluster = new KafkaTestCluster(
+ 1,
+ brokerProperties,
+ Collections.singletonList(listener)
+ );
+ kafkaTestCluster.start();
+ logger.debug("Cluster started at: {}", kafkaTestCluster.getKafkaConnectString());
+ }
+
+ static {
+ System.setProperty("java.security.auth.login.config", "src/test/resources/jaas.conf");
+ }
+}
\ No newline at end of file
diff --git a/kafkaClient/src/test/java/org/onap/dmaap/kafka/TestConfiguration.java b/kafkaClient/src/test/java/org/onap/dmaap/kafka/TestConfiguration.java
new file mode 100644
index 0000000..b5fa9d1
--- /dev/null
+++ b/kafkaClient/src/test/java/org/onap/dmaap/kafka/TestConfiguration.java
@@ -0,0 +1,110 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * dmaap-kafka-client
+ * ================================================================================
+ * Copyright (C) 2023 Nordix Foundation. 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.dmaap.kafka;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import lombok.SneakyThrows;
+
+public class TestConfiguration implements org.onap.dmaap.kafka.IKafkaConfig {
+
+ private Properties loadProperties(String configFileName) throws IOException {
+ Properties configuration = new Properties();
+ try (InputStream inputStream = TestConfiguration.class
+ .getClassLoader()
+ .getResourceAsStream(configFileName)) {
+ configuration.load(inputStream);
+ }
+ return configuration;
+ }
+
+ private final Properties testConfig;
+ private List<String> bootstrapServers;
+ private List<String> consumerTopics;
+
+ @SneakyThrows
+ public TestConfiguration(String configFilename) {
+ testConfig = loadProperties(configFilename);
+ bootstrapServers = new ArrayList<>(Arrays.asList(((String) testConfig.get("kafka.kafkaBootstrapServers")).split(",")));
+ }
+
+ @Override
+ public List<String> getKafkaBootstrapServers() {
+ return bootstrapServers;
+ }
+
+ public void setBootstrapServers(List<String> newBootstrapList) {
+ bootstrapServers = newBootstrapList;
+ }
+
+ @Override
+ public String getKafkaSaslMechanism() {
+ return "PLAIN";
+ }
+
+ @Override
+ public String getKafkaSaslJaasConfig() {
+ return "org.apache.kafka.common.security.plain.PlainLoginModule required username=admin password=admin-secret;";
+ }
+
+ @Override
+ public int getPollingTimeout() {
+ return Integer.parseInt((String) testConfig.get("kafka.pollingTimeout"));
+ }
+
+ @Override
+ public String getConsumerGroup() {
+ return (String) testConfig.get("kafka.consumerGroup");
+ }
+
+ @Override
+ public String getConsumerID() {
+ return (String) testConfig.get("kafka.consumerID");
+ }
+
+ @Override
+ public List<String> getConsumerTopics() {
+ consumerTopics = new ArrayList<>();
+ String topicString = (String) testConfig.get("kafka.consumerTopics");
+ if (topicString != null) {
+ consumerTopics.addAll(Arrays.asList((topicString).split(",")));
+ }
+ return consumerTopics;
+ }
+
+ public void setConsumerTopics(List<String> newTopics) {
+ this.consumerTopics = newTopics;
+ }
+
+ @Override
+ public List<String> getProducerTopics() {
+ List<String> producerTopics = new ArrayList<>();
+ String topicString = (String) testConfig.get("kafka.producerTopics");
+ if (topicString != null) {
+ producerTopics.addAll(Arrays.asList((topicString).split(",")));
+ }
+ return producerTopics;
+ }
+}
--
cgit