Unit test

Issue-ID: DCAEGEN2-1468
Change-Id: Ib0a1bbfa02390a9093e0a8ac4da1ae3fe2c1cd11
Signed-off-by: Guobiao Mo <guobiaomo@chinamobile.com>

30 files changed, 898 insertions, 648 deletions

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java
index 2a653d89..cfd2462b 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java
@@ -142,11 +142,11 @@ public class Db {
 		if (this.getClass() != obj.getClass())
 			return false;
 
-		return name.equals(((Db) obj).getName());
+		return id==((Db) obj).getId();
 	}
 
 	@Override
 	public int hashCode() {
-		return name.hashCode();
+		return id;
 	}
 }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
index c680e71b..13e0163e 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
@@ -38,12 +38,6 @@ import org.apache.commons.lang3.StringUtils;
 import org.json.JSONObject;
 import org.onap.datalake.feeder.dto.TopicConfig;
 import org.onap.datalake.feeder.enumeration.DataFormat;
-import org.onap.datalake.feeder.enumeration.DbTypeEnum;
-import org.onap.datalake.feeder.service.db.CouchbaseService;
-import org.onap.datalake.feeder.service.db.DbStoreService;
-import org.onap.datalake.feeder.service.db.ElasticsearchService;
-import org.onap.datalake.feeder.service.db.HdfsService;
-import org.onap.datalake.feeder.service.db.MongodbService;
 
 import com.fasterxml.jackson.annotation.JsonBackReference;
 
@@ -129,13 +123,6 @@ public class Topic {
 	@Column(name = "`flatten_array_path`")
 	protected String flattenArrayPath;
 	
-	public Topic() {
-	}
-/*
-	public Topic(String name) {//TODO
-		//this.name = name;
-	}
-*/
 	public String getName() {
 		return topicName.getId();
 	}
@@ -147,33 +134,7 @@ public class Topic {
 			return 3650;//default to 10 years for safe
 		}
 	}
