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>

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 a02cd6a..3bdbcdb 100644 (file)
--- 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 (file)
index 0000000..021d2c9
--- /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 2f0761a..b6466a8 100644 (file)
--- 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<>();