@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()+"}";
}
@Autowired
private TopicService topicService;
- ZooKeeper zk;
+ private ZooKeeper zk;
@PreDestroy
public void cleanUp() throws InterruptedException {
- zk.close();
+ if (zk != null) {
+ zk.close();
+ }
}
@PostConstruct
}
//get all topic names from Zookeeper
+ //This method returns empty list if nothing found.
public List<String> getTopics() {
try {
if (zk == null) {
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();
}
}
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) {
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);
}
}
Thread.sleep(config.getDmaapCheckNewTopicInterval());
} catch (InterruptedException e) {
log.error("Thread.sleep(config.getDmaapCheckNewTopicInterval())", e);
+ Thread.currentThread().interrupt();
}
try {
*/
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);
+ }
}
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;
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");
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
--- /dev/null
+/*
+ * ============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
--- /dev/null
+/*
+ * ============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
--- /dev/null
+/*
+ * ============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
--- /dev/null
+/*
+ * ============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
import java.io.IOException;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
/**
* test utils
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));
+ }
+
}
Code Review / dcaegen2 / services.git / commitdiff