-/*
-	public boolean supportHdfs() {
-		return supportDb(DbTypeEnum.HDFS);
-	}
-
-	public boolean supportElasticsearch() {
-		return supportDb(DbTypeEnum.ES);
-	}
-
-	public boolean supportCouchbase() {
-		return supportDb(DbTypeEnum.CB);
-	}
 
-	public boolean supportDruid() {
-		return supportDb(DbTypeEnum.DRUID);
-	}
-
-	public boolean supportMongoDB() {
-		return supportDb(DbTypeEnum.MONGO);
-	}
-
-	private boolean supportDb(DbTypeEnum dbTypeEnum) {
-		for(Db db : dbs) {
-			
-		}
-	}
-*/
 	public DataFormat getDataFormat2() {
 		if (dataFormat != null) {
 			return DataFormat.fromString(dataFormat);
@@ -204,7 +165,7 @@ public class Topic {
 
 	//extract DB id from JSON attributes, support multiple attributes
 	public String getMessageId(JSONObject json) {
-		String id = null;
+		String ret = null;
 
 		if (StringUtils.isNotBlank(messageIdPath)) {
 			String[] paths = messageIdPath.split(",");
@@ -216,10 +177,10 @@ public class Topic {
 				}
 				sb.append(json.query(paths[i]).toString());
 			}
-			id = sb.toString();
+			ret = sb.toString();
 		}
 
-		return id;
+		return ret;
 	}
 
 	public TopicConfig getTopicConfig() {
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
index 942526d2..a51103b7 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
@@ -70,12 +70,12 @@ public class TopicConfig {
 		if (this.getClass() != obj.getClass())
 			return false;
 
-		return name.equals(((TopicConfig) obj).getName());
+		return id==((TopicConfig) obj).getId();
 	}
 
 	@Override
 	public int hashCode() {
-		return name.hashCode();
+		return id;
 	}
 
 }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/CouchbaseService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/CouchbaseService.java
index bd2d9715..f2ac5e94 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/CouchbaseService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/CouchbaseService.java
@@ -73,7 +73,8 @@ public class CouchbaseService implements DbStoreService {
 	}
 	
 	@PostConstruct
-	private void init() {
+	@Override
+	public void init() {
 		// Initialize Couchbase Connection
 		try {
 			//this tunes the SDK (to customize connection timeout)
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/DbStoreService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/DbStoreService.java
index 5ea6e23e..c873c010 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/DbStoreService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/DbStoreService.java
@@ -34,4 +34,6 @@ import org.onap.datalake.feeder.domain.EffectiveTopic;
 public interface DbStoreService {
 
 	void saveJsons(EffectiveTopic topic, List<JSONObject> jsons);
+
+	void init();
 }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/ElasticsearchService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/ElasticsearchService.java
index 4dfcdd22..18b7e2fb 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/ElasticsearchService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/ElasticsearchService.java
@@ -83,7 +83,8 @@ public class ElasticsearchService implements DbStoreService {
 	//ES Encrypted communication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_encrypted_communication.html#_encrypted_communication
 	//Basic authentication https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_basic_authentication.html
 	@PostConstruct
-	private void init() {
+	@Override
+	public void init() {
 		String elasticsearchHost = elasticsearch.getHost();
 
 		// Initialize the Connection
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/HdfsService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/HdfsService.java
index ea0e77aa..1725ee41 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/HdfsService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/HdfsService.java
@@ -32,7 +32,6 @@ import java.util.Map;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
@@ -45,13 +44,11 @@ import org.onap.datalake.feeder.domain.EffectiveTopic;
 import org.onap.datalake.feeder.util.Util;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Scope;
 import org.springframework.stereotype.Service;
 
 import lombok.Getter;
-import lombok.Setter;
 
 /**
  * Service to write data to HDFS
@@ -64,20 +61,18 @@ import lombok.Setter;
 public class HdfsService implements DbStoreService {
 
 	private final Logger log = LoggerFactory.getLogger(this.getClass());
-	
+
 	private Db hdfs;
 
 	@Autowired
 	ApplicationConfiguration config;
 
 	FileSystem fileSystem;
-	private boolean isReady = false;
 
 	private ThreadLocal<Map<String, Buffer>> bufferLocal = ThreadLocal.withInitial(HashMap::new);
 	private ThreadLocal<SimpleDateFormat> dayFormat = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd"));
 	private ThreadLocal<SimpleDateFormat> timeFormat = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss-SSS"));
 
-	@Setter
 	@Getter
 	private class Buffer {
 		long lastFlush;
@@ -107,20 +102,21 @@ public class HdfsService implements DbStoreService {
 			}
 		}
 
-		public void addData(List<Pair<Long, String>> messages) {
-			if (data.isEmpty()) { //reset the last flush time stamp to current if no existing data in buffer
-				lastFlush = System.currentTimeMillis();
-			}
-
-			messages.stream().forEach(message -> data.add(message.getRight()));//note that message left is not used			
-		}
-
+		/*
+				public void addData(List<Pair<Long, String>> messages) {
+					if (data.isEmpty()) { //reset the last flush time stamp to current if no existing data in buffer
+						lastFlush = System.currentTimeMillis();
+					}
+		
+					messages.stream().forEach(message -> data.add(message.getRight()));//note that message left is not used			
+				}
+		*/
 		public void addData2(List<JSONObject> messages) {
 			if (data.isEmpty()) { //reset the last flush time stamp to current if no existing data in buffer
 				lastFlush = System.currentTimeMillis();
 			}
 
-			messages.stream().forEach(message -> data.add(message.toString()));	
+			messages.stream().forEach(message -> data.add(message.toString()));
 		}
 
 		private void saveMessages(String topic, List<String> bufferList) throws IOException {
@@ -157,9 +153,10 @@ public class HdfsService implements DbStoreService {
 	public HdfsService(Db db) {
 		hdfs = db;
 	}
-	
+
 	@PostConstruct
-	private void init() {
+	@Override
+	public void init() {
 		// Initialize HDFS Connection 
 		try {
 			//Get configuration of Hadoop system
@@ -179,10 +176,8 @@ public class HdfsService implements DbStoreService {
 			ShutdownHookManager hadoopShutdownHookManager = ShutdownHookManager.get();
 			hadoopShutdownHookManager.clearShutdownHooks();
 
-			isReady = true;
 		} catch (Exception ex) {
 			log.error("error connection to HDFS.", ex);
-			isReady = false;
 		}
 	}
 
@@ -212,22 +207,23 @@ public class HdfsService implements DbStoreService {
 		bufferLocal.get().forEach((topic, buffer) -> buffer.flushStall(topic));
 	}
 
-	//used if raw data should be saved
-	public void saveMessages(EffectiveTopic topic, List<Pair<Long, String>> messages) {
-		String topicStr = topic.getName();
-
-		Map<String, Buffer> bufferMap = bufferLocal.get();
-		final Buffer buffer = bufferMap.computeIfAbsent(topicStr, k -> new Buffer());
-
-		buffer.addData(messages);
-
-		if (!config.isAsync() || buffer.getData().size() >= config.getHdfsBatchSize()) {
-			buffer.flush(topicStr);
-		} else {
-			log.debug("buffer size too small to flush {}: bufferData.size() {} < config.getHdfsBatchSize() {}", topicStr, buffer.getData().size(), config.getHdfsBatchSize());
+	/*
+		//used if raw data should be saved
+		public void saveMessages(EffectiveTopic topic, List<Pair<Long, String>> messages) {
+			String topicStr = topic.getName();
+	
+			Map<String, Buffer> bufferMap = bufferLocal.get();
+			final Buffer buffer = bufferMap.computeIfAbsent(topicStr, k -> new Buffer());
+	
+			buffer.addData(messages);
+	
+			if (!config.isAsync() || buffer.getData().size() >= config.getHdfsBatchSize()) {
+				buffer.flush(topicStr);
+			} else {
+				log.debug("buffer size too small to flush {}: bufferData.size() {} < config.getHdfsBatchSize() {}", topicStr, buffer.getData().size(), config.getHdfsBatchSize());
+			}
 		}
-	}
-
+	*/
 	@Override
 	public void saveJsons(EffectiveTopic topic, List<JSONObject> jsons) {
 		String topicStr = topic.getName();
@@ -242,7 +238,7 @@ public class HdfsService implements DbStoreService {
 		} else {
 			log.debug("buffer size too small to flush {}: bufferData.size() {} < config.getHdfsBatchSize() {}", topicStr, buffer.getData().size(), config.getHdfsBatchSize());
 		}
-		
+
 	}
- 	
+
 }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/MongodbService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/MongodbService.java
index 5cc4070d..a044790e 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/MongodbService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/db/MongodbService.java
@@ -48,6 +48,7 @@ import com.mongodb.MongoClient;
 import com.mongodb.MongoClientOptions;
 import com.mongodb.MongoClientOptions.Builder;
 import com.mongodb.MongoCredential;
+import com.mongodb.MongoTimeoutException;
 import com.mongodb.ServerAddress;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
@@ -84,7 +85,8 @@ public class MongodbService implements DbStoreService {
 	}
 	
 	@PostConstruct
-	private void init() {
+	@Override
+	public void init() {
 		String host = mongodb.getHost();
 
 		Integer port = mongodb.getPort();
@@ -172,6 +174,8 @@ public class MongodbService implements DbStoreService {
 			for (BulkWriteError bulkWriteError : bulkWriteErrors) {
 				log.error("Failed record: {}", bulkWriteError);
 			}
+		} catch (MongoTimeoutException e) {
+			log.error("saveJsons()", e);			
 		}
 
 		log.debug("saved text to effectiveTopic = {}, batch count = {} ", effectiveTopic, jsons.size());
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/TopicControllerTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/TopicControllerTest.java
index 4fdcf94a..d55e6457 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/TopicControllerTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/TopicControllerTest.java
@@ -31,10 +31,13 @@ import org.onap.datalake.feeder.controller.domain.PostReturnBody;
 import org.onap.datalake.feeder.dto.TopicConfig;
 import org.onap.datalake.feeder.domain.Db;
 import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.repository.TopicNameRepository;
 import org.onap.datalake.feeder.repository.TopicRepository;
 import org.onap.datalake.feeder.service.DbService;
 import org.onap.datalake.feeder.service.DmaapService;
 import org.onap.datalake.feeder.service.TopicService;
+import org.onap.datalake.feeder.util.TestUtil;
 import org.springframework.validation.BindingResult;
 
 import javax.servlet.http.HttpServletResponse;
@@ -53,115 +56,123 @@ import static org.mockito.Mockito.when;
 @RunWith(MockitoJUnitRunner.class)
 public class TopicControllerTest {
 
-    static String DEFAULT_TOPIC_NAME = "_DL_DEFAULT_";
-
-    @Mock
-    private HttpServletResponse httpServletResponse;
-
-    @Mock
-    private BindingResult mockBindingResult;
-
-    @Mock
-    private TopicRepository topicRepository;
-
-    @Mock
-
-    private TopicService topicServiceMock;
-
-    @InjectMocks
-    private TopicService topicService1;
-
-    @Mock
-    private ApplicationConfiguration config;
-
-    @Mock
-    private DbService dbService1;
-
-    @Mock
-    private DmaapService dmaapService1;
-
-    @Before
-    public void setupTest() {
-        MockitoAnnotations.initMocks(this);
-        // While the default boolean return value for a mock is 'false',
-        // it's good to be explicit anyway:
-        when(mockBindingResult.hasErrors()).thenReturn(false);
-    }
-
-    public void setAccessPrivateFields(TopicController topicController) throws NoSuchFieldException,
-            IllegalAccessException {
-        Field topicService = topicController.getClass().getDeclaredField("topicService");
-        topicService.setAccessible(true);
-        topicService.set(topicController, topicService1);
-        Field topicRepository1 = topicController.getClass().getDeclaredField("topicRepository");
-        topicRepository1.setAccessible(true);
-        topicRepository1.set(topicController, topicRepository);
-//        Field dbService = topicController.getClass().getDeclaredField("dbService");
-  //      dbService.setAccessible(true);
-    //    dbService.set(topicController, dbService1);
-    }
-
-    @Test
-    public void testListTopic() throws IOException, NoSuchFieldException, IllegalAccessException{
-        TopicController topicController = new TopicController();
-        setAccessPrivateFields(topicController);
-    }
-
-    //@Test
-    public void testCreateTopic() throws IOException, NoSuchFieldException, IllegalAccessException {
-        TopicController topicController = new TopicController();
-        setAccessPrivateFields(topicController);
-        //when(topicRepository.findById("ab")).thenReturn(Optional.of(new Topic("ab")));
-       // when(config.getDefaultTopicName()).thenReturn(DEFAULT_TOPIC_NAME);
-        PostReturnBody<TopicConfig> postTopic = topicController.createTopic(new TopicConfig(), mockBindingResult, httpServletResponse);
-        assertEquals(postTopic.getStatusCode(), 200);
-        when(mockBindingResult.hasErrors()).thenReturn(true);
-        PostReturnBody<TopicConfig> topicConfig= topicController.createTopic(new TopicConfig(), mockBindingResult, httpServletResponse);
-        assertEquals(null, topicConfig);
-        when(mockBindingResult.hasErrors()).thenReturn(false);
-        TopicConfig a = new TopicConfig();
-        a.setName(DEFAULT_TOPIC_NAME);
-        //when(topicRepository.findById(DEFAULT_TOPIC_NAME)).thenReturn(Optional.of(new Topic(DEFAULT_TOPIC_NAME)));
-        PostReturnBody<TopicConfig> postTopic2= topicController.createTopic(a, mockBindingResult, httpServletResponse);
-        //assertEquals(null, postTopic2);
-    }
-
-    @Test
-    public void testUpdateTopic() throws IOException, NoSuchFieldException, IllegalAccessException {
-        TopicController topicController = new TopicController();
-        setAccessPrivateFields(topicController);
-        PostReturnBody<TopicConfig> postTopic = topicController.updateTopic(1, new TopicConfig(), mockBindingResult, httpServletResponse);
-        assertEquals(null, postTopic);
-        Topic a = new Topic();
-        a.setId(1);
-        //when(topicRepository.findById(1)).thenReturn(Optional.of(a));
-        TopicConfig ac = new TopicConfig();
-        ac.setName("a");
-        ac.setEnabled(true);
-        PostReturnBody<TopicConfig> postConfig1 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
-        //assertEquals(200, postConfig1.getStatusCode());
-        assertNull(postConfig1);
-        //TopicConfig ret = postConfig1.getReturnBody();
-        //assertEquals("a", ret.getName());
-        //assertEquals(true, ret.isEnabled());
-        when(mockBindingResult.hasErrors()).thenReturn(true);
-        PostReturnBody<TopicConfig> postConfig2 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
-        assertEquals(null, postConfig2);
-
-    }
-
-    //@Test
-    public void testListDmaapTopics() throws NoSuchFieldException, IllegalAccessException, IOException {
-        TopicController topicController = new TopicController();
-        Field dmaapService = topicController.getClass().getDeclaredField("dmaapService");
-        dmaapService.setAccessible(true);
-        dmaapService.set(topicController, dmaapService1);
-        ArrayList<String> topics = new ArrayList<>();
-        topics.add("a");
-        when(dmaapService1.getTopics()).thenReturn(topics);
-        List<String> strings = topicController.listDmaapTopics("KAFKA");
-        for (String topic : strings) {
-            assertEquals("a", topic);
-        }
-    }
+	static String DEFAULT_TOPIC_NAME = "_DL_DEFAULT_";
+
+	@Mock
+	private HttpServletResponse httpServletResponse;
+
+	@Mock
+	private BindingResult mockBindingResult;
+
+	@Mock
+	private TopicRepository topicRepository;
+
+	@Mock
+	private TopicService topicService;
+
+	@Mock
+	private TopicNameRepository topicNameRepository;
+
+	@InjectMocks
+	TopicController topicController;
+
+	@Mock
+	private ApplicationConfiguration config;
+
+	@Mock
+	private DbService dbService;
+
+	@Mock
+	private DmaapService dmaapService;
+
+	@Before
+	public void setupTest() throws NoSuchFieldException, IllegalAccessException {
+		// While the default boolean return value for a mock is 'false',
+		// it's good to be explicit anyway:
+		when(mockBindingResult.hasErrors()).thenReturn(false);
+	}
+
+	@Test
+	public void testListTopic() throws IOException, NoSuchFieldException, IllegalAccessException {
+	}
+
+	@Test
+	public void testCreateTopic() throws IOException {
+		Topic a = TestUtil.newTopic("a");
+		a.setId(1);
+		a.setEnabled(true);
+
+		TopicConfig ac = a.getTopicConfig();
+
+		when(topicService.fillTopicConfiguration(ac)).thenReturn(a);
+		PostReturnBody<TopicConfig> postTopic = topicController.createTopic(ac, mockBindingResult, httpServletResponse);
+		assertEquals(postTopic.getStatusCode(), 200);
+
+		when(mockBindingResult.hasErrors()).thenReturn(true);
+		PostReturnBody<TopicConfig> topicConfig = topicController.createTopic(ac, mockBindingResult, httpServletResponse);
+		assertEquals(null, topicConfig);
+	}
+
+	@Test
+	public void testUpdateTopic() throws IOException {
+		Topic a = TestUtil.newTopic("a");
+		a.setId(1);
+		a.setEnabled(true);
+
+		TopicConfig ac = a.getTopicConfig();
+
+		when(topicService.getTopic(1)).thenReturn(a);
+		PostReturnBody<TopicConfig> postConfig1 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
+		assertEquals(200, postConfig1.getStatusCode());
+		TopicConfig ret = postConfig1.getReturnBody();
+		assertEquals("a", ret.getName());
+		assertEquals(true, ret.isEnabled());
+
+		topicController.updateTopic(0, ac, mockBindingResult, httpServletResponse);
+
+		when(topicService.getTopic(1)).thenReturn(null);
+		topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
+
+		when(mockBindingResult.hasErrors()).thenReturn(true);
+		PostReturnBody<TopicConfig> postConfig2 = topicController.updateTopic(1, ac, mockBindingResult, httpServletResponse);
+		assertNull(postConfig2);
+
+	}
+
+	@Test
+	public void testGetTopic() throws IOException {
+		Topic a = TestUtil.newTopic("a");
+		a.setId(1);
+		a.setEnabled(true);
+
+		when(topicService.getTopic(1)).thenReturn(a);
+		TopicConfig ac = topicController.getTopic(1, httpServletResponse);
+		when(topicService.getTopic(1)).thenReturn(null);
+		ac = topicController.getTopic(1, httpServletResponse);
+	}
+
+	@Test
+	public void testDeleteTopic() throws IOException {
+		Topic a = TestUtil.newTopic("a");
+		a.setId(1);
+		a.setEnabled(true);
+
+		when(topicService.getTopic(1)).thenReturn(a);
+		topicController.deleteTopic(1, httpServletResponse);
+		when(topicService.getTopic(1)).thenReturn(null);
+		topicController.deleteTopic(1, httpServletResponse);
+	}
+
+	@Test
+	public void testList() {
+		ArrayList<Topic> topics = new ArrayList<>();
+		topics.add(TestUtil.newTopic("a"));
+		topics.add(TestUtil.newTopic(DEFAULT_TOPIC_NAME));
+		when(topicRepository.findAll()).thenReturn(topics);
+
+		List<String> strings = topicController.list();
+		for (String topic : strings) {
+			System.out.println(topic);
+		}
+	}
 }
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTest.java
index b7befcf3..0accf5a8 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTest.java
@@ -46,10 +46,24 @@ public class DbTest {
         Db mongoDB2 = TestUtil.newDb("MongoDB");
         assertNotEquals(couchbase.hashCode(), mongoDB.hashCode());
         assertNotEquals(couchbase, mongoDB);
-        assertEquals(mongoDB, mongoDB2);
-        assertFalse(mongoDB2.equals(null));
-        assertFalse(mongoDB2.equals(new Topic()));
+        assertNotEquals(mongoDB, mongoDB2);
+        assertEquals(mongoDB, mongoDB);
+        assertFalse(mongoDB2.equals(null)); 
+        
+        DbType dbType = new DbType("MONGO", "MongoDB");
+        dbType.setTool(false);
+        mongoDB.setDbType(dbType);
 
+        assertNotEquals(mongoDB2, dbType); 
+        assertFalse(mongoDB.isTool()); 
+        assertFalse(mongoDB.isHdfs()); 
+        assertFalse(mongoDB.isElasticsearch()); 
+        assertFalse(mongoDB.isCouchbase()); 
+        assertFalse(mongoDB.isDruid());    
+        assertTrue(mongoDB.isMongoDB());    
+        assertFalse(mongoDB.getDbType().isTool());      
+        System.out.println(mongoDB);
+        
         new Db();
         mongoDB2.setHost("localhost");
         mongoDB2.setPort(1234);
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTypeTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTypeTest.java
index 37c77966..4a75df17 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTypeTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DbTypeTest.java
@@ -27,10 +27,27 @@ public class DbTypeTest {
 
     @Test
     public void test(){
-        DbType dbType = new DbType("123","Elasticsearch");
+        DbType dbType = new DbType("ES","Elasticsearch");
+        
+
+        DbType dbType2 = new DbType("MONGO", "MongoDB");
+        dbType.setTool(false); 
+        
+
         assertNotNull(dbType.toString());
         assertEquals(dbType, dbType);
+        assertNotEquals(dbType, null);
+        assertNotEquals(dbType, "ES");
+        assertNotEquals(dbType, dbType2);
         assertNotNull(dbType.hashCode());
+
+        assertEquals("MongoDB", dbType2.getName());
+        dbType2.setName(null);
+        dbType2.setDefaultPort(1);
+        assertTrue(1==dbType2.getDefaultPort());
+
+        dbType2.setDbs(null);
+        assertNull(dbType2.getDbs());
     }
 
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DesignTypeTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DesignTypeTest.java
index 5f3b5993..ad930630 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DesignTypeTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/DesignTypeTest.java
@@ -27,17 +27,23 @@ import static org.junit.Assert.*;
 public class DesignTypeTest {
 
     @Test
-    public void testIs(){
-
+    public void test(){
         DesignType designType = new DesignType();
         designType.setName("Kibana Dashboard");
         designType.setNote("test");
         Portal portal = new Portal();
         portal.setName("Kibana");
         designType.setPortal(portal);
-        assertTrue("Kibana Dashboard".equals(designType.getName()));
-        assertTrue("test".equals(designType.getNote()));
-        assertFalse("Kibana".equals(designType.getPortal()));
-    }
+        assertEquals("Kibana Dashboard", designType.getName());
+        assertEquals("test", designType.getNote());
+        assertNotEquals("Kibana", designType.getPortal());
 
+        designType.setDbType(null);
+        designType.getDbType();
+
+        designType.setDesigns(null);
+        designType.getDesigns();
+        
+        designType.getDesignTypeConfig();
+    }
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/EffectiveTopicTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/EffectiveTopicTest.java
new file mode 100644
index 00000000..cb02f1d7
--- /dev/null
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/EffectiveTopicTest.java
@@ -0,0 +1,48 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP : DataLake
+ * ================================================================================
+ * Copyright 2019 China Mobile
+ *=================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.datalake.feeder.domain;
+
+import org.junit.Test;
+import org.onap.datalake.feeder.util.TestUtil;
+
+/**
+ * Test TopicName
+ *
+ */
+
+public class EffectiveTopicTest {
+
+
+	@Test
+	public void test() {
+
+		Topic topic = TestUtil.newTopic("test Topic");
+		
+		EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+		effectiveTopic = new EffectiveTopic(topic);
+		effectiveTopic.getName();
+		effectiveTopic.setName("");
+		effectiveTopic.getName();
+		effectiveTopic.setTopic(topic);
+		effectiveTopic.getTopic();
+		System.out.println(effectiveTopic);
+	}
+
+}
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/PortalDesignTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/PortalDesignTest.java
index 304628e2..9c3e8008 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/PortalDesignTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/PortalDesignTest.java
@@ -42,6 +42,7 @@ public class PortalDesignTest {
         designType.setName("Kibana");
         portalDesign.setDesignType(designType);
         portalDesign.setNote("test");
+        portalDesign.setDbs(null);
         assertFalse("1".equals(portalDesign.getId()));
         assertTrue("templateTest".equals(portalDesign.getName()));
         assertTrue("jsonString".equals(portalDesign.getBody()));
@@ -49,6 +50,7 @@ public class PortalDesignTest {
         assertTrue("test".equals(portalDesign.getNote()));
         assertFalse("Kibana".equals(portalDesign.getDesignType()));
         assertFalse("false".equals(portalDesign.getSubmitted()));
+        assertNull(portalDesign.getDbs());
     }
 
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/TopicNameTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/TopicNameTest.java
new file mode 100644
index 00000000..d031b314
--- /dev/null
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/TopicNameTest.java
@@ -0,0 +1,51 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP : DataLake
+ * ================================================================================
+ * Copyright 2019 China Mobile
+ *=================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.datalake.feeder.domain;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import org.junit.Test;
+
+/**
+ * Test TopicName
+ *
+ * @author Guobiao Mo
+ */
+
+public class TopicNameTest {
+
+
+	@Test
+	public void test() {
+		TopicName topicName = new TopicName("tes");
+		topicName.setDesigns(null);
+		topicName.getDesigns();
+		topicName.setTopics(null);
+		topicName.getTopics();
+		topicName.hashCode();
+
+		assertEquals(topicName, topicName);
+		assertNotEquals(topicName, null);
+		assertNotEquals(topicName, "test");
+
+	}
+
+}
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/TopicTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/TopicTest.java
index 51e472fe..a018f909 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/TopicTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/domain/TopicTest.java
@@ -19,7 +19,9 @@
  */
 package org.onap.datalake.feeder.domain;
 
+import org.json.JSONObject;
 import org.junit.Test;
+import org.onap.datalake.feeder.dto.TopicConfig;
 import org.onap.datalake.feeder.enumeration.DataFormat;
 import org.onap.datalake.feeder.util.TestUtil;
 
@@ -28,6 +30,7 @@ import java.util.HashSet;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -38,6 +41,41 @@ import static org.junit.Assert.assertTrue;
 
 public class TopicTest {
 
+
+	@Test
+	public void getMessageId() {
+		String text = "{ data: { data2 : { value : 'hello'}}}";
+
+		JSONObject json = new JSONObject(text);
+
+		Topic topic = TestUtil.newTopic("test getMessageId");
+		topic.setMessageIdPath("/data/data2/value");
+	}
+
+	@Test
+	public void getMessageIdFromMultipleAttributes() {
+		String text = "{ data: { data2 : { value : 'hello'}, data3 : 'world'}}";
+
+		JSONObject json = new JSONObject(text);
+
+		Topic topic = TestUtil.newTopic("test getMessageId");
+		topic.setMessageIdPath("/data/data2/value,/data/data3");
+
+		assertEquals("hello^world", topic.getMessageId(json));
+		
+		topic.setMessageIdPath("");
+		assertNull(topic.getMessageId(json));
+	}
+/*
+	@Test
+	public void testArrayPath() {
+		Topic topic = TestUtil.newTopic("testArrayPath");
+		topic.setAggregateArrayPath("/data/data2/value,/data/data3");
+		topic.setFlattenArrayPath("/data/data2/value,/data/data3");
+
+		TopicConfig topicConfig = topic.getTopicConfig();
+	}
+ 
     @Test
     public void getMessageIdFromMultipleAttributes() {
         Topic topic = TestUtil.newTopic("test getMessageId"); 
@@ -62,7 +100,29 @@ public class TopicTest {
         assertFalse(topic.equals(null));
         assertFalse(topic.equals(new Db()));
     }
+*/
+    @Test
+    public void testAggregate() {
+        Topic defaultTopic = TestUtil.newTopic("_DL_DEFAULT_");
+        Topic testTopic = TestUtil.newTopic("test");
+        testTopic.setId(1);
+        Topic testTopic2 = TestUtil.newTopic("test2");
+        testTopic2.setId(2);
+    	
+        //test null cases
+        testTopic.getAggregateArrayPath2() ;
+        testTopic.getFlattenArrayPath2() ;
 
+        //test not null cases
+        testTopic.setAggregateArrayPath("/data/data2/value,/data/data3");
+        testTopic.setFlattenArrayPath("/data/data2/value,/data/data3");
+
+        testTopic.getAggregateArrayPath2() ;
+        testTopic.getFlattenArrayPath2() ;
+        
+    }
+    
+    
     @Test
     public void testIs() {
         Topic defaultTopic = TestUtil.newTopic("_DL_DEFAULT_");
@@ -71,7 +131,9 @@ public class TopicTest {
         Topic testTopic2 = TestUtil.newTopic("test2");
         testTopic2.setId(1);
 
-        assertTrue(testTopic.equals(testTopic2));
+        assertEquals(testTopic, testTopic2);
+        assertNotEquals(testTopic, null);
+        assertNotEquals(testTopic, "test");
         assertEquals(testTopic.hashCode(), testTopic2.hashCode());
         assertNotEquals(testTopic.toString(), "test");
 
@@ -87,12 +149,22 @@ public class TopicTest {
         assertTrue(defaultTopic.isEnabled());
         assertTrue(defaultTopic.isSaveRaw());
 
-        //assertEquals(defaultTopic.getTopicConfig().getDataFormat2(), DataFormat.XML);
+        assertEquals(defaultTopic.getDataFormat2(), DataFormat.XML);
+        defaultTopic.setDataFormat(null);
+        assertNull(defaultTopic.getDataFormat2());
 
         defaultTopic.setDataFormat(null);
         assertEquals(testTopic.getDataFormat(), null);
 
         Topic testTopic1 = TestUtil.newTopic("test");
         assertFalse(testTopic1.isCorrelateClearedMessage());
+        
+
+        testTopic.setPass("root123");
+        assertTrue("root123".equals(testTopic.getPass()));
+        
+        assertEquals(3650, testTopic.getTtl());
+        defaultTopic.setTtl(20);
+        assertEquals(20, defaultTopic.getTtl());
     }
 }
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/DbConfigTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/DbConfigTest.java
index 41dfd820..89a4cc4b 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/DbConfigTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/DbConfigTest.java
@@ -24,12 +24,15 @@ import org.junit.Test;
 import org.onap.datalake.feeder.dto.DbConfig;
 
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 
 public class DbConfigTest {
     @Test
     public void testDbConfig() {
         DbConfig dbConfig = new DbConfig();
+        dbConfig.setId(1);
+        assertEquals(1, dbConfig.getId());
         dbConfig.setName("elasticsearch");
         assertTrue("elasticsearch".equals(dbConfig.getName()));
         dbConfig.setHost("localhost");
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalConfigTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalConfigTest.java
index ead28e21..4709b04a 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalConfigTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalConfigTest.java
@@ -47,5 +47,6 @@ public class PortalConfigTest {
         assertEquals(testPortalConfig.getEnabled(), null);
         assertEquals(testPortalConfig.getLogin(), null);
         assertEquals(testPortalConfig.getPass(), null);
+        assertEquals(testPortalConfig.getDb(), "Elasticsearch");
     }
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalDesignConfigTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalDesignConfigTest.java
index 49102a15..1d0f2a8d 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalDesignConfigTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/PortalDesignConfigTest.java
@@ -56,6 +56,8 @@ public class PortalDesignConfigTest {
         assertEquals(testPortalDesignConfig.getNote(), null);
         assertEquals(testPortalDesignConfig.getName(), null);
         assertEquals(testPortalDesignConfig.getSubmitted(), null);
+        assertEquals(testPortalDesignConfig.getDesignType(), null);
+        assertEquals(testPortalDesignConfig.getDisplay(), "test");
     }
 
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java
index d9865979..83329e9b 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/dto/TopicConfigTest.java
@@ -19,20 +19,17 @@
  */
 package org.onap.datalake.feeder.dto;
 
-import org.json.JSONObject;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import java.util.HashSet;
+
 import org.junit.Test;
 import org.onap.datalake.feeder.domain.Db;
+import org.onap.datalake.feeder.domain.Kafka;
 import org.onap.datalake.feeder.domain.Topic;
 import org.onap.datalake.feeder.util.TestUtil;
 
-import java.util.HashSet;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
 /**
  * Test Topic
  *
@@ -40,91 +37,41 @@ import static org.junit.Assert.assertTrue;
  */
 
 public class TopicConfigTest {
-
-    @Test
-    public void getMessageId() {
-        String text = "{ data: { data2 : { value : 'hello'}}}";
-
-        JSONObject json = new JSONObject(text);
-
-        Topic topic = TestUtil.newTopic("test getMessageId");
-        topic.setMessageIdPath("/data/data2/value");
-        
-        TopicConfig topicConfig = topic.getTopicConfig();
-
-//        String value = topicConfig.getMessageId(json);
-
-  //      assertEquals(value, "hello");
-    }
-
-    @Test
-    public void getMessageIdFromMultipleAttributes() {
-        String text = "{ data: { data2 : { value : 'hello'}, data3 : 'world'}}";
-
-        JSONObject json = new JSONObject(text);
-
-        Topic topic = TestUtil.newTopic("test getMessageId");
-        topic.setMessageIdPath("/data/data2/value,/data/data3");
-
-        TopicConfig topicConfig = topic.getTopicConfig();
-        
-//        String value = topicConfig.getMessageId(json);
- //       assertEquals(value, "hello^world");
-
-        topic.setMessageIdPath("");
-        topicConfig = topic.getTopicConfig();
- //       assertNull(topicConfig.getMessageId(json));
-
-    }
-
-    @Test
-    public void testArrayPath() {
-        Topic topic = TestUtil.newTopic("testArrayPath");
-        topic.setAggregateArrayPath("/data/data2/value,/data/data3");
-        topic.setFlattenArrayPath("/data/data2/value,/data/data3");
-
-        TopicConfig topicConfig = topic.getTopicConfig();
-/*
-        String[] value = topicConfig.getAggregateArrayPath2();
-        assertEquals(value[0], "/data/data2/value");
-        assertEquals(value[1], "/data/data3");
-
-        value = topicConfig.getFlattenArrayPath2();
-        assertEquals(value[0], "/data/data2/value");
-        assertEquals(value[1], "/data/data3");*/
-    }
-
-    @Test
-    public void testIs() {
-        Topic testTopic = TestUtil.newTopic("test");
-
-        TopicConfig testTopicConfig = testTopic.getTopicConfig();
-        testTopicConfig.setSinkdbs(null);
-        testTopicConfig.setEnabledSinkdbs(null);
-        //assertFalse(testTopicConfig.supportElasticsearch());
-        //assertNull(testTopicConfig.getDataFormat2());
-                
-        testTopic.setDbs(new HashSet<>());
-        Db esDb = TestUtil.newDb("Elasticsearch");
-        esDb.setEnabled(true);
-        testTopic.getDbs().add(esDb);
-        
-        testTopicConfig = testTopic.getTopicConfig();
-
-        //assertEquals(testTopicConfig, new Topic("test").getTopicConfig());
-        assertNotEquals(testTopicConfig, testTopic);
-        assertNotEquals(testTopicConfig, null);
-        //assertEquals(testTopicConfig.hashCode(), (new Topic("test").getTopicConfig()).hashCode());
-        /*
-        assertTrue(testTopicConfig.supportElasticsearch());
-        assertFalse(testTopicConfig.supportCouchbase());
-        assertFalse(testTopicConfig.supportDruid());
-        assertFalse(testTopicConfig.supportMongoDB());
-        assertFalse(testTopicConfig.supportHdfs());
-
-        testTopic.getDbs().remove(new Db("Elasticsearch"));
-        testTopicConfig = testTopic.getTopicConfig();
-        assertFalse(testTopicConfig.supportElasticsearch());
- */
-    }
+	@Test
+	public void testIs() {
+		Topic testTopic = TestUtil.newTopic("test");
+
+		TopicConfig testTopicConfig = testTopic.getTopicConfig();
+		testTopicConfig.setSinkdbs(null);
+		testTopicConfig.setEnabledSinkdbs(null);
+
+		testTopic.setDbs(null);
+		testTopic.setKafkas(null);
+		testTopicConfig = testTopic.getTopicConfig();
+
+		testTopic.setDbs(new HashSet<>());
+		Db esDb = TestUtil.newDb("Elasticsearch");
+		esDb.setEnabled(true);
+		testTopic.getDbs().add(esDb);
+		
+		esDb = TestUtil.newDb("MongoDB");
+		esDb.setEnabled(false);
+		testTopic.getDbs().add(esDb);
+
+
+		testTopic.setKafkas(new HashSet<>());
+		Kafka kafka = TestUtil.newKafka("k1");
+		kafka.setEnabled(true);
+		testTopic.getKafkas().add(kafka);
+		testTopicConfig = testTopic.getTopicConfig();
+		
+		
+		
+		TopicConfig testTopicConfig2 = TestUtil.newTopic("test").getTopicConfig();
+		assertNotEquals(testTopicConfig, testTopicConfig2);
+		assertEquals(testTopicConfig, testTopicConfig);
+		assertNotEquals(testTopicConfig.hashCode(), testTopicConfig2.hashCode());
+		assertNotEquals(testTopicConfig, testTopic);
+		assertNotEquals(testTopicConfig, null);
+	}
 }
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/DmaapServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/DmaapServiceTest.java
index 92c7a69f..cab2c138 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/DmaapServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/DmaapServiceTest.java
@@ -24,30 +24,48 @@ import static org.junit.Assert.assertNotEquals;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.util.TestUtil;
 
 @RunWith(MockitoJUnitRunner.class)
 public class DmaapServiceTest {
 
     static String DMAPP_ZOOKEEPER_HOST_PORT = "test:2181";
 
-    @InjectMocks
     private DmaapService dmaapService;
 
     @Mock
     private ApplicationConfiguration config;
     @Mock
     private TopicService topicService;
-    
+
+	@Before
+	public void init() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+		Kafka kafka = TestUtil.newKafka("kafka"); 
+		dmaapService = new DmaapService(kafka);
+
+		Field configField = DmaapService.class.getDeclaredField("config");
+		configField.setAccessible(true);
+		configField.set(dmaapService, config);
+
+		/*
+		Method initMethod = DmaapService.class.getDeclaredMethod("init");
+		initMethod.setAccessible(true);
+		initMethod.invoke(dmaapService); */
+	}
+	
     @Test
     public void testGetTopics() throws InterruptedException {
         List<String> list = new ArrayList<>();
@@ -60,8 +78,8 @@ public class DmaapServiceTest {
         //when(config.getDmaapZookeeperHostPort()).thenReturn(DMAPP_ZOOKEEPER_HOST_PORT);
         assertNotEquals(list, dmaapService.getTopics());
 
-		//when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
-    	//dmaapService.cleanUp();
+		when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+    	dmaapService.cleanUp();
     }
 
     @Test
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullServiceTest.java
index fc8eb827..d6298b87 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullServiceTest.java
@@ -28,8 +28,10 @@ import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
 import org.springframework.context.ApplicationContext;
 
+import java.lang.reflect.Field;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import static org.junit.Assert.*;
 import static org.mockito.Mockito.when;
@@ -37,36 +39,62 @@ import static org.mockito.Mockito.when;
 @RunWith(MockitoJUnitRunner.class)
 public class PullServiceTest {
 
-    @InjectMocks
-    private PullService pullService;
+	@InjectMocks
+	private PullService pullService;
 
-    @Mock
-    private ApplicationContext context;
+	@Mock
+	private ApplicationContext context;
 
-    @Mock
-    private ApplicationConfiguration config;
+	@Mock
+	private ApplicationConfiguration config;
 
-    @Mock
-    private ExecutorService executorService;
+	@Mock
+	private ExecutorService executorService;
 
-    @Mock
-    private List<Puller> consumers;
+	@Mock
+	private List<Puller> consumers;
 
-    @Test
-    public void isRunning() {
-        assertEquals(pullService.isRunning(), false);
-    }
+	@Test
+	public void isRunning() {
+		assertFalse(pullService.isRunning());
+	}
 
-    @Test(expected = NullPointerException.class)
-    public void start() {
+	@Test(expected = NullPointerException.class)
+	public void start() {
+		setRunning(false);
+		pullService.start();
+		setRunning(true);
+		pullService.start();
+	}
 
-        //when(config.getKafkaConsumerCount()).thenReturn(1);
+	@Test
+	public void shutdown() {
+		when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+		setRunning(false);
+		pullService.shutdown();
+		setRunning(true);
+		pullService.shutdown();
+	}
 
-        pullService.start();
-    }
+	private void setRunning(boolean running) {
+		Field configField;
+		try {
+			configField = PullService.class.getDeclaredField("isRunning");
+			configField.setAccessible(true);
+			configField.set(pullService, running);
+		} catch (IllegalArgumentException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		} catch (IllegalAccessException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
 
-    @Test
-    public void shutdown() {
-        pullService.shutdown();
-    }
+		} catch (NoSuchFieldException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		} catch (SecurityException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullerTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullerTest.java
index 00878d9d..9e9c99e4 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullerTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullerTest.java
@@ -22,15 +22,18 @@ package org.onap.datalake.feeder.service;
 
 import static org.mockito.Mockito.when;
 
+import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.util.TestUtil;
 import org.springframework.context.ApplicationContext;
 
 /**
@@ -44,8 +47,7 @@ import org.springframework.context.ApplicationContext;
 @RunWith(MockitoJUnitRunner.class)
 public class PullerTest {
 
-	@InjectMocks
-	private Puller puller = new Puller(null);
+	private Puller puller;
 
 	@Mock
 	private ApplicationContext context;
@@ -59,26 +61,34 @@ public class PullerTest {
 	@Mock
 	private TopicConfigPollingService topicConfigPollingService;
 
-	public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException {
-		when(config.isAsync()).thenReturn(true);
+	@Before
+	public void init() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+		Kafka kafka = TestUtil.newKafka("kafka");
+		kafka.setBrokerList("brokerList:1,brokerList2:1");
+		kafka.setGroup("group");
+		kafka.setLogin("login");
+		kafka.setSecure(true);
+		kafka.setSecurityProtocol("securityProtocol");
+		puller = new Puller(kafka);
+
+		Field configField = Puller.class.getDeclaredField("config");
+		configField.setAccessible(true);
+		configField.set(puller, config);
 
-		Method init = puller.getClass().getDeclaredMethod("init");
-		init.setAccessible(true);
-		init.invoke(puller);
+		when(config.isAsync()).thenReturn(true);
+		Method initMethod = Puller.class.getDeclaredMethod("init");
+		initMethod.setAccessible(true);
+		initMethod.invoke(puller);
 	}
 
 	@Test
-	public void testRun() throws InterruptedException, IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
-		testInit();
- 
-
+	public void testRun() throws InterruptedException {
 		Thread thread = new Thread(puller);
 		thread.start();
 
 		Thread.sleep(50);
 		puller.shutdown();
 		thread.join();
-
 	}
 
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java
index 0f222dc3..f4781a59 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java
@@ -20,24 +20,31 @@
 
 package org.onap.datalake.feeder.service;
 
-import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.when;
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 
 import org.apache.commons.lang3.tuple.Pair;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.EffectiveTopic;
 import org.onap.datalake.feeder.domain.Kafka;
-import org.onap.datalake.feeder.dto.TopicConfig;
-import org.onap.datalake.feeder.service.db.*;
+import org.onap.datalake.feeder.domain.Topic;
+import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.service.db.CouchbaseService;
+import org.onap.datalake.feeder.service.db.ElasticsearchService;
+import org.onap.datalake.feeder.service.db.HdfsService;
+import org.onap.datalake.feeder.service.db.MongodbService;
+import org.onap.datalake.feeder.util.TestUtil;
 import org.springframework.context.ApplicationContext;
 
 /**
@@ -72,58 +79,61 @@ public class StoreServiceTest {
 
 	@Mock
 	private HdfsService hdfsService;
-	
+
 	@Mock
 	private Kafka kafka;
 
-	public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+	@Before
+	public void init() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
 		Method init = storeService.getClass().getDeclaredMethod("init");
 		init.setAccessible(true);
 		init.invoke(storeService);
 	}
 
-	private TopicConfig createTopicConfig(String topicStr, String type) {
+	private EffectiveTopic createTopicConfig(String topicStr, String type) {
+		Topic topic = new Topic();
+		topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
+		topic.setDataFormat(type);
+		topic.setSaveRaw(true);
+		topic.setEnabled(true);
+		
 
-		TopicConfig topicConfig = new TopicConfig();
-		topicConfig.setName(topicStr);
-		topicConfig.setDataFormat(type);
-		topicConfig.setSaveRaw(true);
+		EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+		List<EffectiveTopic> effectiveTopics = new ArrayList<>();
+		effectiveTopics.add(effectiveTopic);
 
-//		when(configPollingService.getEffectiveTopicConfig(topicStr)).thenReturn(topicConfig);
+		when(configPollingService.getEffectiveTopic(kafka, topicStr)).thenReturn(effectiveTopics);
 
-		return topicConfig;
+		return effectiveTopic;
 	}
 
 	@Test
 	public void saveMessages() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
-		testInit();
-
-		TopicConfig topicConfig = createTopicConfig("test1", "JSON");
-		topicConfig.setAggregateArrayPath("/test");
-		topicConfig.setFlattenArrayPath("/test");
-
-		topicConfig = createTopicConfig("test2", "XML");
-		topicConfig.setSaveRaw(false);
-
-		topicConfig = createTopicConfig("test3", "YAML");
-
-		topicConfig.setSinkdbs(new ArrayList<>());
-		topicConfig.getSinkdbs().add("Elasticsearch");
-		topicConfig.getSinkdbs().add("Couchbase");
-		topicConfig.getSinkdbs().add("Druid");
-		topicConfig.getSinkdbs().add("MongoDB");
-		topicConfig.getSinkdbs().add("HDFS");
-
-
-		topicConfig.setEnabledSinkdbs(new ArrayList<>());
-		topicConfig.getEnabledSinkdbs().add("Elasticsearch");
+		EffectiveTopic effectiveTopic = createTopicConfig("test1", "JSON");
+		effectiveTopic.getTopic().setAggregateArrayPath("/test");
+		effectiveTopic.getTopic().setFlattenArrayPath("/test"); 
+
+		effectiveTopic = createTopicConfig("test2", "XML");
+		effectiveTopic.getTopic().setSaveRaw(false); 
+
+		effectiveTopic = createTopicConfig("test3", "YAML");
+		effectiveTopic.getTopic().setDbs(new HashSet<>());
+		effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("ES"));
+		effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("CB"));
+		effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("DRUID"));
+		effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("MONGO"));
+		effectiveTopic.getTopic().getDbs().add(TestUtil.newDb("HDFS")); 
+		//		effectiveTopic.getTopic().setEnabledSinkdbs(new ArrayList<>());
+		//	effectiveTopic.getTopic().getEnabledSinkdbs().add("Elasticsearch");
 		//assertTrue(topicConfig.supportElasticsearch());
-		
-		
+
 		createTopicConfig("test4", "TEXT");
+		
+		effectiveTopic = createTopicConfig("test5", "TEXT");
+		effectiveTopic.getTopic().setEnabled(false);
 
-//		when(config.getTimestampLabel()).thenReturn("ts");
-//		when(config.getRawDataLabel()).thenReturn("raw");
+		when(config.getTimestampLabel()).thenReturn("ts");
+		when(config.getRawDataLabel()).thenReturn("raw");
 
 		//JSON
 		List<Pair<Long, String>> messages = new ArrayList<>();
@@ -133,7 +143,7 @@ public class StoreServiceTest {
 
 		//XML
 		List<Pair<Long, String>> messagesXml = new ArrayList<>();
-		messagesXml.add(Pair.of(100L, "<test></test>")); 
+		messagesXml.add(Pair.of(100L, "<test></test>"));
 		messagesXml.add(Pair.of(100L, "<test></test"));//bad xml to trigger exception
 
 		storeService.saveMessages(kafka, "test2", messagesXml);
@@ -149,6 +159,8 @@ public class StoreServiceTest {
 		messagesText.add(Pair.of(100L, "test message"));
 
 		storeService.saveMessages(kafka, "test4", messagesText);
+		
+		storeService.saveMessages(kafka, "test5", messagesText);
 
 		//Null mesg
 		storeService.saveMessages(kafka, "test", null);
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/TopicConfigPollingServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/TopicConfigPollingServiceTest.java
index 731b9a29..04545a9a 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/TopicConfigPollingServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/TopicConfigPollingServiceTest.java
@@ -20,8 +20,7 @@
 
 package org.onap.datalake.feeder.service;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.when;
 
@@ -29,8 +28,12 @@ import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Arrays;
-import java.util.List;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
 
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
@@ -38,6 +41,7 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
 import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.util.TestUtil;
 
 /**
  * Test TopicConfigPollingService
@@ -56,28 +60,31 @@ public class TopicConfigPollingServiceTest {
 	@InjectMocks
 	private TopicConfigPollingService topicConfigPollingService = new TopicConfigPollingService();
 
-	@Test
-	public void testRun() {
-		
-	}
-	
-	/*
-	public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+	static String KAFKA_NAME = "kafka1";
+
+	@Before
+	public void init() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
 		Method init = topicConfigPollingService.getClass().getDeclaredMethod("init");
 		init.setAccessible(true);
 		init.invoke(topicConfigPollingService);
 
-		List<String> activeTopics = Arrays.asList("test");
-		Field activeTopicsField = topicConfigPollingService.getClass().getDeclaredField("activeTopics");
+		Set<String> activeTopics = new HashSet<>(Arrays.asList("test"));
+		Map<String, Set<String>> activeTopicMap = new HashMap<>();
+		activeTopicMap.put(KAFKA_NAME, activeTopics);
+
+		Field activeTopicsField = TopicConfigPollingService.class.getDeclaredField("activeTopicMap");
 		activeTopicsField.setAccessible(true);
-		activeTopicsField.set(topicConfigPollingService, activeTopics);
+		activeTopicsField.set(topicConfigPollingService, activeTopicMap);
+
+		Method initMethod = TopicConfigPollingService.class.getDeclaredMethod("init");
+		initMethod.setAccessible(true);
+		initMethod.invoke(topicConfigPollingService);
 	}
 
 	@Test
-	public void testRun() throws InterruptedException, IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
-		testInit();
+	public void testRun() throws InterruptedException {
 
-		//when(config.getDmaapCheckNewTopicInterval()).thenReturn(1);
+		when(config.getCheckTopicInterval()).thenReturn(1L);
 
 		Thread thread = new Thread(topicConfigPollingService);
 		thread.start();
@@ -91,8 +98,8 @@ public class TopicConfigPollingServiceTest {
 
 	@Test
 	public void testRunNoChange() throws InterruptedException {
-	
-//		when(config.getDmaapCheckNewTopicInterval()).thenReturn(1);
+
+		when(config.getCheckTopicInterval()).thenReturn(1L);
 
 		Thread thread = new Thread(topicConfigPollingService);
 		thread.start();
@@ -101,15 +108,15 @@ public class TopicConfigPollingServiceTest {
 		topicConfigPollingService.shutdown();
 		thread.join();
 
-		assertFalse(topicConfigPollingService.isActiveTopicsChanged(new Kafka()));
+		assertTrue(topicConfigPollingService.isActiveTopicsChanged(new Kafka()));
 	}
 
 	@Test
 	public void testGet() {
-		Kafka kafka=null;
-		assertNull(topicConfigPollingService.getEffectiveTopic (new Kafka(), "test"));
-		assertNull(topicConfigPollingService.getActiveTopics(kafka));
+		Kafka kafka = TestUtil.newKafka(KAFKA_NAME);
+		//assertNull(topicConfigPollingService.getEffectiveTopic (kafka, "test"));
+		assertNotNull(topicConfigPollingService.getActiveTopics(kafka));
 
 	}
-	*/
+
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/CouchbaseServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/CouchbaseServiceTest.java
index 911ae26b..e43d30d5 100755
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/CouchbaseServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/CouchbaseServiceTest.java
@@ -20,13 +20,9 @@
 
 package org.onap.datalake.feeder.service.db;
 
-import com.couchbase.client.java.Cluster;
-import com.couchbase.client.java.CouchbaseCluster;
-import com.couchbase.client.java.env.DefaultCouchbaseEnvironment;
-import com.couchbase.mock.Bucket;
-import com.couchbase.mock.BucketConfiguration;
-import com.couchbase.mock.CouchbaseMock;
-import com.couchbase.mock.client.MockClient;
+import java.util.ArrayList;
+import java.util.List;
+
 import org.jetbrains.annotations.NotNull;
 import org.json.JSONObject;
 import org.junit.After;
@@ -36,121 +32,127 @@ import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
 import org.onap.datalake.feeder.domain.Db;
-import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.domain.EffectiveTopic;
 import org.onap.datalake.feeder.domain.Topic;
-import org.onap.datalake.feeder.service.db.CouchbaseService;
 import org.onap.datalake.feeder.util.TestUtil;
 
-import static org.mockito.Mockito.when;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import com.couchbase.client.java.Cluster;
+import com.couchbase.client.java.CouchbaseCluster;
+import com.couchbase.client.java.env.DefaultCouchbaseEnvironment;
+import com.couchbase.mock.Bucket;
+import com.couchbase.mock.BucketConfiguration;
+import com.couchbase.mock.CouchbaseMock;
+import com.couchbase.mock.client.MockClient;
 
 @RunWith(MockitoJUnitRunner.class)
 public class CouchbaseServiceTest {
-    protected final BucketConfiguration bucketConfiguration = new BucketConfiguration();
-    protected MockClient mockClient;
-    protected CouchbaseMock couchbaseMock;
-    protected Cluster cluster;
-    protected com.couchbase.client.java.Bucket bucket;
-    protected int carrierPort;
-    protected int httpPort;
-
-    protected void getPortInfo(String bucket) throws Exception {
-        httpPort = couchbaseMock.getHttpPort();
-        carrierPort = couchbaseMock.getCarrierPort(bucket);
-    }
-
-    protected void createMock(@NotNull String name, @NotNull String password) throws Exception {
-        bucketConfiguration.numNodes = 1;
-        bucketConfiguration.numReplicas = 1;
-        bucketConfiguration.numVBuckets = 1024;
-        bucketConfiguration.name = name;
-        bucketConfiguration.type = Bucket.BucketType.COUCHBASE;
-        bucketConfiguration.password = password;
-        ArrayList<BucketConfiguration> configList = new ArrayList<BucketConfiguration>();
-        configList.add(bucketConfiguration);
-        couchbaseMock = new CouchbaseMock(0, configList);
-        couchbaseMock.start();
-        couchbaseMock.waitForStartup();
-    }
-
-    protected void createClient() {
-        cluster = CouchbaseCluster.create(DefaultCouchbaseEnvironment.builder()
-                                                  .bootstrapCarrierDirectPort(carrierPort)
-                                                  .bootstrapHttpDirectPort(httpPort)
-                                                  .build(), "couchbase://127.0.0.1");
-        bucket = cluster.openBucket("default");
-    }
-
-    @Before
-    public void setUp() throws Exception {
-        createMock("default", "");
-        getPortInfo("default");
-        createClient();
-    }
-
-    @After
-    public void tearDown() {
-        if (cluster != null) {
-            cluster.disconnect();
-        }
-        if (couchbaseMock != null) {
-            couchbaseMock.stop();
-        }
-        if (mockClient != null) {
-            mockClient.shutdown();
-        }
-    }
-
-    @Test
-    public void testSaveJsonsWithTopicId() {
-    	ApplicationConfiguration appConfig = new ApplicationConfiguration();
-    	appConfig.setTimestampLabel("datalake_ts_");
-
-        String text = "{ data: { data2 : { value : 'hello'}}}";
-
-        JSONObject json = new JSONObject(text);
-
-        Topic topic = TestUtil.newTopic("test getMessageId");
-        topic.setMessageIdPath("/data/data2/value");
-        List<JSONObject> jsons = new ArrayList<>();
-        json.put(appConfig.getTimestampLabel(), 1234);
-        jsons.add(json);
-        CouchbaseService couchbaseService = new CouchbaseService(new Db());
-        couchbaseService.bucket = bucket;
-        couchbaseService.config = appConfig;
- //       couchbaseService.saveJsons(topic.getTopicConfig(), jsons);
-
-    }
-
-    @Test
-    public void testSaveJsonsWithOutTopicId() {
-    	ApplicationConfiguration appConfig = new ApplicationConfiguration();
-    	appConfig.setTimestampLabel("datalake_ts_");
-
-        String text = "{ data: { data2 : { value : 'hello'}}}";
-
-        JSONObject json = new JSONObject(text);
-
-        Topic topic = TestUtil.newTopic("test getMessageId");
-        List<JSONObject> jsons = new ArrayList<>();
-        json.put(appConfig.getTimestampLabel(), 1234);
-        jsons.add(json);
-        CouchbaseService couchbaseService = new CouchbaseService(new Db());
-        couchbaseService.bucket = bucket;
-        couchbaseService.config = appConfig;
-//        couchbaseService.saveJsons(topic.getTopicConfig(), jsons);
-    }
-
-    @Test
-    public void testCleanupBucket() {
-        CouchbaseService couchbaseService = new CouchbaseService(new Db());
-        couchbaseService.bucket = bucket;
-    	ApplicationConfiguration appConfig = new ApplicationConfiguration();
-        couchbaseService.config = appConfig;
-        couchbaseService.cleanUp();
-    }
+	protected final BucketConfiguration bucketConfiguration = new BucketConfiguration();
+	protected MockClient mockClient;
+	protected CouchbaseMock couchbaseMock;
+	protected Cluster cluster;
+	protected com.couchbase.client.java.Bucket bucket;
+	protected int carrierPort;
+	protected int httpPort;
+
+	protected void getPortInfo(String bucket) throws Exception {
+		httpPort = couchbaseMock.getHttpPort();
+		carrierPort = couchbaseMock.getCarrierPort(bucket);
+	}
+
+	protected void createMock(@NotNull String name, @NotNull String password) throws Exception {
+		bucketConfiguration.numNodes = 1;
+		bucketConfiguration.numReplicas = 1;
+		bucketConfiguration.numVBuckets = 1024;
+		bucketConfiguration.name = name;
+		bucketConfiguration.type = Bucket.BucketType.COUCHBASE;
+		bucketConfiguration.password = password;
+		ArrayList<BucketConfiguration> configList = new ArrayList<BucketConfiguration>();
+		configList.add(bucketConfiguration);
+		couchbaseMock = new CouchbaseMock(0, configList);
+		couchbaseMock.start();
+		couchbaseMock.waitForStartup();
+	}
+
+	protected void createClient() {
+		cluster = CouchbaseCluster.create(DefaultCouchbaseEnvironment.builder().bootstrapCarrierDirectPort(carrierPort).bootstrapHttpDirectPort(httpPort).build(), "couchbase://127.0.0.1");
+		bucket = cluster.openBucket("default");
+	}
+
+	@Before
+	public void setUp() throws Exception {
+		createMock("default", "");
+		getPortInfo("default");
+		createClient();
+	}
+
+	@After
+	public void tearDown() {
+		if (cluster != null) {
+			cluster.disconnect();
+		}
+		if (couchbaseMock != null) {
+			couchbaseMock.stop();
+		}
+		if (mockClient != null) {
+			mockClient.shutdown();
+		}
+	}
+
+	@Test
+	public void testSaveJsonsWithTopicId() {
+		ApplicationConfiguration appConfig = new ApplicationConfiguration();
+		appConfig.setTimestampLabel("datalake_ts_");
+
+		String text = "{ data: { data2 : { value : 'hello'}}}";
+
+		JSONObject json = new JSONObject(text);
+
+		Topic topic = TestUtil.newTopic("test getMessageId");
+		topic.setMessageIdPath("/data/data2/value");
+		List<JSONObject> jsons = new ArrayList<>();
+		json.put(appConfig.getTimestampLabel(), 1234);
+		jsons.add(json);
+		CouchbaseService couchbaseService = new CouchbaseService(new Db());
+		couchbaseService.bucket = bucket;
+		couchbaseService.config = appConfig;
+
+		couchbaseService.init();
+		EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+		couchbaseService.saveJsons(effectiveTopic, jsons);
+
+	}
+
+	@Test
+	public void testSaveJsonsWithOutTopicId() {
+		ApplicationConfiguration appConfig = new ApplicationConfiguration();
+		appConfig.setTimestampLabel("datalake_ts_");
+
+		String text = "{ data: { data2 : { value : 'hello'}}}";
+
+		JSONObject json = new JSONObject(text);
+
+		Topic topic = TestUtil.newTopic("test getMessageId");
+		List<JSONObject> jsons = new ArrayList<>();
+		json.put(appConfig.getTimestampLabel(), 1234);
+		jsons.add(json);
+		CouchbaseService couchbaseService = new CouchbaseService(new Db());
+		couchbaseService.bucket = bucket;
+		couchbaseService.config = appConfig;
+
+		couchbaseService.init();
+		EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+		couchbaseService.saveJsons(effectiveTopic, jsons);
+	}
+
+	@Test
+	public void testCleanupBucket() {
+		CouchbaseService couchbaseService = new CouchbaseService(new Db());
+		couchbaseService.bucket = bucket;
+		ApplicationConfiguration appConfig = new ApplicationConfiguration();
+		couchbaseService.config = appConfig;
+
+		couchbaseService.init();
+		couchbaseService.cleanUp();
+	}
 
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/ElasticsearchServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/ElasticsearchServiceTest.java
index 4c7c35f6..b1377d1b 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/ElasticsearchServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/ElasticsearchServiceTest.java
@@ -20,79 +20,78 @@
 
 package org.onap.datalake.feeder.service.db;
 
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.client.RestHighLevelClient;
-import org.json.JSONObject;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
-import org.onap.datalake.feeder.domain.Topic;
-import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.domain.Db;
 import org.onap.datalake.feeder.service.DbService;
-import org.onap.datalake.feeder.service.db.ElasticsearchService;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import static org.mockito.Mockito.when;
+import org.onap.datalake.feeder.util.TestUtil;
 
 @RunWith(MockitoJUnitRunner.class)
 public class ElasticsearchServiceTest {
+	@Mock
+	private ApplicationConfiguration config;
 
-    static String DEFAULT_TOPIC_NAME = "_DL_DEFAULT_";
-
-    @InjectMocks
-    private ElasticsearchService elasticsearchService;
-
-    @Mock
-    private ApplicationConfiguration config;
-
-    @Mock
-    private RestHighLevelClient client;
-
-    @Mock
-    ActionListener<BulkResponse> listener;
-
-    @Mock
-    private DbService dbService;
+	@Mock
+	private RestHighLevelClient client;
 
-    @Test(expected = NullPointerException.class)
-    public void testCleanUp() throws IOException {
+	@Mock
+	ActionListener<BulkResponse> listener;
 
-        elasticsearchService.cleanUp();
+	@Mock
+	private DbService dbService;
 
-    }
+	private ElasticsearchService elasticsearchService;
 
-    @Test(expected = NullPointerException.class)
-    public void testEnsureTableExist() throws IOException {
+	@Before
+	public void init() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+		//MockitoAnnotations.initMocks(this);
 
-        elasticsearchService.ensureTableExist(DEFAULT_TOPIC_NAME);
-    }
+		Db db = TestUtil.newDb("Elasticsearch");
+		db.setHost("host");
+		elasticsearchService = new ElasticsearchService(db);
 
-    @Test
-    public void testSaveJsons() {
+		Field configField = ElasticsearchService.class.getDeclaredField("config");
+		configField.setAccessible(true);
+		configField.set(elasticsearchService, config);
+		
+		elasticsearchService.init();
+	}
 
-        Topic topic = new Topic();
-        topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
-        topic.setCorrelateClearedMessage(true);
-        topic.setMessageIdPath("/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem");
-        String jsonString = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailure\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
-        String jsonString2 = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailureCleared\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+	@Test
+	public void testCleanUp() throws IOException {
+		when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+		elasticsearchService.cleanUp();
+	}
 
-        JSONObject jsonObject = new JSONObject(jsonString);
-        JSONObject jsonObject2 = new JSONObject(jsonString2);
+	@Test(expected = IOException.class)
+	public void testEnsureTableExist() throws IOException {
+		elasticsearchService.ensureTableExist("test");
+	}
 
-        List<JSONObject> jsons = new ArrayList<>();
-        jsons.add(jsonObject);
-        jsons.add(jsonObject2);
-//        when(config.getElasticsearchType()).thenReturn("doc");
-  //      when(config.isAsync()).thenReturn(true);
+	@Test
+	public void testSaveJsons() {
+		when(config.getElasticsearchType()).thenReturn("doc");
 
-        //elasticsearchService.saveJsons(topic.getTopicConfig(), jsons);
+		when(config.isAsync()).thenReturn(true);
+		TestUtil.testSaveJsons(config, elasticsearchService);
 
-    }
+		when(config.isAsync()).thenReturn(false);
+		TestUtil.testSaveJsons(config, elasticsearchService);
+	}
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/HdfsServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/HdfsServiceTest.java
index 94721b01..7f159919 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/HdfsServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/HdfsServiceTest.java
@@ -22,19 +22,18 @@ package org.onap.datalake.feeder.service.db;
 
 import static org.mockito.Mockito.when;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.lang.reflect.Field;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
-import org.apache.commons.lang3.tuple.Pair;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
-import org.onap.datalake.feeder.dto.TopicConfig;
-import org.onap.datalake.feeder.service.db.HdfsService;
+import org.onap.datalake.feeder.domain.Db;
+import org.onap.datalake.feeder.util.TestUtil;
 import org.springframework.context.ApplicationContext;
 
 /**
@@ -45,8 +44,6 @@ import org.springframework.context.ApplicationContext;
  */
 @RunWith(MockitoJUnitRunner.class)
 public class HdfsServiceTest {
-
-	@InjectMocks
 	private HdfsService hdfsService;
 
 	@Mock
@@ -58,20 +55,34 @@ public class HdfsServiceTest {
 	@Mock
 	private ExecutorService executorService;
 
-	@Test
-	public void saveMessages() {
-		TopicConfig topicConfig = new TopicConfig();
-		topicConfig.setName("test");
+	@Before
+	public void init() throws NoSuchFieldException, IllegalAccessException { 
+		Db db = TestUtil.newDb("HDFS");
+		db.setHost("host");
+		db.setLogin("login");
+		hdfsService = new HdfsService(db);
+
+		Field configField = HdfsService.class.getDeclaredField("config");
+		configField.setAccessible(true);
+		configField.set(hdfsService, config);
+		
+		hdfsService.init();
+	}
+	
+	@Test(expected = NullPointerException.class)
+	public void saveJsons() { 
+		when(config.getHdfsBufferSize()).thenReturn(1000);	
 
-		List<Pair<Long, String>> messages = new ArrayList<>();
-		messages.add(Pair.of(100L, "test message"));
+		when(config.isAsync()).thenReturn(true);
+		TestUtil.testSaveJsons(config , hdfsService);
 
-		//when(config.getHdfsBufferSize()).thenReturn(1000);
-		//hdfsService.saveMessages(topicConfig, messages);
+		when(config.isAsync()).thenReturn(false);	
+		TestUtil.testSaveJsons(config , hdfsService);
 	}
 
 	@Test(expected = NullPointerException.class)
 	public void cleanUp() {
+		when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
 		hdfsService.flush();
 		hdfsService.flushStall();
 		hdfsService.cleanUp();
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/MongodbServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/MongodbServiceTest.java
index 29d32941..dbcd88dc 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/MongodbServiceTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/db/MongodbServiceTest.java
@@ -20,75 +20,70 @@
 
 package org.onap.datalake.feeder.service.db;
 
-import com.mongodb.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoDatabase;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
 import org.bson.Document;
-import org.json.JSONObject;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
-import org.onap.datalake.feeder.domain.Topic;
-import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.domain.Db;
 import org.onap.datalake.feeder.service.DbService;
-import org.onap.datalake.feeder.service.db.MongodbService;
-
-import static org.mockito.Mockito.when;
+import org.onap.datalake.feeder.util.TestUtil;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import com.mongodb.MongoClient;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
 
 @RunWith(MockitoJUnitRunner.class)
 public class MongodbServiceTest {
 
-    @InjectMocks
-    private MongodbService mongodbService;
-
-    @Mock
-    private ApplicationConfiguration config;
-
-    @Mock
-    private DbService dbService;
-
-    @Mock
-    private MongoDatabase database;
+	private MongodbService mongodbService;
 
-    @Mock
-    private MongoClient mongoClient;
+	@Mock
+	private ApplicationConfiguration config;
 
-    @Mock
-    private Map<String, MongoCollection<Document>> mongoCollectionMap = new HashMap<>();
+	@Mock
+	private DbService dbService;
 
+	@Mock
+	private MongoDatabase database;
 
-    @Test
-    public void cleanUp() {
-	//	when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
-//        mongodbService.cleanUp();
-    }
+	@Mock
+	private MongoClient mongoClient;
 
-    @Test
-    public void saveJsons() {
+	@Mock
+	private Map<String, MongoCollection<Document>> mongoCollectionMap = new HashMap<>();
 
-        Topic topic = new Topic();
-        topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
-        topic.setCorrelateClearedMessage(true);
-        topic.setMessageIdPath("/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem");
-        String jsonString = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailure\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
-        String jsonString2 = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailureCleared\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+	@Before
+	public void init() throws NoSuchFieldException, IllegalAccessException {
+		Db db = TestUtil.newDb("Mongodb");
+		db.setDatabase("database");
+		db.setLogin("login");
+		mongodbService = new MongodbService(db);
 
-        JSONObject jsonObject = new JSONObject(jsonString);
-        JSONObject jsonObject2 = new JSONObject(jsonString2);
+		Field configField = MongodbService.class.getDeclaredField("config");
+		configField.setAccessible(true);
+		configField.set(mongodbService, config);
+ 
+		mongodbService.init();
+	}
 
-        List<JSONObject> jsons = new ArrayList<>();
-        jsons.add(jsonObject);
-        jsons.add(jsonObject2);
+	@Test
+	public void cleanUp() {
+		when(config.getShutdownLock()).thenReturn(new ReentrantReadWriteLock());
+		mongodbService.cleanUp();
+	}
 
-        //mongodbService.saveJsons(topic.getTopicConfig(), jsons);
-    }
+	@Test
+	public void saveJsons() {
+		TestUtil.testSaveJsons(config, mongodbService);
+	}
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/TestUtil.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/TestUtil.java
index bdd25e0e..a54cfd37 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/TestUtil.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/TestUtil.java
@@ -19,17 +19,20 @@
  */
 
 package org.onap.datalake.feeder.util;
+ 
+import java.util.ArrayList;
+import java.util.List;
 
-import org.junit.Test;
+import org.json.JSONObject;
+import org.onap.datalake.feeder.config.ApplicationConfiguration;
 import org.onap.datalake.feeder.domain.Db;
+import org.onap.datalake.feeder.domain.DbType;
+import org.onap.datalake.feeder.domain.EffectiveTopic;
+import org.onap.datalake.feeder.domain.Kafka;
 import org.onap.datalake.feeder.domain.Topic;
 import org.onap.datalake.feeder.domain.TopicName;
-
-import java.io.IOException;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import org.onap.datalake.feeder.service.db.DbStoreService;
+ 
 
 /**
  * test utils
@@ -40,10 +43,17 @@ public class TestUtil {
 
     static int i=0;
 
+    public static Kafka newKafka(String name) {
+    	Kafka kafka  = new Kafka(); 
+    	kafka.setId(name); 
+    	return kafka ;
+    }
+
     public static Db newDb(String name) {
     	Db db = new Db();
     	db.setId(i++);
-    	db.setName(name);    	
+    	db.setName(name);   
+    	db.setDbType(new DbType(name, name));
     	return db;
     }
 
@@ -55,5 +65,24 @@ public class TestUtil {
     	return topic;
     }
 
+	public static void testSaveJsons(ApplicationConfiguration config, DbStoreService dbStoreService) {
+		Topic topic = new Topic();
+		topic.setTopicName(new TopicName("unauthenticated.SEC_FAULT_OUTPUT"));
+		topic.setCorrelateClearedMessage(true);
+		topic.setMessageIdPath("/event/commonEventHeader/eventName,/event/commonEventHeader/reportingEntityName,/event/faultFields/specificProblem");
+		String jsonString = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailure\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+		String jsonString2 = "{\"event\":{\"commonEventHeader\":{\"sourceId\":\"vnf_test_999\",\"startEpochMicrosec\":2222222222222,\"eventId\":\"ab305d54-85b4-a31b-7db2-fb6b9e546016\",\"sequence\":1,\"domain\":\"fautt\",\"lastEpochMicrosec\":1234567890987,\"eventName\":\"Fault_MultiCloud_VMFailureCleared\",\"sourceName\":\"vSBC00\",\"priority\":\"Low\",\"version\":3,\"reportingEntityName\":\"vnf_test_2_rname\"},\"faultFields\":{\"eventSeverity\":\"CRITILLL\",\"alarmCondition\":\"Guest_Os_FaiLLL\",\"faultFieldsVersion\":3,\"specificProblem\":\"Fault_MultiCloud_VMFailure\",\"alarmInterfaceA\":\"aaaa\",\"alarmAdditionalInformation\":[{\"name\":\"objectType3\",\"value\":\"VIN\"},{\"name\":\"objectType4\",\"value\":\"VIN\"}],\"eventSourceType\":\"single\",\"vfStatus\":\"Active\"}}}";
+
+		JSONObject jsonObject = new JSONObject(jsonString);
+		JSONObject jsonObject2 = new JSONObject(jsonString2);
+
+		List<JSONObject> jsons = new ArrayList<>();
+		jsons.add(jsonObject);
+		jsons.add(jsonObject2);
+
+		EffectiveTopic effectiveTopic = new EffectiveTopic(topic, "test");
+
+		dbStoreService.saveJsons(effectiveTopic, jsons);
 
+	}
 }