gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   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
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.apex.apps.uservice.test.adapt.kafka;
  22
  23 import static org.junit.Assert.assertEquals;
  24
  25 import java.io.IOException;
  26 import java.nio.file.Files;
  27 import java.util.Properties;
  28
  29 import kafka.admin.AdminUtils;
  30 import kafka.admin.RackAwareMode;
  31 import kafka.server.KafkaConfig;
  32 import kafka.server.KafkaServer;
  33 import kafka.utils.TestUtils;
  34 import kafka.utils.ZKStringSerializer$;
  35 import kafka.utils.ZkUtils;
  36 import kafka.zk.EmbeddedZookeeper;
  37
  38 import org.I0Itec.zkclient.ZkClient;
  39 import org.apache.kafka.common.utils.MockTime;
  40 import org.apache.kafka.common.utils.Time;
  41 import org.junit.AfterClass;
  42 import org.junit.BeforeClass;
  43 import org.junit.Test;
  44 import org.onap.policy.apex.core.infrastructure.messaging.MessagingException;
  45 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  46 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
  47 import org.onap.policy.apex.service.engine.main.ApexMain;
  48
  49
  50 /**
  51  * The Class TestKafka2Kafka.
  52  */
  53 public class TestKafka2Kafka {
  54     // The method of starting an embedded Kafka server used in this example is based on the method
  55     // on stack overflow at
  56     // https://github.com/asmaier/mini-kafka
  57
  58     private static final long MAX_TEST_LENGTH = 20000;
  59
  60     private static final int EVENT_COUNT = 10;
  61     private static final int EVENT_INTERVAL = 20;
  62
  63     private static final String ZKHOST = "127.0.0.1";
  64     private static final String BROKERHOST = "127.0.0.1";
  65     private static final String BROKERPORT = "39902";
  66
  67     private static EmbeddedZookeeper zkServer;
  68     private static ZkClient zkClient;
  69     private static KafkaServer kafkaServer;
  70
  71     /**
  72      * Setup dummy kafka server.
  73      *
  74      * @throws IOException Signals that an I/O exception has occurred.
  75      */
  76     @BeforeClass
  77     public static void setupDummyKafkaServer() throws IOException {
  78         // setup Zookeeper
  79         zkServer = new EmbeddedZookeeper();
  80         final String zkConnect = ZKHOST + ":" + zkServer.port();
  81         zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer$.MODULE$);
  82         final ZkUtils zkUtils = ZkUtils.apply(zkClient, false);
  83
  84         // setup Broker
  85         final Properties brokerProps = new Properties();
  86         brokerProps.setProperty("zookeeper.connect", zkConnect);
  87         brokerProps.setProperty("broker.id", "0");
  88         brokerProps.setProperty("log.dirs", Files.createTempDirectory("kafka-").toAbsolutePath().toString());
  89         brokerProps.setProperty("listeners", "PLAINTEXT://" + BROKERHOST + ":" + BROKERPORT);
  90         brokerProps.setProperty("offsets.topic.replication.factor", "1");
  91         brokerProps.setProperty("transaction.state.log.replication.factor", "1");
  92         brokerProps.setProperty("transaction.state.log.min.isr", "1");
  93         final KafkaConfig config = new KafkaConfig(brokerProps);
  94         final Time mock = new MockTime();
  95         kafkaServer = TestUtils.createServer(config, mock);
  96         kafkaServer.startup();
  97
  98         // create topics
  99         AdminUtils.createTopic(zkUtils, "apex-in-0", 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$);
 100         AdminUtils.createTopic(zkUtils, "apex-in-1", 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$);
 101         AdminUtils.createTopic(zkUtils, "apex-out", 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$);
 102
 103     }
 104
 105     /**
 106      * Shutdown dummy kafka server.
 107      *
 108      * @throws IOException Signals that an I/O exception has occurred.
 109      */
 110     @AfterClass
 111     public static void shutdownDummyKafkaServer() throws IOException {
 112         if (kafkaServer != null) {
 113             kafkaServer.shutdown();
 114         }
 115         if (zkClient != null) {
 116             zkClient.close();
 117         }
 118         if (zkServer != null) {
 119             zkServer.shutdown();
 120         }
 121     }
 122
 123     /**
 124      * Test json kafka events.
 125      *
 126      * @throws MessagingException the messaging exception
 127      * @throws ApexException the apex exception
 128      */
 129     @Test
 130     public void testJsonKafkaEvents() throws MessagingException, ApexException {
 131         final String[] args = {"src/test/resources/prodcons/Kafka2KafkaJsonEvent.json"};
 132         testKafkaEvents(args, false, "json");
 133     }
 134
 135     /**
 136      * Test XML kafka events.
 137      *
 138      * @throws MessagingException the messaging exception
 139      * @throws ApexException the apex exception
 140      */
 141     @Test
 142     public void testXmlKafkaEvents() throws MessagingException, ApexException {
 143         final String[] args = {"src/test/resources/prodcons/Kafka2KafkaXmlEvent.json"};
 144         testKafkaEvents(args, true, "xml");
 145     }
 146
 147     /**
 148      * Test kafka events.
 149      *
 150      * @param args the args
 151      * @param xmlEvents the xml events
 152      * @param topicSuffix the topic suffix
 153      * @throws MessagingException the messaging exception
 154      * @throws ApexException the apex exception
 155      */
 156     private void testKafkaEvents(final String[] args, final Boolean xmlEvents, final String topicSuffix)
 157             throws MessagingException, ApexException {
 158         final KafkaEventSubscriber subscriber =
 159                 new KafkaEventSubscriber("apex-out-" + topicSuffix, "localhost:" + BROKERPORT);
 160
 161         final ApexMain apexMain = new ApexMain(args);
 162         ThreadUtilities.sleep(3000);
 163
 164         final KafkaEventProducer producer = new KafkaEventProducer("apex-in-" + topicSuffix, "localhost:" + BROKERPORT,
 165                         EVENT_COUNT, xmlEvents, EVENT_INTERVAL);
 166
 167         producer.sendEvents();
 168
 169         final long testStartTime = System.currentTimeMillis();
 170
 171         while (System.currentTimeMillis() < testStartTime + MAX_TEST_LENGTH
 172                 && subscriber.getEventsReceivedCount() < EVENT_COUNT) {
 173             ThreadUtilities.sleep(EVENT_INTERVAL);
 174         }
 175
 176         ThreadUtilities.sleep(1000);
 177
 178         assertEquals(producer.getEventsSentCount(), subscriber.getEventsReceivedCount());
 179
 180         apexMain.shutdown();
 181         subscriber.shutdown();
 182         producer.shutdown();
 183         ThreadUtilities.sleep(1000);
 184     }
 185 }