From b3f5051484f5b973a47a60fb8f76a67ca5ff88da Mon Sep 17 00:00:00 2001 From: Guobiao Mo Date: Tue, 25 Jun 2019 17:09:18 -0700 Subject: supports multiple Kafka clusters and DBs Domain classes Issue-ID: DCAEGEN2-1631 Change-Id: I54a715b2d3d8e13f347e46b0faf9d120d9a60548 Signed-off-by: Guobiao Mo --- .../java/org/onap/datalake/feeder/domain/Db.java | 22 +++- .../org/onap/datalake/feeder/domain/DbType.java | 92 +++++++++++++++ .../onap/datalake/feeder/domain/DesignType.java | 18 ++- .../org/onap/datalake/feeder/domain/Kafka.java | 127 +++++++++++++++++++++ .../onap/datalake/feeder/domain/PortalDesign.java | 32 ++++-- .../org/onap/datalake/feeder/domain/Topic.java | 32 ++++-- .../org/onap/datalake/feeder/domain/TopicName.java | 86 ++++++++++++++ .../org/onap/datalake/feeder/dto/TopicConfig.java | 2 +- .../datalake/feeder/enumeration/DbTypeEnum.java | 45 ++++++++ .../datalake/feeder/repository/DbRepository.java | 2 +- .../feeder/repository/DbTypeRepository.java | 35 ++++++ .../feeder/repository/KafkaRepository.java | 35 ++++++ .../feeder/repository/TopicRepository.java | 2 +- .../onap/datalake/feeder/service/DbService.java | 3 +- .../feeder/service/PortalDesignService.java | 4 +- .../onap/datalake/feeder/service/PullService.java | 39 +++++-- .../org/onap/datalake/feeder/service/Puller.java | 15 ++- .../onap/datalake/feeder/service/StoreService.java | 3 +- .../feeder/service/TopicConfigPollingService.java | 3 +- .../onap/datalake/feeder/service/TopicService.java | 6 +- 20 files changed, 552 insertions(+), 51 deletions(-) create mode 100644 components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/DbType.java create mode 100644 components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/Kafka.java create mode 100644 components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/domain/TopicName.java create mode 100644 components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/enumeration/DbTypeEnum.java create mode 100644 components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/DbTypeRepository.java create mode 100644 components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/repository/KafkaRepository.java (limited to 'components/datalake-handler/feeder/src/main/java/org') 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 topics; + private Set topics; public Db() { } @@ -94,6 +105,11 @@ public class Db { this.name = name; } + @Override + public String toString() { + return String.format("Db %s (name=%, enabled=%s)", id, name, enabled); + } + @Override public boolean equals(Object obj) { if (obj == null) 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 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 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 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 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 dbs; + @ManyToMany(fetch = FetchType.EAGER) + @JoinTable(name = "map_kafka_topic", joinColumns = { @JoinColumn(name = "topic_id") }, inverseJoinColumns = { @JoinColumn(name = "kafka_id") }) + protected Set 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 designs; + + + @OneToMany(fetch = FetchType.LAZY, mappedBy = "topicName") + protected Set 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 { +public interface DbRepository extends CrudRepository { 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 { + + +} 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 { + +} 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 { +public interface TopicRepository extends CrudRepository { } 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 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 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 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 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 topics = topicConfigPollingService.getActiveTopics(); + List 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> messages) {//pair=ts+text + public void saveMessages(Kafka kafka, String topicStr, List> 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 getActiveTopics() { + public List 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 ret = topicRepository.findById(topicStr); + Optional 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 relateDb = new HashSet<>(); - topic.setName(tConfig.getName()); + //topic.setName(tConfig.getName()); topic.setLogin(tConfig.getLogin()); topic.setPass(tConfig.getPassword()); topic.setEnabled(tConfig.isEnabled()); -- cgit 1.2.3-korg