supports multiple Kafka clusters and DBs

Domain classes

Issue-ID: DCAEGEN2-1631

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

20 files changed, 552 insertions, 51 deletions

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java
index da1f6cab..d84b34f8 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Db.java
@@ -24,10 +24,13 @@ import java.util.Set;
 import javax.persistence.Column;
 import javax.persistence.Entity;
 import javax.persistence.FetchType;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
 import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.JoinTable;
 import javax.persistence.ManyToMany;
+import javax.persistence.ManyToOne;
 import javax.persistence.Table;
 import com.fasterxml.jackson.annotation.JsonBackReference;
 import lombok.Getter;
@@ -46,6 +49,10 @@ import lombok.Setter;
 @Table(name = "db")
 public class Db {
 	@Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    @Column(name = "`id`")
+    private Integer id;
+
 	@Column(name="`name`")
 	private String name;
 
@@ -79,13 +86,17 @@ public class Db {
 	@Column(name="`property3`")
 	private String property3;
 
+	@ManyToOne(fetch = FetchType.EAGER)
+    @JoinColumn(name = "db_type_id", nullable = false)
+	private DbType dbType;
+	
 	@JsonBackReference
 	@ManyToMany(fetch = FetchType.EAGER)
 	@JoinTable(	name 				= "map_db_topic",
-			joinColumns 		= {  @JoinColumn(name="db_name")  },
-			inverseJoinColumns 	= {  @JoinColumn(name="topic_name")  }
+			joinColumns 		= {  @JoinColumn(name="db_id")  },
+			inverseJoinColumns 	= {  @JoinColumn(name="topic_id")  }
 	)
-	protected Set<Topic> topics;
+	private Set<Topic> topics;
 
 	public Db() {
 	}
@@ -95,6 +106,11 @@ public class Db {
 	}
 
 	@Override
+	public String toString() {
+		return String.format("Db %s (name=%, enabled=%s)", id, name, enabled);
+	}
+
+	@Override
 	public boolean equals(Object obj) {
 		if (obj == null)
 			return false;
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DbType.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DbType.java
new file mode 100644
index 00000000..0a88b155
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DbType.java
@@ -0,0 +1,92 @@
+/*
+* ============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.domain;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.persistence.CascadeType;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.FetchType;
+import javax.persistence.Id;
+import javax.persistence.OneToMany;
+import javax.persistence.Table;
+import lombok.Getter;
+import lombok.Setter;
+
+
+/**
+ * Domain class representing bid data storage type
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@Setter
+@Getter
+@Entity
+@Table(name = "db_type")
+public class DbType {
+	@Id
+	@Column(name="`id`")
+	private String id;
+
+	@Column(name="`name`")
+	private String name;
+
+	@Column(name="`default_port`")
+	private Integer defaultPort;
+
+	@Column(name="`tool`")
+	private Boolean tool;
+ 
+	@OneToMany(cascade = CascadeType.ALL, fetch = FetchType.LAZY, mappedBy = "dbType")
+	protected Set<Db> dbs = new HashSet<>();
+
+	public DbType() {
+	}
+
+	public DbType(String id, String name) {
+		this.id = id;
+		this.name = name;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("DbType %s (name=%s)", id, name);
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (obj == null)
+			return false;
+
+		if (this.getClass() != obj.getClass())
+			return false;
+
+		return id.equals(((DbType) obj).getId());
+	}
+
+	@Override
+	public int hashCode() {
+		return id.hashCode();
+	}
+
+}
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DesignType.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DesignType.java
index 62a7c0c6..83e1666a 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DesignType.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DesignType.java
@@ -26,6 +26,9 @@ import lombok.Getter;
 import lombok.Setter;
 import org.onap.datalake.feeder.dto.DesignTypeConfig;
 
+import java.util.HashSet;
+import java.util.Set;
+
 import javax.persistence.*;
 
 /**
@@ -40,16 +43,25 @@ import javax.persistence.*;
 public class DesignType {
 
     @Id
+    @Column(name = "`id`")
+    private String id;
+    
     @Column(name = "`name`")
     private String name;
 
+    //To be removed
     @ManyToOne(fetch=FetchType.EAGER)
     @JoinColumn(name="portal")
     @JsonBackReference
     private Portal portal;
 
-    @Column(name = "`display`")
-    private String display;
+    @ManyToOne(fetch=FetchType.LAZY)
+    @JoinColumn(name="db_type_id", nullable = false)
+    @JsonBackReference
+    private DbType dbType;    
+
+	@OneToMany(cascade = CascadeType.ALL, fetch = FetchType.LAZY, mappedBy = "designType")
+	protected Set<PortalDesign> designs = new HashSet<>();
 
     @Column(name = "`note`")
     private String note;
@@ -58,7 +70,7 @@ public class DesignType {
 
         DesignTypeConfig designTypeConfig = new DesignTypeConfig();
         designTypeConfig.setDesignType(getName());
-        designTypeConfig.setDisplay(getDisplay());
+        //designTypeConfig.setDisplay(getDisplay());
         return designTypeConfig;
     }
 
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Kafka.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Kafka.java
new file mode 100644
index 00000000..e3347a4a
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Kafka.java
@@ -0,0 +1,127 @@
+/*
+* ============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.domain;
+
+import java.util.Set;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.FetchType;
+import javax.persistence.Id;
+import javax.persistence.JoinColumn;
+import javax.persistence.JoinTable;
+import javax.persistence.ManyToMany;
+import javax.persistence.Table;
+import com.fasterxml.jackson.annotation.JsonBackReference;
+import lombok.Getter;
+import lombok.Setter;
+
+
+/**
+ * Domain class representing Kafka cluster
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@Setter
+@Getter
+@Entity
+@Table(name = "kafka")
+public class Kafka {
+	@Id
+	@Column(name="`id`")
+	private String id;
+	
+	@Column(name="`name`")
+	private String name;
+
+	@Column(name="`enabled`")
+	private boolean	enabled;
+
+	@Column(name="broker_list")
+	private String brokerList;//message-router-kafka:9092,message-router-kafka2:9092
+
+	@Column(name="`zk`")
+	private String zooKeeper;//message-router-zookeeper:2181
+
+	@Column(name="`group`", columnDefinition = "varchar(255) DEFAULT 'datalake'")
+	private String group;
+
+	@Column(name="`secure`", columnDefinition = " bit(1) DEFAULT 0")
+	private Boolean secure;
+	
+	@Column(name="`login`")
+	private String login;
+
+	@Column(name="`pass`")
+	private String pass;
+
+	@Column(name="`security_protocol`")
+	private String securityProtocol;
+
+	//by default, all topics started with '__' are excluded, here one can explicitly include them
+	//example: '__consumer_offsets,__transaction_state'
+	@Column(name="`included_topic`")
+	private String includedTopic;
+	
+	//@Column(name="`excluded_topic`", columnDefinition = "varchar(1023) default '__consumer_offsets,__transaction_state'")
+	@Column(name="`excluded_topic`")
+	private String excludedTopic;
+
+	@Column(name="`consumer_count`", columnDefinition = "integer default 3")
+	private Integer consumerCount;
+	
+	//don't show this field in admin UI 
+	@Column(name="`timeout_sec`", columnDefinition = "integer default 10")
+	private Integer timeout;
+
+	//don't show this field in admin UI 
+	@Column(name="`check_topic_interval_sec`", columnDefinition = "integer default 10")
+	private Integer checkTopicInterval;
+	
+	@JsonBackReference
+	@ManyToMany(fetch = FetchType.EAGER)
+	@JoinTable(	name 				= "map_kafka_topic",
+			joinColumns 		= {  @JoinColumn(name="kafka_id")  },
+			inverseJoinColumns 	= {  @JoinColumn(name="topic_id")  }
+	)
+	private Set<Topic> topics;
+
+	@Override
+	public String toString() {
+		return String.format("Kafka %s (name=%, enabled=%s)", id, name, enabled);
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (obj == null)
+			return false;
+
+		if (this.getClass() != obj.getClass())
+			return false;
+
+		return id.equals(((Kafka) obj).getId());
+	}
+
+	@Override
+	public int hashCode() {
+		return id.hashCode();
+	}
+}
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/PortalDesign.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/PortalDesign.java
index 3a39b4e6..1cbf4e59 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/PortalDesign.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/PortalDesign.java
@@ -24,12 +24,14 @@ import com.fasterxml.jackson.annotation.JsonBackReference;
 import lombok.Getter;
 import lombok.Setter;
 
+import java.util.Set;
+
 import javax.persistence.*;
 
 import org.onap.datalake.feeder.dto.PortalDesignConfig;
 
 /**
- * Domain class representing portal_design
+ * Domain class representing design
  *
  * @author guochunmeng
  */
@@ -37,7 +39,7 @@ import org.onap.datalake.feeder.dto.PortalDesignConfig;
 @Getter
 @Setter
 @Entity
-@Table(name = "portal_design")
+@Table(name = "design")
 public class PortalDesign {
 
     @Id
@@ -48,6 +50,10 @@ public class PortalDesign {
     @Column(name = "`name`")
     private String name;
 
+	@ManyToOne(fetch = FetchType.EAGER)
+    @JoinColumn(name = "topic_name_id", nullable = false)
+	private TopicName topicName;//topic name 
+	
     @Column(name = "`submitted`")
     private Boolean submitted;
 
@@ -58,15 +64,17 @@ public class PortalDesign {
     private String note;
 
     @ManyToOne(fetch=FetchType.EAGER)
-    @JoinColumn(name = "topic")
-    @JsonBackReference
-    private Topic topic;
-
-    @ManyToOne(fetch=FetchType.EAGER)
-    @JoinColumn(name = "type")
+    @JoinColumn(name = "design_type_id", nullable = false)
     @JsonBackReference
     private DesignType designType;
-    
+
+	//@ManyToMany(mappedBy = "topics", cascade=CascadeType.ALL)
+	@JsonBackReference
+	//@JsonManagedReference
+	@ManyToMany(fetch = FetchType.EAGER)
+	@JoinTable(name = "map_db_design", joinColumns = { @JoinColumn(name = "design_id") }, inverseJoinColumns = { @JoinColumn(name = "db_id") })
+	protected Set<Db> dbs;
+
     public PortalDesignConfig getPortalDesignConfig() {
     	
     	PortalDesignConfig portalDesignConfig = new PortalDesignConfig();
@@ -76,9 +84,9 @@ public class PortalDesign {
 		portalDesignConfig.setName(getName());
 		portalDesignConfig.setNote(getNote());
 		portalDesignConfig.setSubmitted(getSubmitted());
-		portalDesignConfig.setTopic(getTopic().getName());
-		portalDesignConfig.setDesignType(getDesignType().getName());
-        portalDesignConfig.setDisplay(getDesignType().getDisplay());
+		portalDesignConfig.setTopic(getTopicName().getId());
+		portalDesignConfig.setDesignType(getDesignType().getId());
+        portalDesignConfig.setDisplay(getDesignType().getName());
 		
 		return portalDesignConfig;
     }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
index c171c569..cb07e140 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Topic.java
@@ -30,6 +30,7 @@ import javax.persistence.Id;
 import javax.persistence.JoinColumn;
 import javax.persistence.JoinTable;
 import javax.persistence.ManyToMany;
+import javax.persistence.ManyToOne;
 import javax.persistence.Table;
 
 import org.onap.datalake.feeder.dto.TopicConfig;
@@ -51,9 +52,13 @@ import lombok.Setter;
 @Table(name = "topic")
 public class Topic {
 	@Id
-	@Column(name = "`name`")
-	private String name;//topic name 
+    @Column(name = "`id`")
+    private Integer id;
 
+	@ManyToOne(fetch = FetchType.EAGER)
+    @JoinColumn(name = "topic_name_id", nullable = false)
+	private TopicName topicName;//topic name 
+	
 	//for protected Kafka topics
 	@Column(name = "`login`")
 	private String login;
@@ -65,9 +70,13 @@ public class Topic {
 	@JsonBackReference
 	//@JsonManagedReference
 	@ManyToMany(fetch = FetchType.EAGER)
-	@JoinTable(name = "map_db_topic", joinColumns = { @JoinColumn(name = "topic_name") }, inverseJoinColumns = { @JoinColumn(name = "db_name") })
+	@JoinTable(name = "map_db_topic", joinColumns = { @JoinColumn(name = "topic_id") }, inverseJoinColumns = { @JoinColumn(name = "db_id") })
 	protected Set<Db> dbs;
 
+	@ManyToMany(fetch = FetchType.EAGER)
+	@JoinTable(name = "map_kafka_topic", joinColumns = { @JoinColumn(name = "topic_id") }, inverseJoinColumns = { @JoinColumn(name = "kafka_id") })
+	protected Set<Kafka> kafkas;
+
 	/**
 	 * indicate if we should monitor this topic
 	 */
@@ -90,6 +99,7 @@ public class Topic {
 	/**
 	 * TTL in day
 	 */
+	@Column(name = "`ttl_day`")
 	private Integer ttl;
 
 	//if this flag is true, need to correlate alarm cleared message to previous alarm 
@@ -112,10 +122,14 @@ public class Topic {
 	public Topic() {
 	}
 
-	public Topic(String name) {
-		this.name = name;
+	public Topic(String name) {//TODO
+		//this.name = name;
 	}
 
+	public String getName() {
+		return topicName.getId();
+	}
+	
 	public boolean isEnabled() {
 		return is(enabled);
 	}
@@ -151,7 +165,7 @@ public class Topic {
 	public TopicConfig getTopicConfig() {
 		TopicConfig tConfig = new TopicConfig();
 
-		tConfig.setName(getName());
+		//tConfig.setName(getName());
 		tConfig.setLogin(getLogin());
 		tConfig.setEnabled(isEnabled());
 		tConfig.setDataFormat(dataFormat);
@@ -181,7 +195,7 @@ public class Topic {
 
 	@Override
 	public String toString() {
-		return name;
+		return String.format("Topic %s (enabled=%s, dbs=%s, kafkas=%s)", topicName, enabled, dbs, kafkas);
 	}
 
 	@Override
@@ -192,12 +206,12 @@ public class Topic {
 		if (this.getClass() != obj.getClass())
 			return false;
 
-		return name.equals(((Topic) obj).getName());
+		return id.equals(((Topic) obj).getId());
 	}
 
 	@Override
 	public int hashCode() {
-		return name.hashCode();
+		return id;
 	}
 
 }
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/TopicName.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/TopicName.java
new file mode 100644
index 00000000..35e6ea54
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/TopicName.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.domain;
+
+
+import java.util.Set;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.FetchType;
+import javax.persistence.Id;
+import javax.persistence.OneToMany;
+import javax.persistence.Table;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Domain class representing unique topic names
+ * 
+ * @author Guobiao Mo
+ *
+ */
+@Setter
+@Getter
+@Entity
+@Table(name = "topic_name")
+public class TopicName {
+	@Id	
+	@Column(name = "`id`")
+	private String id;//topic name 
+
+
+	@OneToMany(fetch = FetchType.LAZY, mappedBy = "topicName")
+	protected Set<PortalDesign> designs;
+	
+
+	@OneToMany(fetch = FetchType.LAZY, mappedBy = "topicName")
+	protected Set<Topic> topics;
+	
+	public TopicName() {
+	}
+
+	public TopicName(String name) {
+		id = name;
+	}
+
+	@Override
+	public String toString() {
+		return "TopicName "+ id;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (obj == null)
+			return false;
+
+		if (this.getClass() != obj.getClass())
+			return false;
+
+		return id.equals(((TopicName) obj).getId());
+	}
+
+	@Override
+	public int hashCode() {
+		return id.hashCode();
+	}
+
+}
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
index 70778bb3..1fffa7ec 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/dto/TopicConfig.java
@@ -129,7 +129,7 @@ public class TopicConfig {
 
 	@Override
 	public String toString() {
-		return String.format("Topic %s(enabled=%s, enabledSinkdbs=%s)", name, enabled, enabledSinkdbs);
+		return String.format("TopicConfig %s(enabled=%s, enabledSinkdbs=%s)", name, enabled, enabledSinkdbs);
 	}
 
 	@Override
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DbTypeEnum.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DbTypeEnum.java
new file mode 100644
index 00000000..9b1eb23b
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DbTypeEnum.java
@@ -0,0 +1,45 @@
+/*
+* ============LICENSE_START=======================================================
+* ONAP : DCAE
+* ================================================================================
+* Copyright 2018 TechMahindra
+*=================================================================================
+* 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.enumeration;
+
+/**
+ * Database type
+ * 
+ * @author Guobiao Mo
+ *
+ */
+public enum DbTypeEnum { 
+	CB("Couchbase"), DRUID("Druid"), ES("Elasticsearch"), HDFS("HDFS"), MONGO("MongoDB"), KIBANA("Kibana");
+
+	private final String name;
+
+	DbTypeEnum(String name) {
+		this.name = name;
+	}
+
+	public static DbTypeEnum fromString(String s) {
+		for (DbTypeEnum df : DbTypeEnum.values()) {
+			if (df.name.equalsIgnoreCase(s)) {
+				return df;
+			}
+		}
+		throw new IllegalArgumentException("Invalid value for db: " + s);
+	}
+}
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbRepository.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbRepository.java
index b09dcdca..a744da6f 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbRepository.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbRepository.java
@@ -31,7 +31,7 @@ import org.springframework.data.repository.CrudRepository;
  *

  */ 

 

-public interface DbRepository extends CrudRepository<Db, String> {

+public interface DbRepository extends CrudRepository<Db, Integer> {

 

     Db findByName(String Name);

 

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbTypeRepository.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbTypeRepository.java
new file mode 100644
index 00000000..b93cb1d1
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbTypeRepository.java
@@ -0,0 +1,35 @@
+/*
+ * ============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.repository;
+
+import org.onap.datalake.feeder.domain.DbType;
+import org.springframework.data.repository.CrudRepository;
+
+/**
+ * DbTypeEnum Repository
+ *
+ * @author Guobiao Mo
+ */
+
+public interface DbTypeRepository extends CrudRepository<DbType, String> {
+    
+    
+}
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/KafkaRepository.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/KafkaRepository.java
new file mode 100644
index 00000000..8e78e5c2
--- /dev/null
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/KafkaRepository.java
@@ -0,0 +1,35 @@
+/*

+* ============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.repository;

+

+import org.onap.datalake.feeder.domain.Kafka;

+import org.springframework.data.repository.CrudRepository;

+

+/**

+ * 

+ * Kafka Repository 

+ * 

+ * @author Guobiao Mo

+ *

+ */ 

+

+public interface KafkaRepository extends CrudRepository<Kafka, String> {

+

+}

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/TopicRepository.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/TopicRepository.java
index 2d9adef8..182bf6f1 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/TopicRepository.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/TopicRepository.java
@@ -31,6 +31,6 @@ import org.springframework.data.repository.CrudRepository;
  *

  */ 

 

-public interface TopicRepository extends CrudRepository<Topic, String> {

+public interface TopicRepository extends CrudRepository<Topic, Integer> {

 

 }

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DbService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DbService.java
index 58bb433a..6d6fb750 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DbService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/DbService.java
@@ -40,8 +40,7 @@ public class DbService {
 	private DbRepository dbRepository;
 
 	public Db getDb(String name) {
-		Optional<Db> ret = dbRepository.findById(name);
-		return ret.isPresent() ? ret.get() : null;
+		return dbRepository.findByName(name);
 	}
 
 	public Db getCouchbase() {
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PortalDesignService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PortalDesignService.java
index 33093ee0..df701e88 100755
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PortalDesignService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PortalDesignService.java
@@ -88,7 +88,7 @@ public class PortalDesignService {
 		if (portalDesignConfig.getTopic() != null) {

 			Topic topic = topicService.getTopic(portalDesignConfig.getTopic());

 			if (topic == null) throw new IllegalArgumentException("topic is null");

-			portalDesign.setTopic(topic);

+			portalDesign.setTopicName(topic.getTopicName());

 		}else {

 			throw new IllegalArgumentException("Can not find topic in DB, topic name: "+portalDesignConfig.getTopic());

 		}

@@ -138,7 +138,7 @@ public class PortalDesignService {
 			//TODO

 			flag = false;

 		} else if (portalDesign.getDesignType() != null && "es_mapping".equals(designTypeName)) {

-			flag = postEsMappingTemplate(portalDesign, portalDesign.getTopic().getName().toLowerCase());

+			flag = postEsMappingTemplate(portalDesign, portalDesign.getTopicName().getId().toLowerCase());

 		} else if (portalDesign.getDesignType() != null && "druid_kafka_spec".equals(designTypeName)) {

 			//TODO

 			flag =false;

diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PullService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PullService.java
index 84d5f337..dc04cf60 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PullService.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/PullService.java
@@ -21,14 +21,19 @@
 package org.onap.datalake.feeder.service;
 
 import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
+import org.onap.datalake.feeder.repository.KafkaRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Service;
 
 /**
@@ -46,9 +51,10 @@ public class PullService {
 	private boolean isRunning = false;
 	private ExecutorService executorService;
 	private Thread topicConfigPollingThread;
+	private Set<Puller> pullers;
 
 	@Autowired
-	private Puller puller;
+	private KafkaRepository kafkaRepository;
 
 	@Autowired
 	private TopicConfigPollingService topicConfigPollingService;
@@ -56,6 +62,9 @@ public class PullService {
 	@Autowired
 	private ApplicationConfiguration config;
 
+	@Autowired
+	private ApplicationContext context;
+
 	/**
 	 * @return the isRunning
 	 */
@@ -73,12 +82,16 @@ public class PullService {
 			return;
 		}
 
-		logger.info("start pulling ...");
-		int numConsumers = config.getKafkaConsumerCount();
-		executorService = Executors.newFixedThreadPool(numConsumers);
+		logger.info("PullService starting ...");
 
-		for (int i = 0; i < numConsumers; i++) {
-			executorService.submit(puller);
+		pullers = new HashSet<>();
+		executorService = Executors.newCachedThreadPool();
+
+		Iterable<Kafka> kafkas = kafkaRepository.findAll();
+		for (Kafka kafka : kafkas) {
+			if (kafka.isEnabled()) {
+				doKafka(kafka);
+			}
 		}
 
 		topicConfigPollingThread = new Thread(topicConfigPollingService);
@@ -90,6 +103,14 @@ public class PullService {
 		Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
 	}
 
+	private void doKafka(Kafka kafka) {
+		Puller puller = context.getBean(Puller.class, kafka);
+		pullers.add(puller);
+		for (int i = 0; i < kafka.getConsumerCount(); i++) {
+			executorService.submit(puller);
+		}
+	}
+
 	/**
 	 * stop pulling
 	 */
@@ -101,7 +122,9 @@ public class PullService {
 		config.getShutdownLock().writeLock().lock();
 		try {
 			logger.info("stop pulling ...");
-			puller.shutdown();
+			for (Puller puller : pullers) {
+				puller.shutdown();
+			}
 
 			logger.info("stop TopicConfigPollingService ...");
 			topicConfigPollingService.shutdown();
@@ -118,7 +141,7 @@ public class PullService {
 		} finally {
 			config.getShutdownLock().writeLock().unlock();
 		}
-		
+
 		isRunning = false;
 	}
 
diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/Puller.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/Puller.java
index e7121ddb..5cc3b55d 100644
--- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/Puller.java
+++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/Puller.java
@@ -40,6 +40,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -71,13 +72,19 @@ public class Puller implements Runnable {
 
 	private boolean active = false;
 	private boolean async;
+	
+	private Kafka kafka;
+
+	public Puller(Kafka kafka) {
+		this.kafka = kafka;
+	}
 
 	@PostConstruct
 	private void init() {
 		async = config.isAsync();
 	}
 
-	private Properties getConsumerConfig() {
+	private Properties getConsumerConfig() {//00
 		Properties consumerConfig = new Properties();
 
 		consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getDmaapKafkaHostPort());
@@ -105,7 +112,7 @@ public class Puller implements Runnable {
 	public void run() {
 		active = true;
 		Properties consumerConfig = getConsumerConfig();
-		log.info("Kafka ConsumerConfig: {}", consumerConfig);
+		log.info("Kafka: {}, ConsumerConfig: {}", kafka, consumerConfig);
 		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerConfig);
 		consumerLocal.set(consumer);
 
@@ -114,7 +121,7 @@ public class Puller implements Runnable {
 		try {
 			while (active) {
 				if (topicConfigPollingService.isActiveTopicsChanged(true)) {//true means update local version as well
-					List<String> topics = topicConfigPollingService.getActiveTopics();
+					List<String> topics = topicConfigPollingService.getActiveTopics(kafka);//00
 					log.info("Active Topic list is changed, subscribe to the latest topics: {}", topics);
 					consumer.subscribe(topics, rebalanceListener);
 				}
@@ -146,7 +153,7 @@ public class Puller implements Runnable {
 					messages.add(Pair.of(record.timestamp(), record.value()));
 					//log.debug("threadid={} topic={}, timestamp={} key={}, offset={}, partition={}, value={}", id, record.topic(), record.timestamp(), record.key(), record.offset(), record.partition(), record.value());
 				}
-				storeService.saveMessages(partition.topic(), messages);
+				storeService.saveMessages(kafka, partition.topic(), messages);//00
 				log.info("saved to topic={} count={}", partition.topic(), partitionRecords.size());//TODO we may record this number to DB
 
 				if (!async) {//for reliability, sync commit offset to Kafka, this slows down a bit
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 2a2f997e..291f1cad 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
@@ -32,6 +32,7 @@ import org.apache.commons.lang3.tuple.Pair;
 import org.json.JSONObject;
 import org.json.XML;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
 import org.onap.datalake.feeder.dto.TopicConfig;
 import org.onap.datalake.feeder.enumeration.DataFormat;
 import org.onap.datalake.feeder.util.JsonUtil;
@@ -81,7 +82,7 @@ public class StoreService {
 		yamlReader = new ObjectMapper(new YAMLFactory());
 	}
 
-	public void saveMessages(String topicStr, List<Pair<Long, String>> messages) {//pair=ts+text
+	public void saveMessages(Kafka kafka, String topicStr, List<Pair<Long, String>> messages) {//pair=ts+text
 		if (CollectionUtils.isEmpty(messages)) {
 			return;
 		}
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 21e1a08f..453b3ee9 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
@@ -30,6 +30,7 @@ import javax.annotation.PostConstruct;
 
 import org.apache.commons.collections.CollectionUtils;
 import org.onap.datalake.feeder.config.ApplicationConfiguration;
+import org.onap.datalake.feeder.domain.Kafka;
 import org.onap.datalake.feeder.dto.TopicConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -84,7 +85,7 @@ public class TopicConfigPollingService implements Runnable {
 		return changed;
 	}
 
-	public List<String> getActiveTopics() {
+	public List<String> getActiveTopics(Kafka kafka) {
 		return activeTopics;
 	}
 
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 64e8b8b1..dd8664e8 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
@@ -84,7 +84,7 @@ public class TopicService {
 	}
 
 	public Topic getTopic(String topicStr) {
-		Optional<Topic> ret = topicRepository.findById(topicStr);
+		Optional<Topic> ret = topicRepository.findById(null);//FIXME
 		return ret.isPresent() ? ret.get() : null;
 	}
 
@@ -96,7 +96,7 @@ public class TopicService {
 		if (topic == null) {
 			return false;
 		}
-		return topic.getName().equals(config.getDefaultTopicName());
+		return true;//topic.getName().equals(config.getDefaultTopicName());
 	}
 
 	public void fillTopicConfiguration(TopicConfig tConfig, Topic wTopic)
@@ -114,7 +114,7 @@ public class TopicService {
 	private void fillTopic(TopicConfig tConfig, Topic topic)
 	{
 		Set<Db> relateDb = new HashSet<>();
-		topic.setName(tConfig.getName());
+		//topic.setName(tConfig.getName());
 		topic.setLogin(tConfig.getLogin());
 		topic.setPass(tConfig.getPassword());
 		topic.setEnabled(tConfig.isEnabled());