diff options
author | 2018-08-14 09:34:46 -0400 | |
---|---|---|
committer | 2018-08-14 09:39:23 -0400 | |
commit | b32effcaf5684d5e2f338a4537b71a2375c534e5 (patch) | |
tree | e1b80407f414509ffcc766b987ec6a95f7254b4e /src/test/java/com/att/nsa/cambria/embed/EmbedConfigurationReader.java | |
parent | 0823cb186012c8e6b7de3d979dfabb9f838da7c2 (diff) |
update the testcases after the kafka 11 changes
Issue-ID: DMAAP-526
Change-Id: I477a8ee05fb3cdd76af726b6ca0d1a69aa9eef93
Signed-off-by: sunil unnava <su622b@att.com>
Diffstat (limited to 'src/test/java/com/att/nsa/cambria/embed/EmbedConfigurationReader.java')
-rw-r--r-- | src/test/java/com/att/nsa/cambria/embed/EmbedConfigurationReader.java | 57 |
1 files changed, 37 insertions, 20 deletions
diff --git a/src/test/java/com/att/nsa/cambria/embed/EmbedConfigurationReader.java b/src/test/java/com/att/nsa/cambria/embed/EmbedConfigurationReader.java index 7233c6c..30cb460 100644 --- a/src/test/java/com/att/nsa/cambria/embed/EmbedConfigurationReader.java +++ b/src/test/java/com/att/nsa/cambria/embed/EmbedConfigurationReader.java @@ -23,6 +23,7 @@ package com.att.nsa.cambria.embed;
import java.io.File;
+import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
@@ -30,25 +31,29 @@ import org.apache.commons.io.FileUtils; import org.apache.curator.framework.CuratorFramework;
import com.att.ajsc.filemonitor.AJSCPropertiesMap;
-import com.att.nsa.cambria.backends.kafka.KafkaPublisher;
-import com.att.nsa.cambria.backends.memory.MemoryMetaBroker;
-import com.att.nsa.cambria.backends.memory.MemoryQueue;
-import com.att.nsa.cambria.beans.DMaaPKafkaConsumerFactory;
-import com.att.nsa.cambria.beans.DMaaPKafkaMetaBroker;
-import com.att.nsa.cambria.beans.DMaaPMetricsSet;
-import com.att.nsa.cambria.beans.DMaaPZkClient;
-import com.att.nsa.cambria.beans.DMaaPZkConfigDb;
-import com.att.nsa.cambria.constants.CambriaConstants;
-import com.att.nsa.cambria.security.DMaaPAuthenticator;
-import com.att.nsa.cambria.security.DMaaPAuthenticatorImpl;
-import com.att.nsa.cambria.utils.ConfigurationReader;
-import com.att.nsa.cambria.utils.DMaaPCuratorFactory;
-import com.att.nsa.cambria.utils.PropertyReader;
+import com.att.dmf.mr.backends.kafka.KafkaPublisher;
+import com.att.dmf.mr.backends.memory.MemoryMetaBroker;
+import com.att.dmf.mr.backends.memory.MemoryQueue;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.KafkaFuture;
+import com.att.dmf.mr.beans.DMaaPKafkaConsumerFactory;
+import com.att.dmf.mr.beans.DMaaPKafkaMetaBroker;
+import com.att.dmf.mr.beans.DMaaPMetricsSet;
+import com.att.dmf.mr.beans.DMaaPZkClient;
+import com.att.dmf.mr.beans.DMaaPZkConfigDb;
+import com.att.dmf.mr.constants.CambriaConstants;
+import com.att.dmf.mr.security.DMaaPAuthenticator;
+import com.att.dmf.mr.security.DMaaPAuthenticatorImpl;
+import com.att.dmf.mr.utils.ConfigurationReader;
+import com.att.dmf.mr.utils.DMaaPCuratorFactory;
+import com.att.dmf.mr.utils.PropertyReader;
import com.att.nsa.security.db.BaseNsaApiDbImpl;
import com.att.nsa.security.db.simple.NsaSimpleApiKey;
import com.att.nsa.security.db.simple.NsaSimpleApiKeyFactory;
-import kafka.admin.AdminUtils;
public class EmbedConfigurationReader {
private static final String DEFAULT_KAFKA_LOG_DIR = "/kafka_embedded";
@@ -63,7 +68,7 @@ public class EmbedConfigurationReader { private static final String groupId = "groupID";
String dir;
-
+ private AdminClient fKafkaAdminClient;
KafkaLocal kafkaLocal;
public void setUp() throws Exception {
@@ -89,8 +94,18 @@ public class EmbedConfigurationReader { map.put("kafka.metadata.broker.list", LOCALHOST_BROKER);
DMaaPZkClient dMaaPZkClient = new DMaaPZkClient(new PropertyReader());
- if(!AdminUtils.topicExists(dMaaPZkClient, TEST_TOPIC))
- AdminUtils.createTopic(dMaaPZkClient, TEST_TOPIC, 3, 1, new Properties());
+
+ final Properties props = new Properties ();
+ props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092" );
+ props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin_secret'");
+ props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
+ props.put("sasl.mechanism", "PLAIN");
+ fKafkaAdminClient = AdminClient.create ( props );
+
+ // if(!AdminUtils.topicExists(dMaaPZkClient, TEST_TOPIC))
+ // AdminUtils.createTopic(dMaaPZkClient, TEST_TOPIC, 3, 1, new Properties());
+ final NewTopic topicRequest = new NewTopic ( TEST_TOPIC, 3, new Integer(1).shortValue () );
+ fKafkaAdminClient.createTopics ( Arrays.asList ( topicRequest ) );
Thread.sleep(5000);
} catch (Exception e){
e.printStackTrace(System.out);
@@ -118,7 +133,9 @@ public class EmbedConfigurationReader { public void tearDown() throws Exception {
DMaaPZkClient dMaaPZkClient = new DMaaPZkClient(new PropertyReader());
- AdminUtils.deleteTopic(dMaaPZkClient, TEST_TOPIC);
+ if(fKafkaAdminClient!=null)
+ fKafkaAdminClient.deleteTopics(Arrays.asList(TEST_TOPIC));
+ //AdminUtils.deleteTopic(dMaaPZkClient, TEST_TOPIC);
//dMaaPZkClient.delete(dir + DEFAULT_KAFKA_LOG_DIR);
//dMaaPZkClient.delete(dir + DEFAULT_ZOOKEEPER_LOG_DIR);
kafkaLocal.stop();
@@ -135,7 +152,7 @@ public class EmbedConfigurationReader { DMaaPZkClient dMaaPZkClient = new DMaaPZkClient(propertyReader);
DMaaPZkConfigDb dMaaPZkConfigDb = new DMaaPZkConfigDb(dMaaPZkClient, propertyReader);
CuratorFramework curatorFramework = DMaaPCuratorFactory.getCurator(new PropertyReader());
- DMaaPKafkaConsumerFactory dMaaPKafkaConsumerFactory = new DMaaPKafkaConsumerFactory(propertyReader, dMaaPMetricsSet, curatorFramework);
+ DMaaPKafkaConsumerFactory dMaaPKafkaConsumerFactory = new DMaaPKafkaConsumerFactory(dMaaPMetricsSet, curatorFramework,null);
MemoryQueue memoryQueue = new MemoryQueue();
MemoryMetaBroker memoryMetaBroker = new MemoryMetaBroker(memoryQueue, dMaaPZkConfigDb);
BaseNsaApiDbImpl<NsaSimpleApiKey> baseNsaApiDbImpl = new BaseNsaApiDbImpl<>(dMaaPZkConfigDb, new NsaSimpleApiKeyFactory());
|