keys =
- table.getTableSchema().getConstraintKeys().stream()
- .filter(
- k ->
- k.getColumnNames().stream()
- .map(
- ConstraintKey.ConstraintKeyColumn
- ::getColumnName)
- .allMatch(matchingFieldNames::contains))
- .collect(Collectors.toList());
- if (!keys.isEmpty()) {
- builder.constraintKey(keys);
- }
- }
-
table =
CatalogTable.of(
TableIdentifier.of(
diff --git a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/serialize/SeaTunnelRowConverterTest.java b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/serialize/SeaTunnelRowConverterTest.java
deleted file mode 100644
index 5755beb3f74..00000000000
--- a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/serialize/SeaTunnelRowConverterTest.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.doris.serialize;
-
-import org.apache.seatunnel.api.table.type.LocalTimeType;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-import java.time.LocalDateTime;
-
-public class SeaTunnelRowConverterTest {
-
- private static final SeaTunnelRowConverter seaTunnelRowConverter = new SeaTunnelRowConverter();
-
- @Test
- void testDateTimeWithNano() {
- Assertions.assertEquals(
- "2021-01-01 00:00:00.123456",
- seaTunnelRowConverter.convert(
- LocalTimeType.LOCAL_DATE_TIME_TYPE,
- LocalDateTime.of(2021, 1, 1, 0, 0, 0, 123456789)));
- Assertions.assertEquals(
- "2021-01-01 00:00:00.000000",
- seaTunnelRowConverter.convert(
- LocalTimeType.LOCAL_DATE_TIME_TYPE,
- LocalDateTime.of(2021, 1, 1, 0, 0, 0, 0)));
- Assertions.assertEquals(
- "2021-01-01 00:00:00.000001",
- seaTunnelRowConverter.convert(
- LocalTimeType.LOCAL_DATE_TIME_TYPE,
- LocalDateTime.of(2021, 1, 1, 0, 0, 0, 1000)));
- Assertions.assertEquals(
- "2021-01-01 00:00:00.000123",
- seaTunnelRowConverter.convert(
- LocalTimeType.LOCAL_DATE_TIME_TYPE,
- LocalDateTime.of(2021, 1, 1, 0, 0, 0, 123456)));
- }
-}
diff --git a/seatunnel-connectors-v2/connector-easysearch/pom.xml b/seatunnel-connectors-v2/connector-easysearch/pom.xml
deleted file mode 100644
index a121fe21868..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/pom.xml
+++ /dev/null
@@ -1,80 +0,0 @@
-
-
-
- 4.0.0
-
- org.apache.seatunnel
- seatunnel-connectors-v2
- ${revision}
-
-
- connector-easysearch
- SeaTunnel : Connectors V2 : Easysearch
-
-
- 1.0.1
-
-
-
-
- com.infinilabs
- easysearch-client
- ${easysearch-client.version}
-
-
- org.apache.httpcomponents
- httpclient
- 4.5.14
-
-
- org.apache.httpcomponents
- httpasyncclient
- 4.1.4
-
-
- io.airlift
- security
- 206
-
-
- com.google.guava
- guava
-
-
-
-
- org.apache.seatunnel
- connector-common
- ${project.version}
- compile
-
-
- org.apache.seatunnel
- seatunnel-format-json
- ${project.version}
-
-
- com.google.guava
- guava
- ${guava.version}
-
-
-
diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchCatalog.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchCatalog.java
deleted file mode 100644
index 42f13077627..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchCatalog.java
+++ /dev/null
@@ -1,225 +0,0 @@
-/*
- * 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.easysearch.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.easysearch.client.EasysearchClient;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.dto.EasysearchClusterInfo;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.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;
-
-/**
- * Easysearch catalog implementation.
- *
- * In Easysearch, we use the index as the database and table.
- */
-public class EasysearchCatalog implements Catalog {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(EasysearchCatalog.class);
-
- private final String catalogName;
- private final String defaultDatabase;
- private final Config pluginConfig;
-
- private EasysearchClient ezsClient;
-
- // todo: do we need default database?
- public EasysearchCatalog(String catalogName, String defaultDatabase, Config easySearchConfig) {
- this.catalogName = checkNotNull(catalogName, "catalogName cannot be null");
- this.defaultDatabase = defaultDatabase;
- this.pluginConfig = checkNotNull(easySearchConfig, "easySearchConfig cannot be null");
- }
-
- @Override
- public void open() throws CatalogException {
- try {
- ezsClient = EasysearchClient.createInstance(pluginConfig);
- EasysearchClusterInfo easysearchClusterInfo = ezsClient.getClusterInfo();
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "Success open ezs catalog: {}, cluster info: {}",
- catalogName,
- easysearchClusterInfo);
- }
- } catch (Exception e) {
- throw new CatalogException(String.format("Failed to open catalog %s", catalogName), e);
- }
- }
-
- @Override
- public void close() throws CatalogException {
- ezsClient.close();
- }
-
- @Override
- public String name() {
- return catalogName;
- }
-
- @Override
- public String getDefaultDatabase() throws CatalogException {
- return defaultDatabase;
- }
-
- @Override
- public boolean databaseExists(String databaseName) throws CatalogException {
- // check if the index exist
- try {
- List indexDocsCount = ezsClient.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 listDatabases() throws CatalogException {
- return ezsClient.listIndex();
- }
-
- @Override
- public List 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");
- EasysearchDataTypeConvertor easySearchDataTypeConvertor = new EasysearchDataTypeConvertor();
- TableSchema.Builder builder = TableSchema.builder();
- Map fieldTypeMapping =
- ezsClient.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,
- easySearchDataTypeConvertor.toSeaTunnelType(
- fieldName, fieldType),
- (Long) 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);
- }
- }
- ezsClient.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 {
- ezsClient.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 buildTableOptions(TablePath tablePath) {
- Map options = new HashMap<>();
- options.put("connector", "easysearch");
- // todo: Right now, we don't use the config in the plugin config, do we need to add
- // bootstrap servers here?
- options.put("config", ConfigUtil.convertToJsonString(tablePath));
- return options;
- }
-}
diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchCatalogFactory.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchCatalogFactory.java
deleted file mode 100644
index 17ccab123d1..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchCatalogFactory.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.easysearch.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 EasysearchCatalogFactory implements CatalogFactory {
-
- @Override
- public Catalog createCatalog(String catalogName, ReadonlyConfig options) {
- // todo:
- return null;
- }
-
- @Override
- public String factoryIdentifier() {
- // todo:
- return "Easysearch";
- }
-
- @Override
- public OptionRule optionRule() {
- // todo:
- return null;
- }
-}
diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchDataTypeConvertor.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchDataTypeConvertor.java
deleted file mode 100644
index 400fbd1c4d1..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/catalog/EasysearchDataTypeConvertor.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.easysearch.catalog;
-
-import org.apache.seatunnel.api.table.catalog.DataTypeConvertor;
-import org.apache.seatunnel.api.table.type.BasicType;
-import org.apache.seatunnel.api.table.type.LocalTimeType;
-import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
-import org.apache.seatunnel.api.table.type.SqlType;
-
-import com.google.auto.service.AutoService;
-
-import java.util.Map;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-@AutoService(DataTypeConvertor.class)
-public class EasysearchDataTypeConvertor implements DataTypeConvertor {
-
- public static final String STRING = "string";
- public static final String KEYWORD = "keyword";
- public static final String TEXT = "text";
- public static final String BOOLEAN = "boolean";
- public static final String BYTE = "byte";
- public static final String SHORT = "short";
- public static final String INTEGER = "integer";
- public static final String LONG = "long";
- public static final String FLOAT = "float";
- public static final String HALF_FLOAT = "half_float";
- public static final String DOUBLE = "double";
- public static final String DATE = "date";
-
- @Override
- public SeaTunnelDataType toSeaTunnelType(String field, String connectorDataType) {
- return toSeaTunnelType(field, connectorDataType, null);
- }
-
- @Override
- public SeaTunnelDataType toSeaTunnelType(
- String field, String connectorDataType, Map dataTypeProperties) {
- checkNotNull(connectorDataType, "connectorDataType can not be null");
- switch (connectorDataType) {
- case STRING:
- return BasicType.STRING_TYPE;
- case KEYWORD:
- return BasicType.STRING_TYPE;
- case TEXT:
- return BasicType.STRING_TYPE;
- case BOOLEAN:
- return BasicType.BOOLEAN_TYPE;
- case BYTE:
- return BasicType.BYTE_TYPE;
- case SHORT:
- return BasicType.SHORT_TYPE;
- case INTEGER:
- return BasicType.INT_TYPE;
- case LONG:
- return BasicType.LONG_TYPE;
- case FLOAT:
- return BasicType.FLOAT_TYPE;
- case HALF_FLOAT:
- return BasicType.FLOAT_TYPE;
- case DOUBLE:
- return BasicType.DOUBLE_TYPE;
- case DATE:
- return LocalTimeType.LOCAL_DATE_TIME_TYPE;
- default:
- return BasicType.STRING_TYPE;
- }
- }
-
- @Override
- public String toConnectorType(
- String field,
- SeaTunnelDataType seaTunnelDataType,
- Map dataTypeProperties) {
- checkNotNull(seaTunnelDataType, "seaTunnelDataType can not be null");
- SqlType sqlType = seaTunnelDataType.getSqlType();
- switch (sqlType) {
- case STRING:
- return STRING;
- case BOOLEAN:
- return BOOLEAN;
- case BYTES:
- return BYTE;
- case TINYINT:
- return SHORT;
- case INT:
- return INTEGER;
- case BIGINT:
- return LONG;
- case FLOAT:
- return FLOAT;
- case DOUBLE:
- return DOUBLE;
- case TIMESTAMP:
- return DATE;
- default:
- return STRING;
- }
- }
-
- @Override
- public String getIdentity() {
- return "Easysearch";
- }
-}
diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/client/EasysearchClient.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/client/EasysearchClient.java
deleted file mode 100644
index e3cc7f661f2..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/client/EasysearchClient.java
+++ /dev/null
@@ -1,598 +0,0 @@
-/*
- * 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.easysearch.client;
-
-import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
-import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
-import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.TextNode;
-import org.apache.seatunnel.shade.com.typesafe.config.Config;
-
-import org.apache.seatunnel.common.utils.JsonUtils;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.config.EzsClusterConnectionConfig;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.dto.BulkResponse;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.dto.EasysearchClusterInfo;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.dto.source.IndexDocsCount;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.dto.source.ScrollResult;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.exception.EasysearchConnectorErrorCode;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.exception.EasysearchConnectorException;
-import org.apache.seatunnel.connectors.seatunnel.easysearch.util.SSLUtils;
-
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.http.HttpHost;
-import org.apache.http.HttpStatus;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.conn.ssl.NoopHostnameVerifier;
-import org.apache.http.conn.ssl.TrustAllStrategy;
-import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.ssl.SSLContexts;
-import org.apache.http.util.Asserts;
-import org.apache.http.util.EntityUtils;
-
-import org.easysearch.client.Request;
-import org.easysearch.client.Response;
-import org.easysearch.client.RestClient;
-import org.easysearch.client.RestClientBuilder;
-
-import lombok.extern.slf4j.Slf4j;
-
-import javax.net.ssl.SSLContext;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-@Slf4j
-public class EasysearchClient {
-
- private static final int CONNECTION_REQUEST_TIMEOUT = 10 * 1000;
-
- private static final int SOCKET_TIMEOUT = 5 * 60 * 1000;
-
- private final RestClient restClient;
-
- private EasysearchClient(RestClient restClient) {
- this.restClient = restClient;
- }
-
- public static EasysearchClient createInstance(Config pluginConfig) {
- List hosts = pluginConfig.getStringList(EzsClusterConnectionConfig.HOSTS.key());
- Optional username = Optional.empty();
- Optional password = Optional.empty();
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.USERNAME.key())) {
- username =
- Optional.of(pluginConfig.getString(EzsClusterConnectionConfig.USERNAME.key()));
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.PASSWORD.key())) {
- password =
- Optional.of(
- pluginConfig.getString(EzsClusterConnectionConfig.PASSWORD.key()));
- }
- }
- Optional keystorePath = Optional.empty();
- Optional keystorePassword = Optional.empty();
- Optional truststorePath = Optional.empty();
- Optional truststorePassword = Optional.empty();
- boolean tlsVerifyCertificate =
- EzsClusterConnectionConfig.TLS_VERIFY_CERTIFICATE.defaultValue();
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.TLS_VERIFY_CERTIFICATE.key())) {
- tlsVerifyCertificate =
- pluginConfig.getBoolean(
- EzsClusterConnectionConfig.TLS_VERIFY_CERTIFICATE.key());
- }
- if (tlsVerifyCertificate) {
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.TLS_KEY_STORE_PATH.key())) {
- keystorePath =
- Optional.of(
- pluginConfig.getString(
- EzsClusterConnectionConfig.TLS_KEY_STORE_PATH.key()));
- }
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.TLS_KEY_STORE_PASSWORD.key())) {
- keystorePassword =
- Optional.of(
- pluginConfig.getString(
- EzsClusterConnectionConfig.TLS_KEY_STORE_PASSWORD.key()));
- }
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.TLS_TRUST_STORE_PATH.key())) {
- truststorePath =
- Optional.of(
- pluginConfig.getString(
- EzsClusterConnectionConfig.TLS_TRUST_STORE_PATH.key()));
- }
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.TLS_TRUST_STORE_PASSWORD.key())) {
- truststorePassword =
- Optional.of(
- pluginConfig.getString(
- EzsClusterConnectionConfig.TLS_TRUST_STORE_PASSWORD.key()));
- }
- }
- boolean tlsVerifyHostnames = EzsClusterConnectionConfig.TLS_VERIFY_HOSTNAME.defaultValue();
- if (pluginConfig.hasPath(EzsClusterConnectionConfig.TLS_VERIFY_HOSTNAME.key())) {
- tlsVerifyHostnames =
- pluginConfig.getBoolean(EzsClusterConnectionConfig.TLS_VERIFY_HOSTNAME.key());
- }
- return createInstance(
- hosts,
- username,
- password,
- tlsVerifyCertificate,
- tlsVerifyHostnames,
- keystorePath,
- keystorePassword,
- truststorePath,
- truststorePassword);
- }
-
- public static EasysearchClient createInstance(
- List hosts,
- Optional username,
- Optional password,
- boolean tlsVerifyCertificate,
- boolean tlsVerifyHostnames,
- Optional keystorePath,
- Optional keystorePassword,
- Optional truststorePath,
- Optional truststorePassword) {
- RestClientBuilder restClientBuilder =
- getRestClientBuilder(
- hosts,
- username,
- password,
- tlsVerifyCertificate,
- tlsVerifyHostnames,
- keystorePath,
- keystorePassword,
- truststorePath,
- truststorePassword);
- return new EasysearchClient(restClientBuilder.build());
- }
-
- private static RestClientBuilder getRestClientBuilder(
- List hosts,
- Optional username,
- Optional password,
- boolean tlsVerifyCertificate,
- boolean tlsVerifyHostnames,
- Optional keystorePath,
- Optional keystorePassword,
- Optional truststorePath,
- Optional truststorePassword) {
- HttpHost[] httpHosts = new HttpHost[hosts.size()];
- for (int i = 0; i < hosts.size(); i++) {
- httpHosts[i] = HttpHost.create(hosts.get(i));
- }
-
- RestClientBuilder restClientBuilder =
- RestClient.builder(httpHosts)
- .setRequestConfigCallback(
- requestConfigBuilder ->
- requestConfigBuilder
- .setConnectionRequestTimeout(
- CONNECTION_REQUEST_TIMEOUT)
- .setSocketTimeout(SOCKET_TIMEOUT));
-
- restClientBuilder.setHttpClientConfigCallback(
- httpClientBuilder -> {
- if (username.isPresent()) {
- CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
- credentialsProvider.setCredentials(
- AuthScope.ANY,
- new UsernamePasswordCredentials(username.get(), password.get()));
- httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
- }
-
- try {
- if (tlsVerifyCertificate) {
- Optional sslContext =
- SSLUtils.buildSSLContext(
- keystorePath,
- keystorePassword,
- truststorePath,
- truststorePassword);
- sslContext.ifPresent(e -> httpClientBuilder.setSSLContext(e));
- } else {
- SSLContext sslContext =
- SSLContexts.custom()
- .loadTrustMaterial(new TrustAllStrategy())
- .build();
- httpClientBuilder.setSSLContext(sslContext);
- }
- if (!tlsVerifyHostnames) {
- httpClientBuilder.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- return httpClientBuilder;
- });
- return restClientBuilder;
- }
-
- private static Map getFieldTypeMappingFromProperties(
- JsonNode properties, List source) {
- Map allEasysearchFieldTypeInfoMap = new HashMap<>();
- properties
- .fields()
- .forEachRemaining(
- entry -> {
- String fieldName = entry.getKey();
- JsonNode fieldProperty = entry.getValue();
- if (fieldProperty.has("type")) {
- allEasysearchFieldTypeInfoMap.put(
- fieldName, fieldProperty.get("type").asText());
- }
- });
- if (CollectionUtils.isEmpty(source)) {
- return allEasysearchFieldTypeInfoMap;
- }
-
- return source.stream()
- .collect(
- Collectors.toMap(
- Function.identity(),
- fieldName -> {
- String fieldType = allEasysearchFieldTypeInfoMap.get(fieldName);
- if (fieldType == null) {
- log.warn(
- "fail to get easysearch field {} mapping type,so give a default type text",
- fieldName);
- return "text";
- }
- return fieldType;
- }));
- }
-
- public BulkResponse bulk(String requestBody) {
- Request request = new Request("POST", "/_bulk");
- request.setJsonEntity(requestBody);
- try {
- Response response = restClient.performRequest(request);
- if (response == null) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.BULK_RESPONSE_ERROR,
- "bulk ezs Response is null");
- }
- if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- ObjectMapper objectMapper = new ObjectMapper();
- String entity = EntityUtils.toString(response.getEntity());
- JsonNode json = objectMapper.readTree(entity);
- int took = json.get("took").asInt();
- boolean errors = json.get("errors").asBoolean();
- return new BulkResponse(errors, took, entity);
- } else {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.BULK_RESPONSE_ERROR,
- String.format(
- "bulk ezs response status code=%d,request boy=%s",
- response.getStatusLine().getStatusCode(), requestBody));
- }
- } catch (IOException e) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.BULK_RESPONSE_ERROR,
- String.format("bulk ezs error,request boy=%s", requestBody),
- e);
- }
- }
-
- public EasysearchClusterInfo getClusterInfo() {
- Request request = new Request("GET", "/");
- try {
- Response response = restClient.performRequest(request);
- String result = EntityUtils.toString(response.getEntity());
- ObjectMapper objectMapper = new ObjectMapper();
- JsonNode jsonNode = objectMapper.readTree(result);
- JsonNode versionNode = jsonNode.get("version");
- return EasysearchClusterInfo.builder()
- .clusterVersion(versionNode.get("number").asText())
- .distribution(
- Optional.ofNullable(versionNode.get("distribution"))
- .map(e -> e.asText())
- .orElse(null))
- .build();
- } catch (IOException e) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.GET_EZS_VERSION_FAILED,
- "fail to get easysearch version.",
- e);
- }
- }
-
- public void close() {
- try {
- restClient.close();
- } catch (IOException e) {
- log.warn("close easysearch connection error", e);
- }
- }
-
- /**
- * first time to request search documents by scroll call /${index}/_search?scroll=${scroll}
- *
- * @param index index name
- * @param source select fields
- * @param scrollTime such as:1m
- * @param scrollSize fetch documents count in one request
- */
- public ScrollResult searchByScroll(
- String index,
- List source,
- Map query,
- String scrollTime,
- int scrollSize) {
- Map param = new HashMap<>();
- param.put("query", query);
- param.put("_source", source);
- param.put("sort", new String[] {"_doc"});
- param.put("size", scrollSize);
- String endpoint = "/" + index + "/_search?scroll=" + scrollTime;
- ScrollResult scrollResult =
- getDocsFromScrollRequest(endpoint, JsonUtils.toJsonString(param));
- return scrollResult;
- }
-
- /**
- * scroll to get result call _search/scroll
- *
- * @param scrollId the scroll id of the last request
- * @param scrollTime such as:1m
- */
- public ScrollResult searchWithScrollId(String scrollId, String scrollTime) {
- Map param = new HashMap<>();
- param.put("scroll_id", scrollId);
- param.put("scroll", scrollTime);
- ScrollResult scrollResult =
- getDocsFromScrollRequest("/_search/scroll", JsonUtils.toJsonString(param));
- return scrollResult;
- }
-
- private ScrollResult getDocsFromScrollRequest(String endpoint, String requestBody) {
- Request request = new Request("POST", endpoint);
- request.setJsonEntity(requestBody);
- try {
- Response response = restClient.performRequest(request);
- if (response == null) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.SCROLL_REQUEST_ERROR,
- "POST " + endpoint + " response null");
- }
- if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity());
- ObjectNode responseJson = JsonUtils.parseObject(entity);
-
- JsonNode shards = responseJson.get("_shards");
- int totalShards = shards.get("total").intValue();
- int successful = shards.get("successful").intValue();
- Asserts.check(
- totalShards == successful,
- String.format(
- "POST %s,total shards(%d)!= successful shards(%d)",
- endpoint, totalShards, successful));
-
- ScrollResult scrollResult = getDocsFromScrollResponse(responseJson);
- return scrollResult;
- } else {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.SCROLL_REQUEST_ERROR,
- String.format(
- "POST %s response status code=%d,request boy=%s",
- endpoint, response.getStatusLine().getStatusCode(), requestBody));
- }
- } catch (IOException e) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.SCROLL_REQUEST_ERROR,
- String.format("POST %s error,request boy=%s", endpoint, requestBody),
- e);
- }
- }
-
- private ScrollResult getDocsFromScrollResponse(ObjectNode responseJson) {
- ScrollResult scrollResult = new ScrollResult();
- String scrollId = responseJson.get("_scroll_id").asText();
- scrollResult.setScrollId(scrollId);
-
- JsonNode hitsNode = responseJson.get("hits").get("hits");
- List> docs = new ArrayList<>(hitsNode.size());
- scrollResult.setDocs(docs);
-
- Iterator iter = hitsNode.iterator();
- while (iter.hasNext()) {
- Map doc = new HashMap<>();
- JsonNode hitNode = iter.next();
- doc.put("_index", hitNode.get("_index").textValue());
- doc.put("_id", hitNode.get("_id").textValue());
- JsonNode source = hitNode.get("_source");
- for (Iterator> iterator = source.fields();
- iterator.hasNext(); ) {
- Map.Entry entry = iterator.next();
- String fieldName = entry.getKey();
- if (entry.getValue() instanceof TextNode) {
- doc.put(fieldName, entry.getValue().textValue());
- } else {
- doc.put(fieldName, entry.getValue());
- }
- }
- docs.add(doc);
- }
- return scrollResult;
- }
-
- public List getIndexDocsCount(String index) {
- String endpoint = String.format("/_cat/indices/%s?h=index,docsCount&format=json", index);
- Request request = new Request("GET", endpoint);
- try {
- Response response = restClient.performRequest(request);
- if (response == null) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED,
- "GET " + endpoint + " response null");
- }
- if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity());
- List indexDocsCounts =
- JsonUtils.toList(entity, IndexDocsCount.class);
- return indexDocsCounts;
- } else {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED,
- String.format(
- "GET %s response status code=%d",
- endpoint, response.getStatusLine().getStatusCode()));
- }
- } catch (IOException ex) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED, ex);
- }
- }
-
- public List listIndex() {
- String endpoint = "/_cat/indices?format=json";
- Request request = new Request("GET", endpoint);
- try {
- Response response = restClient.performRequest(request);
- if (response == null) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.LIST_INDEX_FAILED,
- "GET " + endpoint + " response null");
- }
- if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity());
- return JsonUtils.toList(entity, Map.class).stream()
- .map(map -> map.get("index").toString())
- .collect(Collectors.toList());
- } else {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.LIST_INDEX_FAILED,
- String.format(
- "GET %s response status code=%d",
- endpoint, response.getStatusLine().getStatusCode()));
- }
- } catch (IOException ex) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.LIST_INDEX_FAILED, ex);
- }
- }
-
- // todo: We don't support set the index mapping now.
- public void createIndex(String indexName) {
- String endpoint = String.format("/%s", indexName);
- Request request = new Request("PUT", endpoint);
- try {
- Response response = restClient.performRequest(request);
- if (response == null) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.CREATE_INDEX_FAILED,
- "PUT " + endpoint + " response null");
- }
- if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.CREATE_INDEX_FAILED,
- String.format(
- "PUT %s response status code=%d",
- endpoint, response.getStatusLine().getStatusCode()));
- }
- } catch (IOException ex) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.CREATE_INDEX_FAILED, ex);
- }
- }
-
- public void dropIndex(String tableName) {
- String endpoint = String.format("/%s", tableName);
- Request request = new Request("DELETE", endpoint);
- try {
- Response response = restClient.performRequest(request);
- if (response == null) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.DROP_INDEX_FAILED,
- "DELETE " + endpoint + " response null");
- }
- // todo: if the index doesn't exist, the response status code is 200?
- if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- return;
- } else {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.DROP_INDEX_FAILED,
- String.format(
- "DELETE %s response status code=%d",
- endpoint, response.getStatusLine().getStatusCode()));
- }
- } catch (IOException ex) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.DROP_INDEX_FAILED, ex);
- }
- }
-
- /**
- * get ezs field name and type mapping realtion
- *
- * @param index index name
- * @return {key-> field name,value->ezs type}
- */
- public Map getFieldTypeMapping(String index, List source) {
- String endpoint = String.format("/%s/_mappings", index);
- Request request = new Request("GET", endpoint);
- Map mapping = new HashMap<>();
- try {
- Response response = restClient.performRequest(request);
- if (response == null) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED,
- "GET " + endpoint + " response null");
- }
- if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED,
- String.format(
- "GET %s response status code=%d",
- endpoint, response.getStatusLine().getStatusCode()));
- }
- String entity = EntityUtils.toString(response.getEntity());
- log.info(String.format("GET %s respnse=%s", endpoint, entity));
- ObjectNode responseJson = JsonUtils.parseObject(entity);
- for (Iterator it = responseJson.elements(); it.hasNext(); ) {
- JsonNode indexProperty = it.next();
- JsonNode mappingsProperty = indexProperty.get("mappings");
- if (mappingsProperty.has("mappingsProperty")) {
- JsonNode properties = mappingsProperty.get("properties");
- mapping = getFieldTypeMappingFromProperties(properties, source);
- } else {
- for (JsonNode typeNode : mappingsProperty) {
- JsonNode properties;
- if (typeNode.has("properties")) {
- properties = typeNode.get("properties");
- } else {
- properties = typeNode;
- }
- mapping.putAll(getFieldTypeMappingFromProperties(properties, source));
- }
- }
- }
- } catch (IOException ex) {
- throw new EasysearchConnectorException(
- EasysearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED, ex);
- }
- return mapping;
- }
-}
diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/EzsClusterConnectionConfig.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/EzsClusterConnectionConfig.java
deleted file mode 100644
index 500da50d2fe..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/EzsClusterConnectionConfig.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.easysearch.config;
-
-import org.apache.seatunnel.api.configuration.Option;
-import org.apache.seatunnel.api.configuration.Options;
-
-import java.util.List;
-
-public class EzsClusterConnectionConfig {
-
- public static final Option> HOSTS =
- Options.key("hosts")
- .listType()
- .noDefaultValue()
- .withDescription(
- "Easysearch cluster http address, the format is host:port, allowing multiple hosts to be specified. Such as [\"host1:9200\", \"host2:9200\"]");
-
- public static final Option USERNAME =
- Options.key("username")
- .stringType()
- .noDefaultValue()
- .withDescription("security username");
-
- public static final Option PASSWORD =
- Options.key("password")
- .stringType()
- .noDefaultValue()
- .withDescription("security password");
-
- public static final Option TLS_VERIFY_CERTIFICATE =
- Options.key("tls_verify_certificate")
- .booleanType()
- .defaultValue(true)
- .withDescription("Enable certificates validation for HTTPS endpoints");
-
- public static final Option TLS_VERIFY_HOSTNAME =
- Options.key("tls_verify_hostname")
- .booleanType()
- .defaultValue(true)
- .withDescription("Enable hostname validation for HTTPS endpoints");
-
- public static final Option TLS_KEY_STORE_PATH =
- Options.key("tls_keystore_path")
- .stringType()
- .noDefaultValue()
- .withDescription(
- "The path to the PEM or JKS key store. This file must be readable by the operating system user running SeaTunnel.");
-
- public static final Option TLS_KEY_STORE_PASSWORD =
- Options.key("tls_keystore_password")
- .stringType()
- .noDefaultValue()
- .withDescription("The key password for the key store specified");
-
- public static final Option TLS_TRUST_STORE_PATH =
- Options.key("tls_truststore_path")
- .stringType()
- .noDefaultValue()
- .withDescription(
- "The path to PEM or JKS trust store. This file must be readable by the operating system user running SeaTunnel.");
-
- public static final Option TLS_TRUST_STORE_PASSWORD =
- Options.key("tls_truststore_password")
- .stringType()
- .noDefaultValue()
- .withDescription("The key password for the trust store specified");
-}
diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/SinkConfig.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/SinkConfig.java
deleted file mode 100644
index 2b07b3c342d..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/SinkConfig.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.easysearch.config;
-
-import org.apache.seatunnel.api.configuration.Option;
-import org.apache.seatunnel.api.configuration.Options;
-
-import java.util.List;
-
-public class SinkConfig {
-
- public static final Option INDEX =
- Options.key("index")
- .stringType()
- .noDefaultValue()
- .withDescription(
- "Easysearch index name.Index support contains variables of field name,such as seatunnel_${age},and the field must appear at seatunnel row. If not, we will treat it as a normal index");
- public static final Option> PRIMARY_KEYS =
- Options.key("primary_keys")
- .listType(String.class)
- .noDefaultValue()
- .withDescription("Primary key fields used to generate the document `_id`");
-
- public static final Option KEY_DELIMITER =
- Options.key("key_delimiter")
- .stringType()
- .defaultValue("_")
- .withDescription(
- "Delimiter for composite keys (\"_\" by default), e.g., \"$\" would result in document `_id` \"KEY1$KEY2$KEY3\".");
-
- public static final Option MAX_BATCH_SIZE =
- Options.key("max_batch_size")
- .intType()
- .defaultValue(10)
- .withDescription("batch bulk doc max size");
-
- public static final Option MAX_RETRY_COUNT =
- Options.key("max_retry_count")
- .intType()
- .defaultValue(3)
- .withDescription("one bulk request max try count");
-}
diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/SourceConfig.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/SourceConfig.java
deleted file mode 100644
index f80c4aa9293..00000000000
--- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/SourceConfig.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.easysearch.config;
-
-import org.apache.seatunnel.api.configuration.Option;
-import org.apache.seatunnel.api.configuration.Options;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class SourceConfig {
-
- public static final Option INDEX =
- Options.key("index")
- .stringType()
- .noDefaultValue()
- .withDescription("Easysearch index name, support * fuzzy matching");
-
- public static final Option> SOURCE =
- Options.key("source")
- .listType()
- .noDefaultValue()
- .withDescription(
- "The fields of index. You can get the document id by specifying the field _id.If sink _id to other index,you need specify an alias for _id due to the Easysearch limit");
-
- public static final Option SCROLL_TIME =
- Options.key("scroll_time")
- .stringType()
- .defaultValue("1m")
- .withDescription(
- "Amount of time Easysearch will keep the search context alive for scroll requests");
-
- public static final Option SCROLL_SIZE =
- Options.key("scroll_size")
- .intType()
- .defaultValue(100)
- .withDescription(
- "Maximum number of hits to be returned with each Easysearch scroll request");
-
- public static final Option