summaryrefslogtreecommitdiffstats
path: root/components/datalake-handler/feeder
diff options
context:
space:
mode:
Diffstat (limited to 'components/datalake-handler/feeder')
-rw-r--r--components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicConfigPollingService.java13
-rw-r--r--components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicNameService.java59
-rw-r--r--components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/TopicService.java2
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<>();