src/test/java/org/onap/dmaap/mr/cambria/embed/EmbedConfigurationReader.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  Modification copyright (C) 2021 Nordix Foundation.
   7  *  ================================================================================
   8  *  Licensed under the Apache License, Version 2.0 (the "License");
   9  *  you may not use this file except in compliance with the License.
  10  *  You may obtain a copy of the License at
  11  *        http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *  Unless required by applicable law or agreed to in writing, software
  14  *  distributed under the License is distributed on an "AS IS" BASIS,
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *  See the License for the specific language governing permissions and
  17  *  limitations under the License.
  18  *  ============LICENSE_END=========================================================
  19  *
  20  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  21  *
  22  *******************************************************************************/
  23
  24  package org.onap.dmaap.mr.cambria.embed;
  25
  26 import java.io.File;
  27 import java.util.Arrays;
  28 import java.util.Map;
  29 import java.util.Properties;
  30
  31 import org.apache.commons.io.FileUtils;
  32 import org.apache.curator.framework.CuratorFramework;
  33
  34 import com.att.ajsc.filemonitor.AJSCPropertiesMap;
  35 import org.onap.dmaap.dmf.mr.backends.kafka.KafkaPublisher;
  36 import org.onap.dmaap.dmf.mr.backends.memory.MemoryMetaBroker;
  37 import org.onap.dmaap.dmf.mr.backends.memory.MemoryQueue;
  38 import org.apache.kafka.clients.admin.AdminClient;
  39 import org.apache.kafka.clients.admin.AdminClientConfig;
  40 import org.apache.kafka.clients.admin.NewTopic;
  41 import org.onap.dmaap.dmf.mr.beans.DMaaPKafkaConsumerFactory;
  42 import org.onap.dmaap.dmf.mr.beans.DMaaPKafkaMetaBroker;
  43 import org.onap.dmaap.dmf.mr.beans.DMaaPMetricsSet;
  44 import org.onap.dmaap.dmf.mr.beans.DMaaPZkClient;
  45 import org.onap.dmaap.dmf.mr.beans.DMaaPZkConfigDb;
  46 import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
  47 import org.onap.dmaap.dmf.mr.security.DMaaPAuthenticator;
  48 import org.onap.dmaap.dmf.mr.security.DMaaPAuthenticatorImpl;
  49 import org.onap.dmaap.dmf.mr.utils.ConfigurationReader;
  50 import org.onap.dmaap.dmf.mr.utils.DMaaPCuratorFactory;
  51 import org.onap.dmaap.dmf.mr.utils.PropertyReader;
  52 import com.att.nsa.security.db.BaseNsaApiDbImpl;
  53 import com.att.nsa.security.db.simple.NsaSimpleApiKey;
  54 import com.att.nsa.security.db.simple.NsaSimpleApiKeyFactory;
  55
  56
  57 public class EmbedConfigurationReader {
  58         private static final String DEFAULT_KAFKA_LOG_DIR = "/kafka_embedded";
  59     public static final String TEST_TOPIC = "testTopic";
  60     private static final int BROKER_ID = 0;
  61     private static final int BROKER_PORT = 5000;
  62     private static final String LOCALHOST_BROKER = String.format("localhost:%d", BROKER_PORT);
  63
  64     private static final String DEFAULT_ZOOKEEPER_LOG_DIR = "/zookeeper";
  65     private static final int ZOOKEEPER_PORT = 2000;
  66     private static final String ZOOKEEPER_HOST = String.format("localhost:%d", ZOOKEEPER_PORT);
  67
  68     private static final String groupId = "groupID";
  69     String dir;
  70     private  AdminClient fKafkaAdminClient;
  71     KafkaLocal kafkaLocal;
  72
  73         public void setUp() throws Exception {
  74
  75                 ClassLoader classLoader = getClass().getClassLoader();
  76                 AJSCPropertiesMap.refresh(new File(classLoader.getResource(CambriaConstants.msgRtr_prop).getFile()));
  77
  78                 Properties kafkaProperties;
  79         Properties zkProperties;
  80
  81         try {
  82             //load properties
  83                 dir = new File(classLoader.getResource(CambriaConstants.msgRtr_prop).getFile()).getParent();
  84             kafkaProperties = getKafkaProperties(dir + DEFAULT_KAFKA_LOG_DIR, BROKER_PORT, BROKER_ID);
  85             zkProperties = getZookeeperProperties(ZOOKEEPER_PORT,dir + DEFAULT_ZOOKEEPER_LOG_DIR);
  86
  87             //start kafkaLocalServer
  88             kafkaLocal = new KafkaLocal(kafkaProperties, zkProperties);
  89
  90             Map<String, String> map = AJSCPropertiesMap.getProperties(CambriaConstants.msgRtr_prop);
  91             map.put(CambriaConstants.kSetting_ZkConfigDbServers, ZOOKEEPER_HOST);
  92             map.put("kafka.client.zookeeper", ZOOKEEPER_HOST);
  93             map.put("kafka.metadata.broker.list", LOCALHOST_BROKER);
  94
  95             DMaaPZkClient dMaaPZkClient = new DMaaPZkClient(new PropertyReader());
  96
  97             final Properties props = new Properties ();
  98              props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092" );
  99              props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin_secret';");
 100                  props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
 101              props.put("sasl.mechanism", "PLAIN");
 102              fKafkaAdminClient = AdminClient.create ( props );
 103
 104            // if(!AdminUtils.topicExists(dMaaPZkClient, TEST_TOPIC))
 105             //  AdminUtils.createTopic(dMaaPZkClient, TEST_TOPIC, 3, 1, new Properties());
 106              final NewTopic topicRequest = new NewTopic ( TEST_TOPIC, 3, new Integer(1).shortValue () );
 107                          fKafkaAdminClient.createTopics ( Arrays.asList ( topicRequest ) );
 108             Thread.sleep(5000);
 109         } catch (Exception e){
 110             e.printStackTrace(System.out);
 111         }
 112         }
 113
 114         private static Properties getKafkaProperties(String logDir, int port, int brokerId) {
 115         Properties properties = new Properties();
 116         properties.put("port", port + "");
 117         properties.put("broker.id", brokerId + "");
 118         properties.put("log.dir", logDir);
 119         properties.put("zookeeper.connect", ZOOKEEPER_HOST);
 120         properties.put("default.replication.factor", "1");
 121         properties.put("delete.topic.enable", "true");
 122         properties.put("consumer.timeout.ms", -1);
 123         return properties;
 124     }
 125
 126         private static Properties getZookeeperProperties(int port, String zookeeperDir) {
 127         Properties properties = new Properties();
 128         properties.put("clientPort", port + "");
 129         properties.put("dataDir", zookeeperDir);
 130         return properties;
 131     }
 132
 133         public void tearDown() throws Exception {
 134                 DMaaPZkClient dMaaPZkClient = new DMaaPZkClient(new PropertyReader());
 135                 if(fKafkaAdminClient!=null)
 136                 fKafkaAdminClient.deleteTopics(Arrays.asList(TEST_TOPIC));
 137                 //AdminUtils.deleteTopic(dMaaPZkClient, TEST_TOPIC);
 138                 //dMaaPZkClient.delete(dir + DEFAULT_KAFKA_LOG_DIR);
 139                 //dMaaPZkClient.delete(dir + DEFAULT_ZOOKEEPER_LOG_DIR);
 140                 kafkaLocal.stop();
 141                 FileUtils.cleanDirectory(new File(dir + DEFAULT_KAFKA_LOG_DIR));
 142         }
 143
 144
 145         public ConfigurationReader buildConfigurationReader() throws Exception {
 146
 147                 setUp();
 148
 149                 PropertyReader propertyReader = new PropertyReader();
 150                 DMaaPMetricsSet dMaaPMetricsSet = new DMaaPMetricsSet(propertyReader);
 151                 DMaaPZkClient dMaaPZkClient = new DMaaPZkClient(propertyReader);
 152                 DMaaPZkConfigDb dMaaPZkConfigDb = new DMaaPZkConfigDb(dMaaPZkClient, propertyReader);
 153                 CuratorFramework curatorFramework = DMaaPCuratorFactory.getCurator(new PropertyReader());
 154                 DMaaPKafkaConsumerFactory dMaaPKafkaConsumerFactory = new DMaaPKafkaConsumerFactory(dMaaPMetricsSet, curatorFramework,null);
 155                 MemoryQueue memoryQueue = new MemoryQueue();
 156                 MemoryMetaBroker memoryMetaBroker = new MemoryMetaBroker(memoryQueue, dMaaPZkConfigDb);
 157                 BaseNsaApiDbImpl<NsaSimpleApiKey> baseNsaApiDbImpl = new BaseNsaApiDbImpl<>(dMaaPZkConfigDb, new NsaSimpleApiKeyFactory());
 158                 DMaaPAuthenticator<NsaSimpleApiKey> dMaaPAuthenticator = new DMaaPAuthenticatorImpl<>(baseNsaApiDbImpl);
 159                 KafkaPublisher kafkaPublisher = new KafkaPublisher(propertyReader);
 160                 DMaaPKafkaMetaBroker dMaaPKafkaMetaBroker = new DMaaPKafkaMetaBroker(propertyReader, dMaaPZkClient, dMaaPZkConfigDb);
 161
 162                 return new ConfigurationReader(propertyReader,
 163                                 dMaaPMetricsSet, dMaaPZkClient, dMaaPZkConfigDb, kafkaPublisher,
 164                                 curatorFramework, dMaaPKafkaConsumerFactory, dMaaPKafkaMetaBroker,
 165                                 memoryQueue, memoryMetaBroker, baseNsaApiDbImpl, dMaaPAuthenticator);
 166
 167         }
 168 }