2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2016-2018 Ericsson. All rights reserved.
4 * Modifications Copyright (C) 2020 Nordix Foundation.
5 * ================================================================================
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
18 * SPDX-License-Identifier: Apache-2.0
19 * ============LICENSE_END=========================================================
22 package org.onap.policy.apex.testsuites.integration.uservice.adapt.kafka;
24 import static org.awaitility.Awaitility.await;
25 import static org.junit.Assert.assertEquals;
26 import static org.junit.Assert.fail;
28 import com.salesforce.kafka.test.junit4.SharedKafkaTestResource;
30 import java.io.IOException;
31 import java.util.concurrent.TimeUnit;
32 import org.junit.Before;
33 import org.junit.ClassRule;
34 import org.junit.Test;
35 import org.onap.policy.apex.service.engine.main.ApexMain;
36 import org.onap.policy.common.utils.resources.TextFileUtils;
39 * The Class TestKafka2Kafka tests Kafka event sending and reception.
41 public class TestKafka2Kafka {
42 private static final long MAX_TEST_LENGTH = 300000;
44 private static final int EVENT_COUNT = 25;
45 private static final int EVENT_INTERVAL = 20;
48 * Clear relative file root environment variable.
51 public void clearRelativeFileRoot() {
52 System.clearProperty("APEX_RELATIVE_FILE_ROOT");
56 public static final SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource()
57 // Start a cluster with 1 brokers.
59 // Disable topic auto-creation.
60 .withBrokerProperty("auto.create.topics.enable", "false");
63 * Test json kafka events.
65 * @throws Exception the apex exception
68 public void testJsonKafkaEvents() throws Exception {
69 final String conditionedConfigFile = getConditionedConfigFile(
70 "target" + File.separator + "examples/config/SampleDomain/Kafka2KafkaJsonEvent.json");
71 final String[] args = {"-rfr", "target", "-c", conditionedConfigFile};
72 testKafkaEvents(args, false, "json");
76 * Test XML kafka events.
78 * @throws Exception the apex exception
81 public void testXmlKafkaEvents() throws Exception {
82 final String conditionedConfigFile = getConditionedConfigFile(
83 "target" + File.separator + "examples/config/SampleDomain/Kafka2KafkaXmlEvent.json");
84 final String[] args = {"-rfr", "target", "-c", conditionedConfigFile};
86 testKafkaEvents(args, true, "xml");
92 * @param args the args
93 * @param xmlEvents the xml events
94 * @param topicSuffix the topic suffix
95 * @throws Exception on errors
97 private void testKafkaEvents(String[] args, final Boolean xmlEvents, final String topicSuffix) throws Exception {
99 sharedKafkaTestResource.getKafkaTestUtils().createTopic("apex-out-" + topicSuffix, 1, (short) 1);
100 sharedKafkaTestResource.getKafkaTestUtils().createTopic("apex-in-" + topicSuffix, 1, (short) 1);
102 final KafkaEventSubscriber subscriber =
103 new KafkaEventSubscriber("apex-out-" + topicSuffix, sharedKafkaTestResource);
105 await().atMost(30, TimeUnit.SECONDS).until(() -> subscriber.isAlive());
107 final ApexMain apexMain = new ApexMain(args);
108 await().atMost(10, TimeUnit.SECONDS).until(() -> apexMain.isAlive());
110 long initWaitEndTIme = System.currentTimeMillis() + 10000;
112 await().atMost(12, TimeUnit.SECONDS).until(() -> initWaitEndTIme < System.currentTimeMillis());
114 final KafkaEventProducer producer = new KafkaEventProducer("apex-in-" + topicSuffix, sharedKafkaTestResource,
115 EVENT_COUNT, xmlEvents, EVENT_INTERVAL);
117 await().atMost(30, TimeUnit.SECONDS).until(() -> producer.isAlive());
119 producer.sendEvents();
121 // Wait for the producer to send all its events
122 await().atMost(MAX_TEST_LENGTH, TimeUnit.MILLISECONDS)
123 .until(() -> producer.getEventsSentCount() >= EVENT_COUNT);
125 await().atMost(MAX_TEST_LENGTH, TimeUnit.MILLISECONDS)
126 .until(() -> subscriber.getEventsReceivedCount() >= EVENT_COUNT);
129 await().atMost(30, TimeUnit.SECONDS).until(() -> !apexMain.isAlive());
131 subscriber.shutdown();
132 await().atMost(30, TimeUnit.SECONDS).until(() -> !subscriber.isAlive());
135 await().atMost(30, TimeUnit.SECONDS).until(() -> !producer.isAlive());
137 assertEquals(producer.getEventsSentCount(), subscriber.getEventsReceivedCount());
140 private String getConditionedConfigFile(final String configurationFileName) {
142 File tempConfigFile = File.createTempFile("Kafka_", ".json");
143 tempConfigFile.deleteOnExit();
144 String configAsString = TextFileUtils.getTextFileAsString(configurationFileName)
145 .replaceAll("localhost:39902", sharedKafkaTestResource.getKafkaConnectString());
146 TextFileUtils.putStringAsFile(configAsString, tempConfigFile.getCanonicalFile());
148 return tempConfigFile.getCanonicalPath();
149 } catch (IOException e) {
150 fail("test should not throw an exception");