Unit tests to improve sonar coverage

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

11 files changed, 499 insertions, 106 deletions

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/controller/FeederController.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/controller/FeederController.java
index 4fc9b7b6..6a44c4f2 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/controller/FeederController.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/controller/FeederController.java
@@ -86,7 +86,7 @@ public class FeederController {
 	@ApiOperation(value="Retrieve feeder status.")
     public String status() {    	
     	String status = "Feeder is running: "+pullService.isRunning();
-    	log.info("senting feeder status ...");//TODO we can send what topics are monitored, how many messages are sent, etc.
+    	log.info("sending feeder status ...");//TODO we can send what topics are monitored, how many messages are sent, etc.
 
         return "{\"version\": \""+config.getDatalakeVersion()+"\", \"running\": "+pullService.isRunning()+"}";
     }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DmaapService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DmaapService.java
index 2274ce99..3be5be6e 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DmaapService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DmaapService.java
@@ -58,11 +58,13 @@ public class DmaapService {
 	@Autowired
 	private TopicService topicService;
 
-	ZooKeeper zk;
+	private ZooKeeper zk;
 
 	@PreDestroy
 	public void cleanUp() throws InterruptedException {
-		zk.close();
+		if (zk != null) {
+			zk.close();
+		}
 	}
 
 	@PostConstruct
@@ -71,6 +73,7 @@ public class DmaapService {
 	}
 
 	//get all topic names from Zookeeper
+	//This method returns empty list if nothing found.
 	public List<String> getTopics() {
 		try {
 			if (zk == null) {
@@ -84,7 +87,7 @@ public class DmaapService {
 			return topics;
 		} catch (Exception e) {
 			zk = null;
-			log.error("Can not get topic list from Zookeeper, for testing, going to use hard coded topic list.", e);
+			log.error("Can not get topic list from Zookeeper, return empty list.", e);
 			return Collections.emptyList();
 		}
 	}
@@ -119,9 +122,6 @@ public class DmaapService {
 	public List<TopicConfig> getActiveTopicConfigs() throws IOException {
 		log.debug("entering getActiveTopicConfigs()...");
 		List<String> allTopics = getTopics();
-		if (allTopics == null) {
-			return Collections.emptyList();
-		}
 
 		List<TopicConfig> ret = new ArrayList<>(allTopics.size());
 		for (String topicStr : allTopics) {
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/StoreService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/StoreService.java
index 126e23b2..03faeb81 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/StoreService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/StoreService.java
@@ -91,8 +91,9 @@ public class StoreService {
 		for (Pair<Long, String> pair : messages) {
 			try {
 				docs.add(messageToJson(topicConfig, pair));
-			} catch (IOException e) {
-				log.error(pair.getRight(), e);
+			} catch (Exception e) {
+				//may see org.json.JSONException.
+				log.error("Error when converting this message to JSON: " + pair.getRight(), e);
 			}
 		}
 
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicConfigPollingService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicConfigPollingService.java
index 80da55fd..58b27834 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicConfigPollingService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicConfigPollingService.java
@@ -102,6 +102,7 @@ public class TopicConfigPollingService implements Runnable {
 				Thread.sleep(config.getDmaapCheckNewTopicInterval());
 			} catch (InterruptedException e) {
 				log.error("Thread.sleep(config.getDmaapCheckNewTopicInterval())", e);
+				Thread.currentThread().interrupt();
 			}
 
 			try {
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/FeederControllerTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/FeederControllerTest.java
index 05295f71..84d7d0a9 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/FeederControllerTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/controller/FeederControllerTest.java
@@ -19,100 +19,58 @@
  */
 package org.onap.datalake.feeder.controller;
 
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.junit.Before;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+
 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.service.DmaapService;
 import org.onap.datalake.feeder.service.PullService;
-import org.onap.datalake.feeder.service.Puller;
-import org.springframework.context.ApplicationContext;
-
-import java.io.IOException;
-import java.lang.reflect.Field;
-
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.when;
-
 
+@RunWith(MockitoJUnitRunner.class)
 public class FeederControllerTest {
+	@Mock
+	private PullService pullService1;
 
-    @InjectMocks
-    private PullService pullService1;
-
-    @Mock
-    private ApplicationConfiguration config;
-
-    @Mock
-    private ApplicationContext context;
+	@Mock
+	private ApplicationConfiguration config;
 
-    @Mock
-    private DmaapService dmaapService1;
+	@InjectMocks
+	private FeederController feederController;
 
-    @Mock
-    private KafkaConsumer<String, String> kafkaConsumer;
+	@Test
+	public void testStart() throws IOException {
+		when(pullService1.isRunning()).thenReturn(true);
+		String start = feederController.start();
+		assertEquals("{\"running\": true}", start);
 
-    @Before
-    public void setupTest() {
-        MockitoAnnotations.initMocks(this);
-    }
+		when(pullService1.isRunning()).thenReturn(false);
+		start = feederController.start();
+		assertEquals("{\"running\": true}", start);
+	}
 
-    private void setAccessPrivateFields(FeederController feederController) throws NoSuchFieldException,
-            IllegalAccessException {
-        Field pullService = feederController.getClass().getDeclaredField("pullService");
-        pullService.setAccessible(true);
-        pullService.set(feederController, pullService1);
-    }
+	@Test
+	public void testStop() {
+		when(pullService1.isRunning()).thenReturn(true);
+		String stop = feederController.stop();
+		assertEquals("{\"running\": false}", stop);
 
-    @Test
-    public void testStart() throws IOException, NoSuchFieldException, IllegalAccessException {
-        FeederController feederController = new FeederController();
-        setAccessPrivateFields(feederController);
-        PullService pullService2 = new PullService();
-        Field applicationConfig = pullService2.getClass().getDeclaredField("config");
-        applicationConfig.setAccessible(true);
-        applicationConfig.set(pullService2, config);
-/*        Field applicationContext = pullService2.getClass().getDeclaredField("context");
-        applicationContext.setAccessible(true);
-        applicationContext.set(pullService2, context);
-        when(config.getKafkaConsumerCount()).thenReturn(1);
-        Puller pullThread = new Puller();
-        Field dmaapService = pullThread.getClass().getDeclaredField("dmaapService");
-        dmaapService.setAccessible(true);
-        dmaapService.set(pullThread, dmaapService1);
-        /*Field kafkaConsumer1 = pullThread.getClass().getDeclaredField("consumer");
-        kafkaConsumer1.setAccessible(true);
-        kafkaConsumer1.set(pullThread, kafkaConsumer);
-        applicationConfig = pullThread.getClass().getDeclaredField("config");
-        applicationConfig.setAccessible(true);
-        applicationConfig.set(pullThread, config);
-        when(context.getBean(Puller.class, 0)).thenReturn(pullThread);
-        ConsumerRecords<String, String> records = ConsumerRecords.empty();
-        when(kafkaConsumer.poll(2)).thenReturn(records);
-        String start = feederController.start();
-        assertEquals("{\"running\": true}", start);*/
-    }
+		when(pullService1.isRunning()).thenReturn(false);
+		stop = feederController.stop();
+		assertEquals("{\"running\": false}", stop);
+	}
 
-    @Test
-    public void testStop() throws NoSuchFieldException, IllegalAccessException {
-        FeederController feederController = new FeederController();
-        setAccessPrivateFields(feederController);
-        String stop = feederController.stop();
-        assertEquals("{\"running\": false}", stop);
-    }
+	@Test
+	public void testStatus() {
+		when(pullService1.isRunning()).thenReturn(false);
+		when(config.getDatalakeVersion()).thenReturn("0.0.1");
 
-    @Test
-    public void testStatus() throws NoSuchFieldException, IllegalAccessException {
-        ApplicationConfiguration conf = new ApplicationConfiguration();
-        conf.setDatalakeVersion("0.0.1");
-        FeederController feederController = new FeederController();
-        feederController.config = conf;
-        setAccessPrivateFields(feederController);
-        String status = feederController.status();
-        assertEquals("{\"version\": \"0.0.1\", \"running\": false}", status);
-    }
+		String status = feederController.status();
+		assertEquals("{\"version\": \"0.0.1\", \"running\": false}", status);
+	}
 }
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 31de53a8..81c37185 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
@@ -20,25 +20,24 @@
 
 package org.onap.datalake.feeder.service;
 
+import static org.junit.Assert.assertNotEquals;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
 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 java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.mockito.Mockito.when;
-import static org.junit.Assert.*;
 
 @RunWith(MockitoJUnitRunner.class)
 public class DmaapServiceTest {
 
-    static String DMAPP_ZOOKEEPER_HOST_PORT = "message-router-zookeeper:2181";
+    static String DMAPP_ZOOKEEPER_HOST_PORT = "test:2181";
 
     @InjectMocks
     private DmaapService dmaapService;
@@ -47,22 +46,23 @@ public class DmaapServiceTest {
     private ApplicationConfiguration config;
     @Mock
     private TopicService topicService;
-
+    
     @Test
-    public void testGetTopics() {
-
+    public void testGetTopics() throws InterruptedException {
         List<String> list = new ArrayList<>();
         list.add("unauthenticated.DCAE_CL_OUTPUT");
         list.add("AAI-EVENT");
         list.add("__consumer_offsets");
         list.add("unauthenticated.SEC_FAULT_OUTPUT");
         list.add("msgrtr.apinode.metrics.dmaap");
+//		when(config.getDmaapKafkaExclude()).thenReturn(new String[] { "AAI-EVENT" });
         when(config.getDmaapZookeeperHostPort()).thenReturn(DMAPP_ZOOKEEPER_HOST_PORT);
         assertNotEquals(list, dmaapService.getTopics());
+    	dmaapService.cleanUp();
     }
 
-    /*@Test
-    public void testGetActiveTopics() throws IOException {
+    @Test
+    public void testGetActiveTopicConfigs() throws IOException {
 
         List<String> list = new ArrayList<>();
         list.add("unauthenticated.DCAE_CL_OUTPUT");
@@ -73,9 +73,9 @@ public class DmaapServiceTest {
 
         when(config.getDmaapZookeeperHostPort()).thenReturn(DMAPP_ZOOKEEPER_HOST_PORT);
         try {
-        	assertNotEquals(list, dmaapService.getActiveTopics());
+        	assertNotEquals(list, dmaapService.getActiveTopicConfigs());
         } catch (Exception e) {
             e.printStackTrace();
         }
-    }*/
+    }
 }
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/HdfsServiceTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/HdfsServiceTest.java
new file mode 100644
index 00000000..23ad794f
--- /dev/null
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/HdfsServiceTest.java
@@ -0,0 +1,78 @@
+/*
+ * ============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.service;
+
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.commons.lang3.tuple.Pair;
+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.springframework.context.ApplicationContext;
+
+/**
+ * Test HdfsService
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class HdfsServiceTest {
+
+	@InjectMocks
+	private HdfsService hdfsService;
+
+	@Mock
+	private ApplicationContext context;
+
+	@Mock
+	private ApplicationConfiguration config;
+
+	@Mock
+	private ExecutorService executorService;
+
+	@Test(expected = NullPointerException.class)
+	public void saveMessages() {
+		TopicConfig topicConfig = new TopicConfig();
+		topicConfig.setName("test");
+
+		List<Pair<Long, String>> messages = new ArrayList<>();
+		messages.add(Pair.of(100L, "test message"));
+
+		when(config.getHdfsBufferSize()).thenReturn(1000);
+		hdfsService.saveMessages(topicConfig, messages);
+	}
+
+	@Test(expected = NullPointerException.class)
+	public void cleanUp() {
+		hdfsService.flush();
+		hdfsService.flushStall();
+		hdfsService.cleanUp();
+	}
+}
\ 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
new file mode 100644
index 00000000..fab5d4cd
--- /dev/null
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/PullerTest.java
@@ -0,0 +1,86 @@
+/*
+ * ============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.service;
+
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+
+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.springframework.context.ApplicationContext;
+
+/**
+ * Test Puller
+ * 
+ * Without a Kafka server, the coverage is low.
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class PullerTest {
+
+	@InjectMocks
+	private Puller puller = new Puller();
+
+	@Mock
+	private ApplicationContext context;
+
+	@Mock
+	private ApplicationConfiguration config;
+
+	@Mock
+	private StoreService storeService;
+
+	@Mock
+	private TopicConfigPollingService topicConfigPollingService;
+
+	public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+		when(config.isAsync()).thenReturn(true);
+
+		Method init = puller.getClass().getDeclaredMethod("init");
+		init.setAccessible(true);
+		init.invoke(puller);
+	}
+
+	@Test
+	public void testRun() throws InterruptedException, IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+		testInit();
+
+		when(config.getDmaapKafkaHostPort()).thenReturn("test:1000");
+		when(config.getDmaapKafkaGroup()).thenReturn("test");
+
+		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
new file mode 100644
index 00000000..44e76328
--- /dev/null
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/StoreServiceTest.java
@@ -0,0 +1,148 @@
+/*
+ * ============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.service;
+
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+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.springframework.context.ApplicationContext;
+
+/**
+ * Test StoreService
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class StoreServiceTest {
+
+	@InjectMocks
+	private StoreService storeService = new StoreService();
+
+	@Mock
+	private ApplicationContext context;
+
+	@Mock
+	private ApplicationConfiguration config;
+
+	@Mock
+	private TopicConfigPollingService configPollingService;
+
+	@Mock
+	private MongodbService mongodbService;
+
+	@Mock
+	private CouchbaseService couchbaseService;
+
+	@Mock
+	private ElasticsearchService elasticsearchService;
+
+	@Mock
+	private HdfsService hdfsService;
+
+	public void testInit() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+		Method init = storeService.getClass().getDeclaredMethod("init");
+		init.setAccessible(true);
+		init.invoke(storeService);
+	}
+
+	private TopicConfig createTopicConfig(String topicStr, String type) {
+
+		TopicConfig topicConfig = new TopicConfig();
+		topicConfig.setName(topicStr);
+		topicConfig.setDataFormat(type);
+		topicConfig.setSaveRaw(true);
+
+		when(configPollingService.getEffectiveTopicConfig(topicStr)).thenReturn(topicConfig);
+
+		return topicConfig;
+	}
+
+	@Test
+	public void saveMessages() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+		testInit();
+
+		TopicConfig topicConfig = createTopicConfig("test1", "JSON");
+
+		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");
+
+		createTopicConfig("test4", "TEXT");
+
+		when(config.getTimestampLabel()).thenReturn("ts");
+		when(config.getRawDataLabel()).thenReturn("raw");
+
+		//JSON
+		List<Pair<Long, String>> messages = new ArrayList<>();
+		messages.add(Pair.of(100L, "{test: 1}"));
+
+		storeService.saveMessages("test1", messages);
+
+		//XML
+		List<Pair<Long, String>> messagesXml = new ArrayList<>();
+		messagesXml.add(Pair.of(100L, "<test></test>")); 
+		messagesXml.add(Pair.of(100L, "<test></test"));//bad xml to trigger exception
+
+		storeService.saveMessages("test2", messagesXml);
+
+		//YAML
+		List<Pair<Long, String>> messagesYaml = new ArrayList<>();
+		messagesYaml.add(Pair.of(100L, "test: yes"));
+
+		storeService.saveMessages("test3", messagesYaml);
+
+		//TEXT
+		List<Pair<Long, String>> messagesText = new ArrayList<>();
+		messagesText.add(Pair.of(100L, "test message"));
+
+		storeService.saveMessages("test4", messagesText);
+
+		//Null mesg
+		storeService.saveMessages("test", null);
+	}
+
+	@Test
+	public void testFlush() {
+		storeService.flush();
+		storeService.flushStall();
+	}
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..a341d2a6
--- /dev/null
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/service/TopicConfigPollingServiceTest.java
@@ -0,0 +1,106 @@
+/*
+ * ============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.service;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.List;
+
+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;
+
+/**
+ * Test TopicConfigPollingService
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class TopicConfigPollingServiceTest {
+	@Mock
+	private ApplicationConfiguration config;
+
+	@Mock
+	private DmaapService dmaapService;
+
+	@InjectMocks
+	private TopicConfigPollingService topicConfigPollingService = new TopicConfigPollingService();
+
+	public void testInit() 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");
+		activeTopicsField.setAccessible(true);
+		activeTopicsField.set(topicConfigPollingService, activeTopics);
+	}
+
+	@Test
+	public void testRun() throws InterruptedException, IllegalAccessException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
+		testInit();
+
+		when(config.getDmaapCheckNewTopicInterval()).thenReturn(1);
+
+		Thread thread = new Thread(topicConfigPollingService);
+		thread.start();
+
+		Thread.sleep(50);
+		topicConfigPollingService.shutdown();
+		thread.join();
+
+		assertTrue(topicConfigPollingService.isActiveTopicsChanged(true));
+	}
+
+	@Test
+	public void testRunNoChange() throws InterruptedException {
+	
+		when(config.getDmaapCheckNewTopicInterval()).thenReturn(1);
+
+		Thread thread = new Thread(topicConfigPollingService);
+		thread.start();
+
+		Thread.sleep(50);
+		topicConfigPollingService.shutdown();
+		thread.join();
+
+		assertFalse(topicConfigPollingService.isActiveTopicsChanged(false));
+	}
+
+	@Test
+	public void testGet() {
+		assertNull(topicConfigPollingService.getEffectiveTopicConfig("test"));
+		assertNull(topicConfigPollingService.getActiveTopics());
+
+	}
+}
\ No newline at end of file
diff --git a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/UtilTest.java b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/UtilTest.java
index 918c0701..753087e4 100644
--- a/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/UtilTest.java
+++ b/components/datalake-handler/feeder/src/test/java/org/onap/datalake/feeder/util/UtilTest.java
@@ -25,6 +25,8 @@ import org.junit.Test;
 import java.io.IOException;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * test utils
@@ -48,4 +50,17 @@ public class UtilTest {
 	public void validateNull() throws IOException {
 		Util.getTextFromFile("no_such_file");
 	}
+
+	@Test
+	//only dot(.) in key got replaced
+	public void isStall() {
+		long lastTime = 10L;
+		long checkInterval = 10000L;
+
+		assertTrue(Util.isStall(lastTime, checkInterval));
+
+		lastTime = System.currentTimeMillis();
+		assertFalse(Util.isStall(lastTime, checkInterval));
+	}
+
 }