1 /*******************************************************************************
2 * ============LICENSE_START=======================================================
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
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=========================================================
20 * ECOMP is a trademark and service mark of AT&T Intellectual Property.
22 *******************************************************************************/
24 package org.onap.dmaap.mr.cambria.embed;
27 import java.util.Arrays;
29 import java.util.Properties;
31 import org.apache.commons.io.FileUtils;
32 import org.apache.curator.framework.CuratorFramework;
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;
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);
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);
68 private static final String groupId = "groupID";
70 private AdminClient fKafkaAdminClient;
71 KafkaLocal kafkaLocal;
73 public void setUp() throws Exception {
75 ClassLoader classLoader = getClass().getClassLoader();
76 AJSCPropertiesMap.refresh(new File(classLoader.getResource(CambriaConstants.msgRtr_prop).getFile()));
78 Properties kafkaProperties;
79 Properties zkProperties;
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);
87 //start kafkaLocalServer
88 kafkaLocal = new KafkaLocal(kafkaProperties, zkProperties);
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);
95 DMaaPZkClient dMaaPZkClient = new DMaaPZkClient(new PropertyReader());
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 );
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 ) );
109 } catch (Exception e){
110 e.printStackTrace(System.out);
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);
126 private static Properties getZookeeperProperties(int port, String zookeeperDir) {
127 Properties properties = new Properties();
128 properties.put("clientPort", port + "");
129 properties.put("dataDir", zookeeperDir);
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);
141 FileUtils.cleanDirectory(new File(dir + DEFAULT_KAFKA_LOG_DIR));
145 public ConfigurationReader buildConfigurationReader() throws Exception {
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);
162 return new ConfigurationReader(propertyReader,
163 dMaaPMetricsSet, dMaaPZkClient, dMaaPZkConfigDb, kafkaPublisher,
164 curatorFramework, dMaaPKafkaConsumerFactory, dMaaPKafkaMetaBroker,
165 memoryQueue, memoryMetaBroker, baseNsaApiDbImpl, dMaaPAuthenticator);