feat(hw):华为云适配

seatunnel-connectors-v2/connector-elasticsearch-hw/pom.xml

@@ -0,0 +1,105 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You 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">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-elasticsearch-hw</artifactId>
+    <name>SeaTunnel : Connectors V2 : Elasticsearch-hw</name>
+
+    <properties>
+        <elasticsearch-rest-client.version>7.5.1</elasticsearch-rest-client.version>
+        <guava.version>31.1-jre</guava.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-client-huawei</artifactId>
+            <version>6.7.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpasyncclient</artifactId>
+            <version>4.1.4</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-json</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>security</artifactId>
+            <version>206</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <!-- I'm not sure which version of guava to use, so I choose the latest; sure enough, the version inherited from the parent project is not compatible -->
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${guava.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.14</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+    <repositories>
+        <repository>
+            <id>libs-release</id>
+            <url>http://192.168.12.88:8082/artifactory/libs-release-local/</url>
+        </repository>
+        <repository>
+            <id>libs-snapshot</id>
+            <url>http://192.168.12.88:8082/artifactory/libs-snapshot-local/</url>
+        </repository>
+        <repository>
+            <releases>
+                <enabled>true</enabled>
+            </releases>
+            <snapshots>
+                <enabled>true</enabled>
+            </snapshots>
+            <id>huaweicloudsdk</id>
+            <url>https://mirrors.huaweicloud.com/repository/maven/huaweicloudsdk/</url>
+        </repository>
+
+    </repositories>
+</project>

seatunnel-connectors-v2/connector-elasticsearch-hw/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/hw/catalog/ElasticSearchCatalog.java

@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.seatunnel.connectors.seatunnel.elasticsearch.hw.catalog;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import org.apache.seatunnel.api.configuration.util.ConfigUtil;
+import org.apache.seatunnel.api.table.catalog.Catalog;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
+import org.apache.seatunnel.api.table.catalog.exception.DatabaseAlreadyExistException;
+import org.apache.seatunnel.api.table.catalog.exception.DatabaseNotExistException;
+import org.apache.seatunnel.api.table.catalog.exception.TableAlreadyExistException;
+import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.hw.client.EsRestClient;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.hw.dto.ElasticsearchClusterInfo;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.hw.dto.source.IndexDocsCount;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Elasticsearch catalog implementation.
+ *
+ * <p>In ElasticSearch, we use the index as the database and table.
+ */
+public class ElasticSearchCatalog implements Catalog {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchCatalog.class);
+
+    private final String catalogName;
+    private final String defaultDatabase;
+    private final Config pluginConfig;
+
+    private EsRestClient esRestClient;
+
+    // todo: do we need default database?
+    public ElasticSearchCatalog(
+            String catalogName, String defaultDatabase, Config elasticSearchConfig) {
+        this.catalogName = checkNotNull(catalogName, "catalogName cannot be null");
+        this.defaultDatabase = defaultDatabase;
+        this.pluginConfig = checkNotNull(elasticSearchConfig, "elasticSearchConfig cannot be null");
+    }
+
+    @Override
+    public void open() throws CatalogException {
+        try {
+            esRestClient = EsRestClient.createInstance(pluginConfig);
+            ElasticsearchClusterInfo elasticsearchClusterInfo = esRestClient.getClusterInfo();
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug(
+                        "Success open es catalog: {}, cluster info: {}",
+                        catalogName,
+                        elasticsearchClusterInfo);
+            }
+        } catch (Exception e) {
+            throw new CatalogException(String.format("Failed to open catalog %s", catalogName), e);
+        }
+    }
+
+    @Override
+    public void close() throws CatalogException {
+        esRestClient.close();
+    }
+
+    @Override
+    public String getDefaultDatabase() throws CatalogException {
+        return defaultDatabase;
+    }
+
+    @Override
+    public boolean databaseExists(String databaseName) throws CatalogException {
+        // check if the index exist
+        try {
+            List<IndexDocsCount> indexDocsCount = esRestClient.getIndexDocsCount(databaseName);
+            return true;
+        } catch (Exception e) {
+            throw new CatalogException(
+                    String.format(
+                            "Failed to check if catalog %s database %s exists",
+                            catalogName, databaseName),
+                    e);
+        }
+    }
+
+    @Override
+    public List<String> listDatabases() throws CatalogException {
+        return esRestClient.listIndex();
+    }
+
+    @Override
+    public List<String> listTables(String databaseName)
+            throws CatalogException, DatabaseNotExistException {
+        if (!databaseExists(databaseName)) {
+            throw new DatabaseNotExistException(catalogName, databaseName);
+        }
+        return Lists.newArrayList(databaseName);
+    }
+
+    @Override
+    public boolean tableExists(TablePath tablePath) throws CatalogException {
+        checkNotNull(tablePath);
+        // todo: Check if the database name is the same with table name
+        return databaseExists(tablePath.getTableName());
+    }
+
+    @Override
+    public CatalogTable getTable(TablePath tablePath)
+            throws CatalogException, TableNotExistException {
+        // Get the index mapping?
+        checkNotNull(tablePath, "tablePath cannot be null");
+        ElasticSearchDataTypeConvertor elasticSearchDataTypeConvertor =
+                new ElasticSearchDataTypeConvertor();
+        TableSchema.Builder builder = TableSchema.builder();
+        Map<String, String> fieldTypeMapping =
+                esRestClient.getFieldTypeMapping(tablePath.getTableName(), Collections.emptyList());
+        fieldTypeMapping.forEach(
+                (fieldName, fieldType) -> {
+                    // todo: we need to add a new type TEXT or add length in STRING type
+                    PhysicalColumn physicalColumn =
+                            PhysicalColumn.of(
+                                    fieldName,
+                                    elasticSearchDataTypeConvertor.toSeaTunnelType(
+                                            fieldName, fieldType),
+                                    null,
+                                    true,
+                                    null,
+                                    null);
+                    builder.column(physicalColumn);
+                });
+
+        return CatalogTable.of(
+                TableIdentifier.of(
+                        catalogName, tablePath.getDatabaseName(), tablePath.getTableName()),
+                builder.build(),
+                buildTableOptions(tablePath),
+                Collections.emptyList(),
+                "");
+    }
+
+    @Override
+    public void createTable(TablePath tablePath, CatalogTable table, boolean ignoreIfExists)
+            throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
+        // Create the index
+        checkNotNull(tablePath, "tablePath cannot be null");
+        if (tableExists(tablePath)) {
+            if (ignoreIfExists) {
+                return;
+            } else {
+                throw new TableAlreadyExistException(catalogName, tablePath, null);
+            }
+        }
+        esRestClient.createIndex(tablePath.getTableName());
+    }
+
+    @Override
+    public void dropTable(TablePath tablePath, boolean ignoreIfNotExists)
+            throws TableNotExistException, CatalogException {
+        checkNotNull(tablePath);
+        if (!tableExists(tablePath) && !ignoreIfNotExists) {
+            throw new TableNotExistException(catalogName, tablePath);
+        }
+        try {
+            esRestClient.dropIndex(tablePath.getTableName());
+        } catch (Exception ex) {
+            throw new CatalogException(
+                    String.format(
+                            "Failed to drop table %s in catalog %s",
+                            tablePath.getTableName(), catalogName),
+                    ex);
+        }
+    }
+
+    @Override
+    public void createDatabase(TablePath tablePath, boolean ignoreIfExists)
+            throws DatabaseAlreadyExistException, CatalogException {
+        createTable(tablePath, null, ignoreIfExists);
+    }
+
+    @Override
+    public void dropDatabase(TablePath tablePath, boolean ignoreIfNotExists)
+            throws DatabaseNotExistException, CatalogException {
+        dropTable(tablePath, ignoreIfNotExists);
+    }
+
+    private Map<String, String> buildTableOptions(TablePath tablePath) {
+        Map<String, String> options = new HashMap<>();
+        options.put("connector", "elasticsearch");
+        // todo: Right now, we don't use the config in the plugin config, do we need to add
+        // bootstrapt servers here?
+        options.put("config", ConfigUtil.convertToJsonString(tablePath));
+        return options;
+    }
+}

seatunnel-connectors-v2/connector-elasticsearch-hw/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/hw/catalog/ElasticSearchCatalogFactory.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.seatunnel.connectors.seatunnel.elasticsearch.hw.catalog;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.table.catalog.Catalog;
+import org.apache.seatunnel.api.table.factory.CatalogFactory;
+
+public class ElasticSearchCatalogFactory implements CatalogFactory {
+
+    @Override
+    public Catalog createCatalog(String catalogName, ReadonlyConfig options) {
+        // todo:
+        return null;
+    }
+
+    @Override
+    public String factoryIdentifier() {
+        // todo:
+        return "Elasticsearch";
+    }
+
+    @Override
+    public OptionRule optionRule() {
+        // todo:
+        return null;
+    }
+}