Auto populate table 'topic_name'

We continually monitor Kafka; when a new topic is created, its name is
auto inserted into table 'topic_name', which is a look up table and UI's
topic selection list is based on it. 

Issue-ID: DCAEGEN2-1715

Change-Id: I35329eb76ed8fb9d41a73f1827f443f7a10b450e
Signed-off-by: Guobiao Mo <guobiaomo@chinamobile.com>

3 files changed, 73 insertions, 1 deletions

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 a02cd6a2..3bdbcdba 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
@@ -23,6 +23,7 @@ package org.onap.datalake.feeder.service;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -60,6 +61,9 @@ public class TopicConfigPollingService implements Runnable {
 	@Autowired
 	private KafkaRepository kafkaRepository;
 	
+	@Autowired
+	private TopicNameService topicNameService;
+	
 	//effectiveTopic Map, 1st key is kafkaId, 2nd is topic name, the value is a list of EffectiveTopic.
 	private Map<Integer, Map<String, List<EffectiveTopic>>> effectiveTopicMap = new HashMap<>();
 	//private Map<String, TopicConfig> effectiveTopicConfigMap;
@@ -114,7 +118,7 @@ public class TopicConfigPollingService implements Runnable {
 		log.info("TopicConfigPollingService started.");
 
 		while (active) {
-			try { //sleep first since we already pool in init()
+			try { //sleep first since we already called poll() in init()
 				Thread.sleep(config.getCheckTopicInterval());
 				if(!active) {
 					break;
@@ -138,6 +142,7 @@ public class TopicConfigPollingService implements Runnable {
 						log.info("activeTopics list is updated, new={}", newTopics);
 
 						activeTopicMap.put(kafkaId, newTopics);
+						//update version
 						currentActiveTopicsVersionMap.put(kafkaId, currentActiveTopicsVersionMap.getOrDefault(kafkaId, 1)+1);
 					} else {
 						log.debug("activeTopics list is not updated.");
@@ -156,14 +161,20 @@ public class TopicConfigPollingService implements Runnable {
 	}
 
 	private Map<Integer, Set<String>>  poll() throws IOException {
+		Set<String> allTopicNames = new HashSet<>();
+		
 		Map<Integer, Set<String>> ret = new HashMap<>();
 		Iterable<Kafka> kafkas = kafkaRepository.findAll();
 		for (Kafka kafka : kafkas) {
 			if (kafka.isEnabled()) {
 				Set<String> topics = poll(kafka);
 				ret.put(kafka.getId(), topics);
+				allTopicNames.addAll(topics);
 			}
 		}
+		
+		topicNameService.update(allTopicNames);
+		
 		return ret;
 	}
 
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicNameService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicNameService.java
new file mode 100644
index 00000000..021d2c94
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicNameService.java
@@ -0,0 +1,59 @@
+/*
+* ============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 java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.onap.datalake.feeder.domain.TopicName;
+import org.onap.datalake.feeder.repository.TopicNameRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * Service for TopicName
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@Service
+public class TopicNameService {
+	private final Logger log = LoggerFactory.getLogger(this.getClass());
+
+	@Autowired
+	private TopicNameRepository topicNameRepository;
+
+	public void update(Collection<String> allTopicNames) {
+
+		List<TopicName> all = allTopicNames.stream().map(s-> new TopicName(s)).collect(Collectors.toList());
+		List<TopicName> allInDb = (List<TopicName>) topicNameRepository.findAll();
+		
+		Collection<TopicName> additions =  CollectionUtils.subtract(all, allInDb);
+
+		if(!additions.isEmpty())
+			topicNameRepository.saveAll(additions);
+		 
+	}
+}
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicService.java
index 2f0761ab..b6466a85 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicService.java
@@ -101,6 +101,8 @@ public class TopicService {
 		return ret;
 	}
 
+	// for unique topic string, one can create multiple 'topic' in admin UI.
+	// for example, one 'topic' setting correlates events, and sends data to ES, another 'topic' sends data to HDFS without such setting
 	//TODO use query
 	public List<Topic> findTopics(Kafka kafka, String topicStr) {
 		List<Topic> ret = new ArrayList<>();