You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@linkis.apache.org by ki...@apache.org on 2026/06/11 09:50:51 UTC
(linkis) 07/48: Provide a metadata management common module close #302
This is an automated email from the ASF dual-hosted git repository.
aiceflower pushed a commit to branch release-0.9.4
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 21b5f2223dfe273b884232250c6ea8b84c596df5
Author: xj2jx <xj...@163.com>
AuthorDate: Sun Feb 23 16:04:15 2020 +0800
Provide a metadata management common module
close #302
---
datasource/metadatamanager/common/pom.xml | 74 +++++++
.../linkis/metadatamanager/common/Json.java | 111 +++++++++++
.../metadatamanager/common/MdmConfiguration.java | 28 +++
.../common/cache/CacheConfiguration.java | 28 +++
.../metadatamanager/common/cache/CacheManager.java | 42 ++++
.../common/cache/ConnCacheManager.java | 56 ++++++
.../common/domain/MetaColumnInfo.java | 64 +++++++
.../common/domain/MetaPartitionInfo.java | 93 +++++++++
.../common/exception/MetaRuntimeException.java | 31 +++
.../common/service/AbstractMetaService.java | 213 +++++++++++++++++++++
.../common/service/BaseMetadataService.java | 32 ++++
.../common/service/MetadataConnection.java | 55 ++++++
.../common/service/MetadataDbService.java | 67 +++++++
.../common/service/MetadataService.java | 20 ++
.../common/protocol/MetadataOperateProtocol.scala | 27 +++
.../common/protocol/MetadataProtocol.scala | 20 ++
.../common/protocol/MetadataQueryProtocol.scala | 57 ++++++
.../common/receiver/BaseMetaReceiver.scala | 63 ++++++
18 files changed, 1081 insertions(+)
diff --git a/datasource/metadatamanager/common/pom.xml b/datasource/metadatamanager/common/pom.xml
new file mode 100644
index 0000000000..7f0c5e981e
--- /dev/null
+++ b/datasource/metadatamanager/common/pom.xml
@@ -0,0 +1,74 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ ~ Copyright 2019 WeBank
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>linkis</artifactId>
+ <groupId>com.webank.wedatasphere.linkis</groupId>
+ <version>0.9.3</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>linkis-metadatamanager-common</artifactId>
+
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.webank.wedatasphere.linkis</groupId>
+ <artifactId>linkis-common</artifactId>
+ <version>${linkis.version}</version>
+ <exclusions>
+ <exclusion>
+ <artifactId>asm</artifactId>
+ <groupId>org.ow2.asm</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <!--rpc-->
+ <dependency>
+ <groupId>com.webank.wedatasphere.linkis</groupId>
+ <artifactId>linkis-cloudRPC</artifactId>
+ <version>${linkis.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-deploy-plugin</artifactId>
+ </plugin>
+
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ </plugin>
+ </plugins>
+ <resources>
+ <resource>
+ <directory>src/main/resources</directory>
+ </resource>
+ </resources>
+ <finalName>${project.artifactId}-${project.version}</finalName>
+ </build>
+</project>
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/Json.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/Json.java
new file mode 100644
index 0000000000..203c4a1409
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/Json.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.*;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+
+/**
+ * Json utils
+ * Created by jackyxxie on 2018/9/3.
+ */
+public class Json {
+ private static final String PREFIX = "[";
+ private static final String SUFFIX = "]";
+ private static final Logger logger = LoggerFactory.getLogger(Json.class);
+
+ private static ObjectMapper mapper;
+
+ static{
+ mapper = new ObjectMapper();
+ mapper.configure(JsonParser.Feature.ALLOW_COMMENTS, true);
+ mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);
+ mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
+ mapper.configure(DeserializationFeature.READ_ENUMS_USING_TO_STRING, true);
+ mapper.configure(SerializationFeature.WRITE_ENUMS_USING_TO_STRING, true);
+ mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true);
+ //empty beans allowed
+ mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
+ //ignore unknown properties
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ //cancel to scape non ascii
+ mapper.configure(JsonGenerator.Feature.ESCAPE_NON_ASCII, false);
+ }
+ private Json(){}
+
+ @SuppressWarnings("unchecked")
+ public static <T> T fromJson(String json, Class<?> clazz, Class<?>... parameters){
+ if(StringUtils.isNotBlank(json)){
+ try{
+ if(parameters.length > 0){
+ return (T)mapper.readValue(json, mapper.getTypeFactory().constructParametricType(clazz, parameters));
+ }
+ if(json.startsWith(PREFIX)
+ && json.endsWith(SUFFIX)){
+ JavaType javaType = mapper.getTypeFactory()
+ .constructParametricType(ArrayList.class, clazz);
+ return mapper.readValue(json, javaType);
+ }
+ return (T)mapper.readValue(json, clazz);
+ } catch (Exception e) {
+ logger.info(e.getLocalizedMessage());
+ throw new RuntimeException(e);
+ }
+ }
+ return null;
+ }
+
+ public static <T> T fromJson(InputStream stream, Class<?> clazz, Class<?>... parameters){
+ StringBuilder builder = new StringBuilder();
+ String jsonStr = null;
+ try{
+ BufferedReader reader = new BufferedReader(new InputStreamReader(stream, "UTF-8"));
+ while((jsonStr = reader.readLine()) != null){
+ builder.append(jsonStr);
+ }
+ reader.close();
+ }catch(Exception e){
+ logger.info(e.getLocalizedMessage());
+ throw new RuntimeException(e);
+ }
+ return fromJson(builder.toString(), clazz, parameters);
+ }
+
+ public static String toJson(Object obj, Class<?> model){
+ ObjectWriter writer = mapper.writer();
+ if(null != obj){
+ try{
+ if(null != model){
+ writer = writer.withView(model);
+ }
+ return writer.writeValueAsString(obj);
+ } catch (JsonProcessingException e) {
+ logger.info(e.getLocalizedMessage());
+ throw new RuntimeException(e);
+ }
+ }
+ return null;
+ }
+
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/MdmConfiguration.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/MdmConfiguration.java
new file mode 100644
index 0000000000..480b36d42d
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/MdmConfiguration.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common;
+
+import com.webank.wedatasphere.linkis.common.conf.CommonVars;
+
+/**
+ * Created by jackyxxie on 2020/2/10.
+ */
+public class MdmConfiguration {
+
+ public static CommonVars<String> METADATA_SERVICE_APPLICATION =
+ CommonVars.apply("wds.linkis.server.mdm.service.app.name", "mdm-service");
+
+ public static CommonVars<String> DATA_SOURCE_SERVICE_APPLICATION =
+ CommonVars.apply("wds.linkis.server.dsm.app.name", "dsm-server");
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheConfiguration.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheConfiguration.java
new file mode 100644
index 0000000000..01d083dc66
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheConfiguration.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.cache;
+
+import com.webank.wedatasphere.linkis.common.conf.CommonVars;
+
+/**
+ * Created by jackyxxie on 2020/02/14.
+ */
+public class CacheConfiguration {
+
+ public static CommonVars<Long> CACHE_MAX_SIZE =
+ CommonVars.apply("wds.linkis.server.mdm.service.cache.size", 1000L);
+
+ public static CommonVars<Long> CACHE_EXPIRE_TIME =
+ CommonVars.apply("wds.linkis.server.mdm.service.cache.expire", 600L);
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheManager.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheManager.java
new file mode 100644
index 0000000000..a1c608a931
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheManager.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.cache;
+
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+
+/**
+ * Created by jackyxxie on 2020/2/14.
+ */
+public interface CacheManager {
+ /**
+ * build simple cache
+ * @param cacheId
+ * @param removalListener
+ * @return
+ */
+ <V> Cache<String, V> buildCache(String cacheId, RemovalListener<String, V> removalListener);
+
+ /**
+ * build loading cache
+ * @param cacheId
+ * @param loader
+ * @param removalListener
+ * @return
+ */
+ <V> LoadingCache<String, V> buildCache(String cacheId, CacheLoader<String, V> loader, RemovalListener<String, V> removalListener);
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/ConnCacheManager.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/ConnCacheManager.java
new file mode 100644
index 0000000000..7d96236149
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/ConnCacheManager.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.cache;
+
+import com.google.common.cache.*;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by jackyxxie on 2020/02/14.
+ */
+public class ConnCacheManager implements CacheManager {
+ private ConcurrentHashMap<String, Cache> cacheStore = new ConcurrentHashMap<>();
+ private ConnCacheManager(){
+
+ }
+
+ public static CacheManager custom(){
+ return new ConnCacheManager();
+ }
+
+ @Override
+ public <V> Cache<String, V> buildCache(String cacheId, RemovalListener<String, V> removalListener) {
+ Cache<String, V> vCache = CacheBuilder.newBuilder()
+ .maximumSize(CacheConfiguration.CACHE_MAX_SIZE.getValue())
+ .expireAfterWrite(CacheConfiguration.CACHE_EXPIRE_TIME.getValue(), TimeUnit.SECONDS)
+ .removalListener(removalListener)
+ .build();
+ cacheStore.putIfAbsent(cacheId, vCache);
+ return vCache;
+ }
+
+ @Override
+ public <V> LoadingCache<String, V> buildCache(String cacheId, CacheLoader<String, V> loader,
+ RemovalListener<String, V> removalListener) {
+ LoadingCache<String, V> vCache = CacheBuilder.newBuilder()
+ .maximumSize(CacheConfiguration.CACHE_MAX_SIZE.getValue())
+ .expireAfterWrite(CacheConfiguration.CACHE_EXPIRE_TIME.getValue(), TimeUnit.SECONDS)
+ .removalListener(removalListener)
+ .build(loader);
+ cacheStore.putIfAbsent(cacheId, vCache);
+ return vCache;
+ }
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaColumnInfo.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaColumnInfo.java
new file mode 100644
index 0000000000..60911a8b13
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaColumnInfo.java
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.domain;
+
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import java.io.Serializable;
+
+/**
+ * The meta information of field
+ * Created by jackyxxie on 2020/2/10.
+ */
+@JsonSerialize(include= JsonSerialize.Inclusion.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class MetaColumnInfo implements Serializable {
+ private int index = -1;
+ private boolean isPrimaryKey;
+ private String name;
+ private String type;
+
+ public int getIndex() {
+ return index;
+ }
+
+ public void setIndex(int index) {
+ this.index = index;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ public boolean isPrimaryKey() {
+ return isPrimaryKey;
+ }
+
+ public void setPrimaryKey(boolean primaryKey) {
+ isPrimaryKey = primaryKey;
+ }
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaPartitionInfo.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaPartitionInfo.java
new file mode 100644
index 0000000000..d4286c4c60
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaPartitionInfo.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.domain;
+
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The meta information of partition
+ * Created by jackyxxie on 2020/2/10.
+ */
+@JsonSerialize(include= JsonSerialize.Inclusion.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class MetaPartitionInfo implements Serializable {
+ private List<String> partKeys = new ArrayList<>();
+
+ private String name;
+ /**
+ * Partition tree
+ */
+ private PartitionNode root;
+
+ @JsonSerialize(include= JsonSerialize.Inclusion.NON_EMPTY)
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class PartitionNode{
+ /**
+ * Node name
+ */
+ private String name;
+ /**
+ * Key: partition value
+ * Value: child partition node
+ */
+ private Map<String, PartitionNode> partitions = new HashMap<>();
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public Map<String, PartitionNode> getPartitions() {
+ return partitions;
+ }
+
+ public void setPartitions(Map<String, PartitionNode> partitions) {
+ this.partitions = partitions;
+ }
+ }
+
+ public List<String> getPartKeys() {
+ return partKeys;
+ }
+
+ public void setPartKeys(List<String> partKeys) {
+ this.partKeys = partKeys;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public PartitionNode getRoot() {
+ return root;
+ }
+
+ public void setRoot(PartitionNode root) {
+ this.root = root;
+ }
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/exception/MetaRuntimeException.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/exception/MetaRuntimeException.java
new file mode 100644
index 0000000000..e3dcb0ef36
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/exception/MetaRuntimeException.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.exception;
+
+import com.webank.wedatasphere.linkis.common.exception.WarnException;
+
+/**
+ * Created by jackyxxie on 2020/2/10.
+ */
+public class MetaRuntimeException extends WarnException {
+ private static final int ERROR_CODE = 99900;
+ public MetaRuntimeException(String desc) {
+ super(ERROR_CODE, desc);
+ }
+
+ public MetaRuntimeException(String desc, String ip, int port, String serviceKind) {
+ super(ERROR_CODE, desc, ip, port, serviceKind);
+ }
+
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/AbstractMetaService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/AbstractMetaService.java
new file mode 100644
index 0000000000..744440f32b
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/AbstractMetaService.java
@@ -0,0 +1,213 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.service;
+
+import com.google.common.cache.Cache;
+import com.webank.wedatasphere.linkis.common.exception.WarnException;
+import com.webank.wedatasphere.linkis.metadatamanager.common.Json;
+import com.webank.wedatasphere.linkis.metadatamanager.common.cache.CacheManager;
+import com.webank.wedatasphere.linkis.metadatamanager.common.cache.ConnCacheManager;
+import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaColumnInfo;
+import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaPartitionInfo;
+import com.webank.wedatasphere.linkis.metadatamanager.common.exception.MetaRuntimeException;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.PostConstruct;
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+/**
+ * Created by jackyxxie on 2020/2/10.
+ */
+public abstract class AbstractMetaService<C extends Closeable> implements MetadataService {
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractMetaService.class);
+ private static final String CONN_CACHE_REQ = "_STORED";
+
+
+ private CacheManager connCacheManager;
+
+ /**
+ * Caching connections which built by connect parameters requested with
+ */
+ protected Cache<String, MetadataConnection<C>> reqCache;
+
+ @PostConstruct
+ public void init(){
+ connCacheManager = ConnCacheManager.custom();
+ initCache(connCacheManager);
+ }
+ /**
+ * If want to use cache component, you should invoke this in constructor method
+ * @param cacheManager
+ */
+ protected void initCache(CacheManager cacheManager){
+ String prefix = this.getClass().getSimpleName();
+ reqCache = cacheManager.buildCache(prefix + CONN_CACHE_REQ, notification ->{
+ assert notification.getValue() != null;
+ close(notification.getValue().getConnection());
+ });
+ }
+
+ @Override
+ public abstract MetadataConnection<C> getConnection(String operator, Map<String, Object> params) throws Exception;
+
+ @Override
+ public List<String> getDatabases(String operator, Map<String, Object> params) {
+ return this.getConnAndRun(operator, params, this::queryDatabases);
+ }
+
+ @Override
+ public List<String> getTables(String operator, Map<String, Object> params, String database) {
+ return this.getConnAndRun(operator, params, conn -> this.queryTables(conn, database));
+ }
+
+ @Override
+ public Map<String, String> getTableProps(String operator, Map<String, Object> params, String database, String table) {
+ return this.getConnAndRun(operator, params, conn -> this.queryTableProps(conn, database, table));
+ }
+
+ @Override
+ public MetaPartitionInfo getPartitions(String operator, Map<String, Object> params, String database, String table) {
+ return this.getConnAndRun(operator, params, conn -> this.queryPartitions(conn, database, table));
+ }
+
+ @Override
+ public List<MetaColumnInfo> getColumns(String operator, Map<String, Object> params, String database, String table) {
+ return this.getConnAndRun(operator, params, conn -> this.queryColumns(conn, database, table));
+ }
+
+ /**
+ * Get database list by connection
+ * @param connection metadata connection
+ * @return
+ */
+ public List<String> queryDatabases(C connection){
+ throw new WarnException(-1, "This method is no supported");
+ }
+
+ /**
+ * Get table list by connection and database
+ * @param connection metadata connection
+ * @param database database
+ * @return
+ */
+ public List<String> queryTables(C connection, String database){
+ throw new WarnException(-1, "This method is no supported");
+ }
+
+ /**
+ * Get partitions by connection, database and table
+ * @param connection metadata connection
+ * @param database database
+ * @param table table
+ * @return
+ */
+ public MetaPartitionInfo queryPartitions(C connection, String database, String table){
+ throw new WarnException(-1, "This method is no supported");
+ }
+
+ /**
+ * Get columns by connection, database and table
+ * @param connection metadata connection
+ * @param database database
+ * @param table table
+ * @return
+ */
+ public List<MetaColumnInfo> queryColumns(C connection, String database, String table){
+ throw new WarnException(-1, "This method is no supported");
+ }
+
+ /**
+ * Get table properties
+ * @param connection metadata connection
+ * @param database database
+ * @param table table
+ * @return
+ */
+ public Map<String, String> queryTableProps(C connection, String database, String table){
+ throw new WarnException(-1, "This method is no supported");
+ }
+
+ protected void close(C connection){
+ try {
+ connection.close();
+ } catch (IOException e) {
+ throw new MetaRuntimeException("Fail to close connection[关闭连接失败], [" + e.getMessage() + "]");
+ }
+ }
+
+ protected <R>R getConnAndRun(String operator, Map<String, Object> params, Function<C, R> action) {
+ String cacheKey = "";
+ try{
+ cacheKey = md5String(Json.toJson(params, null), "", 2);
+ MetadataConnection<C> connection;
+ if(null != reqCache) {
+ connection = reqCache.get(cacheKey, () -> getConnection(operator, params));
+ }else{
+ connection = getConnection(operator, params);
+ }
+ return run(connection, action);
+ }catch(Exception e){
+ LOG.error("Error to invoke meta service", e);
+ if(StringUtils.isNotBlank(cacheKey)){
+ reqCache.invalidate(cacheKey);
+ }
+ throw new MetaRuntimeException(e.getMessage());
+ }
+ }
+ private <R>R run(MetadataConnection<C> connection, Function<C, R> action){
+ if(connection.isLock()){
+ connection.getLock().lock();
+ try{
+ return action.apply(connection.getConnection());
+ }finally{
+ connection.getLock().unlock();
+ }
+ }else{
+ return action.apply(connection.getConnection());
+ }
+ }
+
+ private String md5String(String source, String salt, int iterator){
+ StringBuilder token = new StringBuilder();
+ try{
+ MessageDigest digest = MessageDigest.getInstance("md5");
+ if(StringUtils.isNotEmpty(salt)){
+ digest.update(salt.getBytes(StandardCharsets.UTF_8));
+ }
+ byte[] result = digest.digest(source.getBytes());
+ for(int i = 0; i < iterator - 1; i++){
+ digest.reset();
+ result = digest.digest(result);
+ }
+ for (byte aResult : result) {
+ int temp = aResult & 0xFF;
+ if (temp <= 0xF) {
+ token.append("0");
+ }
+ token.append(Integer.toHexString(temp));
+ }
+ }catch(Exception e){
+ throw new RuntimeException(e.getMessage());
+ }
+ return token.toString();
+ }
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/BaseMetadataService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/BaseMetadataService.java
new file mode 100644
index 0000000000..e9146d4059
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/BaseMetadataService.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.service;
+
+
+import java.io.Closeable;
+import java.util.Map;
+
+/**
+ * Created by jackyxxie on 2020/02/10.
+ */
+public interface BaseMetadataService {
+
+ /**
+ * Get connection
+ *
+ * @param params connect params
+ * @return
+ */
+ MetadataConnection<? extends Closeable> getConnection(String operator, Map<String, Object> params) throws Exception;
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataConnection.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataConnection.java
new file mode 100644
index 0000000000..80a2d9d720
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataConnection.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.service;
+
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Connection for metadata
+ * Created by jackyxxie on 2020/2/10.
+ */
+public class MetadataConnection<C> {
+ /***
+ * Avoid complication when different threads calling connection API
+ */
+ private ReentrantLock lock = new ReentrantLock();
+
+ private boolean isLock = true;
+
+ private C connection;
+
+ public MetadataConnection(C connection){
+ this.connection = connection;
+ }
+
+ public MetadataConnection(C connection, boolean isLock){
+ this.connection = connection;
+ this.isLock = isLock;
+ }
+ public ReentrantLock getLock() {
+ return lock;
+ }
+
+ public C getConnection() {
+ return connection;
+ }
+
+ public boolean isLock() {
+ return isLock;
+ }
+
+ public void setLock(boolean lock) {
+ isLock = lock;
+ }
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataDbService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataDbService.java
new file mode 100644
index 0000000000..b4de097080
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataDbService.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.service;
+
+import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaColumnInfo;
+import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaPartitionInfo;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by jackyxxie on 2020/2/10.
+ */
+public interface MetadataDbService extends BaseMetadataService {
+
+ /**
+ * Get all databases
+ * @param params connect params
+ * @return
+ */
+ List<String> getDatabases(String operator, Map<String, Object> params);
+
+ /**
+ * Get all tables from database specified
+ * @param params params
+ * @param database database name
+ * @return
+ */
+ List<String> getTables(String operator, Map<String, Object> params, String database);
+
+ /**
+ * Get table properties from database specified
+ * @param params params
+ * @param database database name
+ * @return
+ */
+ Map<String, String> getTableProps(String operator, Map<String, Object> params, String database, String table);
+ /**
+ * Get all partitions from table specified
+ * @param params params
+ * @param database
+ * @param table
+ * @return
+ */
+ MetaPartitionInfo getPartitions(String operator, Map<String, Object> params, String database, String table);
+
+ /**
+ * Get all field information from table specified
+ * @param params
+ * @param database
+ * @param table
+ * @return
+ */
+ List<MetaColumnInfo> getColumns(String operator, Map<String, Object> params, String database, String table);
+
+}
diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataService.java
new file mode 100644
index 0000000000..3152182514
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataService.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.service;
+
+/**
+ * Created by jackyxxie on 2020/2/10.
+ */
+public interface MetadataService extends MetadataDbService {
+}
diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataOperateProtocol.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataOperateProtocol.scala
new file mode 100644
index 0000000000..834dd268ba
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataOperateProtocol.scala
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.protocol
+
+import java.util
+
+trait MetadataOperateProtocol {
+
+}
+
+/**
+ * Request to do connect
+ * @param version
+ * @param params
+ */
+case class MetadataConnect(operator: String, params: util.Map[String, Object], version: String) extends MetadataOperateProtocol
diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataProtocol.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataProtocol.scala
new file mode 100644
index 0000000000..38940e24a6
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataProtocol.scala
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.protocol
+
+trait MetadataProtocol {
+
+}
+
+case class MetadataResponse(status: Boolean, data: String) extends MetadataProtocol
diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataQueryProtocol.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataQueryProtocol.scala
new file mode 100644
index 0000000000..2668989c90
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataQueryProtocol.scala
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.protocol
+
+import java.util
+
+trait MetadataQueryProtocol {
+
+}
+
+/**
+ * Request to get database list
+ * @param params
+ */
+case class MetaGetDatabases(params: util.Map[String, Object], operator: String) extends MetadataQueryProtocol
+
+/**
+ * Request to get table list
+ * @param params
+ */
+case class MetaGetTables(params: util.Map[String, Object], database: String, operator: String) extends MetadataQueryProtocol
+
+/**
+ * Request to get table properties
+ * @param params
+ * @param database
+ * @param table
+ */
+case class MetaGetTableProps(params: util.Map[String, Object], database: String, table: String, operator: String) extends MetadataQueryProtocol
+
+/**
+ * Request to get partition list
+ * @param params
+ * @param database
+ * @param table
+ */
+case class MetaGetPartitions(params: util.Map[String, Object], database: String, table: String, operator: String) extends MetadataQueryProtocol
+
+/**
+ * Request to get column list
+ * @param params
+ * @param database
+ * @param table
+ */
+case class MetaGetColumns(params: util.Map[String, Object], database: String, table: String, operator: String) extends MetadataQueryProtocol
+
diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/receiver/BaseMetaReceiver.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/receiver/BaseMetaReceiver.scala
new file mode 100644
index 0000000000..d4ecf7d899
--- /dev/null
+++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/receiver/BaseMetaReceiver.scala
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2019 WeBank
+ * 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.
+ */
+
+package com.webank.wedatasphere.linkis.metadatamanager.common.receiver
+
+import com.webank.wedatasphere.linkis.common.exception.WarnException
+import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils}
+import com.webank.wedatasphere.linkis.metadatamanager.common.protocol._
+import com.webank.wedatasphere.linkis.metadatamanager.common.service.MetadataService
+import com.webank.wedatasphere.linkis.rpc.{Receiver, Sender}
+import com.webank.wedatasphere.linkis.server.BDPJettyServerHelper
+
+import scala.concurrent.duration.Duration
+
+/**
+ * Created by jackyxxie on 2020/2/10.
+ */
+class BaseMetaReceiver extends Receiver with Logging{
+ protected var metadataService: MetadataService = _
+
+ override def receive(message: Any, sender: Sender): Unit = ???
+
+ override def receiveAndReply(message: Any, sender: Sender): Any = invoke(metadataService, message)
+
+ override def receiveAndReply(message: Any, duration: Duration, sender: Sender): Any = invoke(metadataService, message)
+
+
+ def invoke(service: MetadataService, message: Any): Any = Utils.tryCatch{
+ val data = message match {
+ case MetaGetDatabases(params, operator) => service.getDatabases(operator, params)
+ case MetaGetTableProps(params, database, table, operator) => service.getTableProps(operator, params, database, table)
+ case MetaGetTables(params, database, operator) => service.getTables(operator, params, database)
+ case MetaGetPartitions(params, database, table, operator) => service.getPartitions(operator, params, database, table)
+ case MetaGetColumns(params, database, table, operator) => service.getColumns(operator, params, database, table)
+ case MetadataConnect(operator, params, version) =>
+ service.getConnection(operator, params)
+ //MetadataConnection is not scala class
+ null
+ case _ => new Object()
+ }
+ MetadataResponse(status = true, BDPJettyServerHelper.gson.toJson(data))
+ }{
+ case e:WarnException => val errorMsg = e.getMessage
+ info(s"Fail to invoke meta service: [$message],[$errorMsg]")
+ MetadataResponse(status = false, errorMsg)
+ case t:Throwable =>
+ val errorMsg = t.getMessage
+ if (message.isInstanceOf[MetadataConnect])
+ info(s"Fail to invoke meta service: [$message], [$errorMsg]")
+ else error(s"Fail to invoke meta service", t)
+ MetadataResponse(status = false, t.getMessage)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org