2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2016-2018 Ericsson. All rights reserved.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.apex.testsuites.integration.uservice.adapt.kafka;
23 import static org.junit.Assert.assertEquals;
24 import static org.junit.Assert.fail;
26 import com.salesforce.kafka.test.junit4.SharedKafkaTestResource;
29 import java.io.IOException;
31 import org.junit.ClassRule;
32 import org.junit.Test;
33 import org.onap.policy.apex.core.infrastructure.messaging.MessagingException;
34 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
35 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
36 import org.onap.policy.apex.model.utilities.TextFileUtils;
37 import org.onap.policy.apex.service.engine.main.ApexMain;
40 * The Class TestKafka2Kafka tests Kafka event sending and reception.
42 public class TestKafka2Kafka {
43 private static final long MAX_TEST_LENGTH = 60000;
45 private static final int EVENT_COUNT = 100;
46 private static final int EVENT_INTERVAL = 20;
49 public static final SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource()
50 // Start a cluster with 1 brokers.
52 // Disable topic auto-creation.
53 .withBrokerProperty("auto.create.topics.enable", "false");
56 * Test json kafka events.
58 * @throws MessagingException the messaging exception
59 * @throws ApexException the apex exception
62 public void testJsonKafkaEvents() throws MessagingException, ApexException {
64 { "src/test/resources/prodcons/Kafka2KafkaJsonEvent.json" };
65 testKafkaEvents(args, false, "json");
69 * Test XML kafka events.
71 * @throws MessagingException the messaging exception
72 * @throws ApexException the apex exception
75 public void testXmlKafkaEvents() throws MessagingException, ApexException {
77 { "src/test/resources/prodcons/Kafka2KafkaXmlEvent.json" };
78 testKafkaEvents(args, true, "xml");
84 * @param args the args
85 * @param xmlEvents the xml events
86 * @param topicSuffix the topic suffix
87 * @throws MessagingException the messaging exception
88 * @throws ApexException the apex exception
90 private void testKafkaEvents(String[] args, final Boolean xmlEvents, final String topicSuffix)
91 throws MessagingException, ApexException {
94 File tempConfigFile = File.createTempFile("Kafka_", ".json");
95 tempConfigFile.deleteOnExit();
96 String configAsString = TextFileUtils.getTextFileAsString(args[0]).replaceAll("localhost:39902",
97 sharedKafkaTestResource.getKafkaConnectString());
98 TextFileUtils.putStringAsFile(configAsString, tempConfigFile.getCanonicalFile());
99 args[0] = tempConfigFile.getCanonicalPath();
101 } catch (IOException e) {
102 fail("test should not throw an exception");
105 sharedKafkaTestResource.getKafkaTestUtils().createTopic("apex-out-" + topicSuffix, 1, (short) 1);
106 sharedKafkaTestResource.getKafkaTestUtils().createTopic("apex-in-" + topicSuffix, 1, (short) 1);
108 final KafkaEventSubscriber subscriber = new KafkaEventSubscriber("apex-out-" + topicSuffix,
109 sharedKafkaTestResource);
111 final ApexMain apexMain = new ApexMain(args);
112 ThreadUtilities.sleep(3000);
114 final KafkaEventProducer producer = new KafkaEventProducer("apex-in-" + topicSuffix, sharedKafkaTestResource,
115 EVENT_COUNT, xmlEvents, EVENT_INTERVAL);
117 producer.sendEvents();
119 final long testStartTime = System.currentTimeMillis();
121 // Wait for the producer to send all tis events
122 while (System.currentTimeMillis() < testStartTime + MAX_TEST_LENGTH
123 && producer.getEventsSentCount() < EVENT_COUNT) {
124 ThreadUtilities.sleep(EVENT_INTERVAL);
127 while (System.currentTimeMillis() < testStartTime + MAX_TEST_LENGTH
128 && subscriber.getEventsReceivedCount() < EVENT_COUNT) {
129 ThreadUtilities.sleep(EVENT_INTERVAL);
132 ThreadUtilities.sleep(1000);
134 assertEquals(producer.getEventsSentCount(), subscriber.getEventsReceivedCount());
137 subscriber.shutdown();
139 ThreadUtilities.sleep(1000);