This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 02aa837898 [Improve][Connector-V2] Migrate Elasticsearch validation to
declarative OptionRule (#11122)
02aa837898 is described below
commit 02aa837898c821961abbef8ea567bfe2bcea2d2f
Author: zhiwei.niu <[email protected]>
AuthorDate: Sat Jun 27 21:10:53 2026 +0800
[Improve][Connector-V2] Migrate Elasticsearch validation to declarative
OptionRule (#11122)
---
.../catalog/ElasticSearchCatalogFactory.java | 49 ++-
.../client/auth/ApiKeyAuthProvider.java | 29 +-
.../client/auth/ApiKeyEncodedAuthProvider.java | 32 +-
.../client/auth/BasicAuthProvider.java | 38 +--
.../config/ElasticsearchValidators.java | 66 ++++
.../sink/ElasticsearchSinkFactory.java | 20 +-
.../elasticsearch/source/ElasticsearchSource.java | 20 +-
.../source/ElasticsearchSourceFactory.java | 50 ++-
.../catalog/ElasticSearchCatalogFactoryTest.java | 136 ++++++++
.../sink/ElasticsearchSinkFactoryTest.java | 217 +++++++++++++
.../source/ElasticsearchSourceFactoryTest.java | 342 +++++++++++++++++++++
11 files changed, 896 insertions(+), 103 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactory.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactory.java
index 76623f84fc..ecf463c011 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactory.java
@@ -18,13 +18,30 @@
package org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
import org.apache.seatunnel.api.table.factory.Factory;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.AuthTypeEnum;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchValidators.ApiKeyEncodedFormatValidator;
import com.google.auto.service.AutoService;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.API_KEY;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.API_KEY_ENCODED;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.API_KEY_ID;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.AUTH_TYPE;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.HOSTS;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.PASSWORD;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_KEY_STORE_PASSWORD;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_KEY_STORE_PATH;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_TRUST_STORE_PASSWORD;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_TRUST_STORE_PATH;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_VERIFY_CERTIFICATE;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_VERIFY_HOSTNAME;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.USERNAME;
+
@AutoService(Factory.class)
public class ElasticSearchCatalogFactory implements CatalogFactory {
@@ -40,6 +57,36 @@ public class ElasticSearchCatalogFactory implements
CatalogFactory {
@Override
public OptionRule optionRule() {
- return OptionRule.builder().build();
+ return OptionRule.builder()
+ .required(HOSTS)
+ .optional(
+ USERNAME,
+ PASSWORD,
+ TLS_VERIFY_CERTIFICATE,
+ TLS_VERIFY_HOSTNAME,
+ TLS_KEY_STORE_PATH,
+ TLS_KEY_STORE_PASSWORD,
+ TLS_TRUST_STORE_PATH,
+ TLS_TRUST_STORE_PASSWORD)
+ .optional(AUTH_TYPE)
+ .conditionalRule(
+ AUTH_TYPE,
+ AuthTypeEnum.BASIC,
+ OptionRule.builder().bundled(USERNAME,
PASSWORD).build())
+ .conditional(AUTH_TYPE, AuthTypeEnum.BASIC,
Conditions.notBlank(USERNAME))
+ .conditional(AUTH_TYPE, AuthTypeEnum.BASIC,
Conditions.notBlank(PASSWORD))
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY, API_KEY_ID,
API_KEY)
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY,
Conditions.notBlank(API_KEY_ID))
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY,
Conditions.notBlank(API_KEY))
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY_ENCODED,
API_KEY_ENCODED)
+ .conditional(
+ AUTH_TYPE,
+ AuthTypeEnum.API_KEY_ENCODED,
+ Conditions.notBlank(API_KEY_ENCODED))
+ .conditional(
+ AUTH_TYPE,
+ AuthTypeEnum.API_KEY_ENCODED,
+ Conditions.extension(API_KEY_ENCODED, new
ApiKeyEncodedFormatValidator()))
+ .build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyAuthProvider.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyAuthProvider.java
index bbd5def631..f89a1324d4 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyAuthProvider.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyAuthProvider.java
@@ -62,20 +62,14 @@ public class ApiKeyAuthProvider extends
AbstractAuthenticationProvider {
return AUTH_TYPE;
}
+ /**
+ * No-op. Presence and non-blankness of {@code auth.api_key_id} / {@code
auth.api_key} are now
+ * enforced declaratively via {@code OptionRule} (conditional on {@code
auth_type=api_key}).
+ * This method is kept to honor the {@link AuthenticationProvider}
contract.
+ */
@Override
public void validate(ReadonlyConfig config) {
- Optional<String> apiKeyId =
config.getOptional(ElasticsearchBaseOptions.API_KEY_ID);
- Optional<String> apiKey =
config.getOptional(ElasticsearchBaseOptions.API_KEY);
- Optional<String> apiKeyEncoded =
- config.getOptional(ElasticsearchBaseOptions.API_KEY_ENCODED);
-
- if (!apiKeyId.isPresent() || !apiKey.isPresent()) {
- throw new IllegalArgumentException(
- "API key authentication with auth_type='api_key' requires
both api_key_id and api_key");
- }
- validateApiKeyIdAndSecret(apiKeyId.get(), apiKey.get());
-
- log.debug("API key authentication configuration validated");
+ // intentionally empty - validation handled by OptionRule
}
/**
@@ -95,15 +89,4 @@ public class ApiKeyAuthProvider extends
AbstractAuthenticationProvider {
return null;
}
-
- /** Validate API key ID and secret. */
- private void validateApiKeyIdAndSecret(String apiKeyId, String apiKey) {
- if (apiKeyId == null || apiKeyId.trim().isEmpty()) {
- throw new IllegalArgumentException("API key ID cannot be null or
empty");
- }
-
- if (apiKey == null || apiKey.trim().isEmpty()) {
- throw new IllegalArgumentException("API key cannot be null or
empty");
- }
- }
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyEncodedAuthProvider.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyEncodedAuthProvider.java
index 200bc13705..d3910303f6 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyEncodedAuthProvider.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/ApiKeyEncodedAuthProvider.java
@@ -24,8 +24,6 @@ import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import lombok.extern.slf4j.Slf4j;
-import java.nio.charset.StandardCharsets;
-import java.util.Base64;
import java.util.Optional;
@Slf4j
@@ -66,34 +64,6 @@ public class ApiKeyEncodedAuthProvider extends
AbstractAuthenticationProvider {
@Override
public void validate(ReadonlyConfig config) {
- Optional<String> apiKeyEncoded =
- config.getOptional(ElasticsearchBaseOptions.API_KEY_ENCODED);
- if (!apiKeyEncoded.isPresent()) {
- throw new IllegalArgumentException(
- "API key authentication with auth_type='api_key_encoded'
requires api_key_encoded");
- }
- validateEncodedApiKey(apiKeyEncoded.get());
-
- log.debug("Encoded API key authentication configuration validated");
- }
-
- /** Validate encoded API key. */
- private void validateEncodedApiKey(String apiKeyEncoded) {
- if (apiKeyEncoded == null || apiKeyEncoded.trim().isEmpty()) {
- throw new IllegalArgumentException("Encoded API key cannot be null
or empty");
- }
-
- try {
- byte[] decoded = Base64.getDecoder().decode(apiKeyEncoded);
- String decodedStr = new String(decoded, StandardCharsets.UTF_8);
-
- if (!decodedStr.contains(":")) {
- throw new IllegalArgumentException(
- "Encoded API key must be Base64 encoded 'id:key'
format");
- }
- } catch (IllegalArgumentException e) {
- throw new IllegalArgumentException(
- "Invalid encoded API key format: " + e.getMessage(), e);
- }
+ // intentionally empty - validation handled by OptionRule
}
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/BasicAuthProvider.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/BasicAuthProvider.java
index 7e4d0ce010..ee169ad1fc 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/BasicAuthProvider.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/auth/BasicAuthProvider.java
@@ -60,38 +60,14 @@ public class BasicAuthProvider extends
AbstractAuthenticationProvider {
return AUTH_TYPE;
}
+ /**
+ * No-op. Username/password presence and pairing are now enforced
declaratively via {@code
+ * OptionRule} (see {@code ElasticsearchSourceFactory} / {@code
ElasticsearchSinkFactory} /
+ * {@code ElasticSearchCatalogFactory}). This method is kept to honor the
{@link
+ * AuthenticationProvider} contract.
+ */
@Override
public void validate(ReadonlyConfig config) {
- Optional<String> username =
config.getOptional(ElasticsearchBaseOptions.USERNAME);
- Optional<String> password =
config.getOptional(ElasticsearchBaseOptions.PASSWORD);
-
- // For backward compatibility, we allow basic auth to be optional
- // If username is provided, password must also be provided
- if (username.isPresent() && !password.isPresent()) {
- throw new IllegalArgumentException(
- "Password is required when username is provided for basic
authentication");
- }
-
- if (!username.isPresent() && password.isPresent()) {
- throw new IllegalArgumentException(
- "Username is required when password is provided for basic
authentication");
- }
-
- if (username.isPresent()) {
- String usernameValue = username.get();
- if (usernameValue == null || usernameValue.trim().isEmpty()) {
- throw new IllegalArgumentException("Username cannot be null or
empty");
- }
-
- String passwordValue = password.get();
- if (passwordValue == null || passwordValue.trim().isEmpty()) {
- throw new IllegalArgumentException("Password cannot be null or
empty");
- }
-
- log.debug("Basic authentication configuration validated for user:
{}", usernameValue);
- } else {
- log.debug(
- "No basic authentication credentials provided -
authentication will be skipped");
- }
+ // intentionally empty - validation handled by OptionRule
}
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchValidators.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchValidators.java
new file mode 100644
index 0000000000..6d429527bc
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchValidators.java
@@ -0,0 +1,66 @@
+/*
+ * 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.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+
+import lombok.experimental.UtilityClass;
+import lombok.extern.slf4j.Slf4j;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+
+/**
+ * Reusable {@link ConditionExtension} validators shared by Elasticsearch
Source / Sink / Catalog
+ * factories.
+ *
+ * <p>All validators return {@code false} on failure (instead of throwing) so
that {@code
+ * ConfigValidator} can aggregate errors across the entire rule.
+ */
+@UtilityClass
+public final class ElasticsearchValidators {
+
+ /**
+ * Validates that {@code auth.api_key_encoded} is a Base64-encoded {@code
id:key} string.
+ *
+ * <p>Skips when the value is null/blank — presence and non-blankness are
enforced separately by
+ * the corresponding conditional rules.
+ */
+ @Slf4j
+ public static class ApiKeyEncodedFormatValidator implements
ConditionExtension<String> {
+ @Override
+ public String description() {
+ return "'auth.api_key_encoded' must be a Base64-encoded 'id:key'
string";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return true;
+ }
+ try {
+ byte[] decoded = Base64.getDecoder().decode(value);
+ return new String(decoded,
StandardCharsets.UTF_8).contains(":");
+ } catch (IllegalArgumentException e) {
+ log.warn("Failed to decode 'auth.api_key_encoded' as Base64",
e);
+ return false;
+ }
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactory.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactory.java
index 31b1438eec..0bbab15e35 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactory.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.connectors.seatunnel.elasticsearch.sink;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
@@ -28,6 +29,7 @@ import
org.apache.seatunnel.api.table.factory.TableSinkFactory;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.AuthTypeEnum;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchValidators.ApiKeyEncodedFormatValidator;
import com.google.auto.service.AutoService;
@@ -63,8 +65,8 @@ public class ElasticsearchSinkFactory implements
TableSinkFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
+ .required(HOSTS)
.required(
- HOSTS,
INDEX,
ElasticsearchSinkOptions.SCHEMA_SAVE_MODE,
ElasticsearchSinkOptions.DATA_SAVE_MODE)
@@ -86,8 +88,24 @@ public class ElasticsearchSinkFactory implements
TableSinkFactory {
VECTORIZATION_FIELDS,
VECTOR_DIMENSIONS)
.optional(AUTH_TYPE)
+ .conditionalRule(
+ AUTH_TYPE,
+ AuthTypeEnum.BASIC,
+ OptionRule.builder().bundled(USERNAME,
PASSWORD).build())
+ .conditional(AUTH_TYPE, AuthTypeEnum.BASIC,
Conditions.notBlank(USERNAME))
+ .conditional(AUTH_TYPE, AuthTypeEnum.BASIC,
Conditions.notBlank(PASSWORD))
.conditional(AUTH_TYPE, AuthTypeEnum.API_KEY, API_KEY_ID,
API_KEY)
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY,
Conditions.notBlank(API_KEY_ID))
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY,
Conditions.notBlank(API_KEY))
.conditional(AUTH_TYPE, AuthTypeEnum.API_KEY_ENCODED,
API_KEY_ENCODED)
+ .conditional(
+ AUTH_TYPE,
+ AuthTypeEnum.API_KEY_ENCODED,
+ Conditions.notBlank(API_KEY_ENCODED))
+ .conditional(
+ AUTH_TYPE,
+ AuthTypeEnum.API_KEY_ENCODED,
+ Conditions.extension(API_KEY_ENCODED, new
ApiKeyEncodedFormatValidator()))
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
index 46b01f2e89..6785b25cdb 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSource.java
@@ -43,8 +43,6 @@ import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.Elasticsea
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SearchApiTypeEnum;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SearchTypeEnum;
-import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorErrorCode;
-import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorException;
import org.apache.commons.collections4.CollectionUtils;
@@ -76,23 +74,10 @@ public class ElasticsearchSource
boolean multiSource =
config.getOptional(ElasticsearchSourceOptions.INDEX_LIST).isPresent();
boolean singleSource =
config.getOptional(ElasticsearchSourceOptions.INDEX).isPresent();
- boolean sqlQuery = config.getOptional(SQL_QUERY).isPresent();
-
- if (SearchTypeEnum.SQL.equals(config.get(SEARCH_TYPE)) && !sqlQuery) {
- throw new ElasticsearchConnectorException(
- ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR_02,
-
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR_02.getDescription());
- }
-
if (multiSource && singleSource) {
log.warn(
"Elasticsearch Source config warn: when both 'index' and
'index_list' are present in the configuration, only the 'index_list'
configuration will take effect");
}
- if (!multiSource && !singleSource) {
- throw new ElasticsearchConnectorException(
- ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR_01,
-
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR_01.getDescription());
- }
if (multiSource) {
this.elasticsearchConfigList = createMultiSource(config);
} else {
@@ -107,6 +92,11 @@ public class ElasticsearchSource
configMaps.stream().map(ReadonlyConfig::fromMap).collect(Collectors.toList());
List<ElasticsearchConfig> elasticsearchConfigList = new
ArrayList<>(configList.size());
for (ReadonlyConfig readonlyConfig : configList) {
+ // NOTE: per-entry configs inside `index_list` are NOT validated
by the factory's
+ // OptionRule (which only validates the top-level config). If an
entry is missing
+ // `index`, or sets `search_type=SQL` without `sql_query`,
parseOneIndexQueryConfig
+ // below will fail at runtime. This is a pre-existing limitation;
revisit if/when
+ // per-entry declarative validation is supported.
ElasticsearchConfig elasticsearchConfig =
parseOneIndexQueryConfig(readonlyConfig);
elasticsearchConfigList.add(elasticsearchConfig);
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactory.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactory.java
index fc13b74d25..cbf9c4757a 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactory.java
@@ -17,6 +17,9 @@
package org.apache.seatunnel.connectors.seatunnel.elasticsearch.source;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
@@ -25,10 +28,13 @@ import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.AuthTypeEnum;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchValidators.ApiKeyEncodedFormatValidator;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SearchTypeEnum;
import com.google.auto.service.AutoService;
import java.io.Serializable;
+import java.util.List;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.API_KEY;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.API_KEY_ENCODED;
@@ -44,6 +50,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.Ela
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_VERIFY_CERTIFICATE;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.TLS_VERIFY_HOSTNAME;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchBaseOptions.USERNAME;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.ARRAY_COLUMN;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.INDEX_LIST;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.PIT_BATCH_SIZE;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.PIT_KEEP_ALIVE;
@@ -53,6 +60,9 @@ import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.Ela
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.SCROLL_TIME;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.SEARCH_API_TYPE;
import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.SEARCH_TYPE;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.SLICE_MAX;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.SOURCE;
+import static
org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSourceOptions.SQL_QUERY;
@AutoService(Factory.class)
public class ElasticsearchSourceFactory implements TableSourceFactory {
@@ -64,7 +74,7 @@ public class ElasticsearchSourceFactory implements
TableSourceFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(HOSTS)
+ .required(HOSTS, Conditions.extension(HOSTS, new
RequireIndexValidator()))
.optional(
INDEX,
INDEX_LIST,
@@ -78,6 +88,9 @@ public class ElasticsearchSourceFactory implements
TableSourceFactory {
PIT_BATCH_SIZE,
SEARCH_API_TYPE,
SEARCH_TYPE,
+ SOURCE,
+ ARRAY_COLUMN,
+ SLICE_MAX,
TLS_VERIFY_CERTIFICATE,
TLS_VERIFY_HOSTNAME,
TLS_KEY_STORE_PATH,
@@ -85,8 +98,26 @@ public class ElasticsearchSourceFactory implements
TableSourceFactory {
TLS_TRUST_STORE_PATH,
TLS_TRUST_STORE_PASSWORD)
.optional(AUTH_TYPE)
+ .conditionalRule(
+ AUTH_TYPE,
+ AuthTypeEnum.BASIC,
+ OptionRule.builder().bundled(USERNAME,
PASSWORD).build())
+ .conditional(AUTH_TYPE, AuthTypeEnum.BASIC,
Conditions.notBlank(USERNAME))
+ .conditional(AUTH_TYPE, AuthTypeEnum.BASIC,
Conditions.notBlank(PASSWORD))
.conditional(AUTH_TYPE, AuthTypeEnum.API_KEY, API_KEY_ID,
API_KEY)
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY,
Conditions.notBlank(API_KEY_ID))
+ .conditional(AUTH_TYPE, AuthTypeEnum.API_KEY,
Conditions.notBlank(API_KEY))
.conditional(AUTH_TYPE, AuthTypeEnum.API_KEY_ENCODED,
API_KEY_ENCODED)
+ .conditional(
+ AUTH_TYPE,
+ AuthTypeEnum.API_KEY_ENCODED,
+ Conditions.notBlank(API_KEY_ENCODED))
+ .conditional(
+ AUTH_TYPE,
+ AuthTypeEnum.API_KEY_ENCODED,
+ Conditions.extension(API_KEY_ENCODED, new
ApiKeyEncodedFormatValidator()))
+ .conditional(SEARCH_TYPE, SearchTypeEnum.SQL, SQL_QUERY)
+ .conditional(SEARCH_TYPE, SearchTypeEnum.SQL,
Conditions.notBlank(SQL_QUERY))
.build();
}
@@ -101,4 +132,21 @@ public class ElasticsearchSourceFactory implements
TableSourceFactory {
public Class<? extends SeaTunnelSource> getSourceClass() {
return ElasticsearchSource.class;
}
+
+ /**
+ * Validates that at least one of {@code index} or {@code index_list} is
provided. Attached to
+ * the always-present {@code HOSTS} so the rule runs even when both index
options are missing.
+ */
+ static class RequireIndexValidator implements
ConditionExtension<List<String>> {
+ @Override
+ public String description() {
+ return "at least one of 'index' or 'index_list' must be provided";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<String> value) {
+ return config.getOptional(INDEX).isPresent()
+ || config.getOptional(INDEX_LIST).isPresent();
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactoryTest.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactoryTest.java
new file mode 100644
index 0000000000..10b0444928
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactoryTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.catalog;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+class ElasticSearchCatalogFactoryTest {
+
+ private OptionRule rule;
+
+ @BeforeEach
+ void setUp() {
+ rule = new ElasticSearchCatalogFactory().optionRule();
+ }
+
+ @Test
+ void testValidConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testMissingHostsFails() {
+ Map<String, Object> config = new HashMap<>();
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("hosts"));
+ }
+
+ @Test
+ void testUsernameWithoutPasswordFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("username", "admin");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("password"));
+ }
+
+ @Test
+ void testValidBasicAuth() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("username", "admin");
+ config.put("password", "secret");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testBasicAuthBlankUsernameFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("username", " ");
+ config.put("password", "secret");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("blank"));
+ }
+
+ @Test
+ void testBasicAuthBlankPasswordFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("username", "admin");
+ config.put("password", " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("password") ||
ex.getMessage().contains("blank"));
+ }
+
+ @Test
+ void testApiKeyAuthMissingKeysFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("auth_type", "API_KEY");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("api_key_id") ||
ex.getMessage().contains("api_key"));
+ }
+
+ @Test
+ void testApiKeyEncodedInvalidFormatFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("auth_type", "API_KEY_ENCODED");
+ config.put("auth.api_key_encoded", "not-base64-!@#");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("Base64"));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactoryTest.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactoryTest.java
new file mode 100644
index 0000000000..38fd8e8d80
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactoryTest.java
@@ -0,0 +1,217 @@
+/*
+ * 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.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+class ElasticsearchSinkFactoryTest {
+
+ private OptionRule rule;
+
+ @BeforeEach
+ void setUp() {
+ rule = new ElasticsearchSinkFactory().optionRule();
+ }
+
+ private Map<String, Object> validSinkConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("schema_save_mode", "CREATE_SCHEMA_WHEN_NOT_EXIST");
+ config.put("data_save_mode", "APPEND_DATA");
+ return config;
+ }
+
+ @Test
+ void testValidConfig() {
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(validSinkConfig())).validate(rule));
+ }
+
+ @Test
+ void testMissingHostsFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.remove("hosts");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("hosts"));
+ }
+
+ @Test
+ void testMissingIndexFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.remove("index");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("index"));
+ }
+
+ @Test
+ void testUsernameWithoutPasswordFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("username", "admin");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("password"));
+ }
+
+ @Test
+ void testPasswordWithoutUsernameFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("password", "secret");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("password"));
+ }
+
+ @Test
+ void testValidBasicAuth() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("username", "admin");
+ config.put("password", "secret");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testBasicAuthBlankUsernameFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("username", " ");
+ config.put("password", "secret");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("blank"));
+ }
+
+ @Test
+ void testBasicAuthBlankPasswordFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("username", "admin");
+ config.put("password", " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("password") ||
ex.getMessage().contains("blank"));
+ }
+
+ @Test
+ void testApiKeyAuthMissingKeysFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("auth_type", "API_KEY");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("api_key_id") ||
ex.getMessage().contains("api_key"));
+ }
+
+ @Test
+ void testApiKeyAuthBlankValueFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("auth_type", "API_KEY");
+ config.put("auth.api_key_id", "valid_id");
+ config.put("auth.api_key", " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("api_key"));
+ }
+
+ @Test
+ void testApiKeyEncodedAuthBlankFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("auth_type", "API_KEY_ENCODED");
+ config.put("auth.api_key_encoded", " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("api_key_encoded"));
+ }
+
+ @Test
+ void testApiKeyEncodedInvalidFormatFails() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("auth_type", "API_KEY_ENCODED");
+ // valid Base64 but decoded value lacks the required ':' separator
+ config.put("auth.api_key_encoded", "bm9Db2xvbkhlcmU=");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("id:key"));
+ }
+
+ @Test
+ void testApiKeyEncodedValidFormatPasses() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("auth_type", "API_KEY_ENCODED");
+ config.put("auth.api_key_encoded", "aWQ6a2V5");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testValidApiKeyAuth() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("auth_type", "API_KEY");
+ config.put("auth.api_key_id", "my_key_id");
+ config.put("auth.api_key", "my_secret");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testApiKeyAuthWithResidualUsernameDoesNotFail() {
+ Map<String, Object> config = validSinkConfig();
+ config.put("auth_type", "API_KEY");
+ config.put("auth.api_key_id", "my_key_id");
+ config.put("auth.api_key", "my_secret");
+ config.put("username", "residual_user");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactoryTest.java
new file mode 100644
index 0000000000..d1b6d2f4db
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactoryTest.java
@@ -0,0 +1,342 @@
+/*
+ * 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.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+class ElasticsearchSourceFactoryTest {
+
+ private OptionRule rule;
+
+ @BeforeEach
+ void setUp() {
+ rule = new ElasticsearchSourceFactory().optionRule();
+ }
+
+ @Test
+ void testValidConfigWithIndex() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testValidConfigWithIndexList() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put(
+ "index_list",
+ Arrays.asList(
+ new HashMap<String, Object>() {
+ {
+ put("index", "idx1");
+ }
+ }));
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testMissingHostsFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("index", "test_index");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("hosts"));
+ }
+
+ @Test
+ void testMissingIndexAndIndexListFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("index") ||
ex.getMessage().contains("index_list"));
+ }
+
+ @Test
+ void testSqlSearchTypeMissingSqlQueryFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("search_type", "SQL");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("sql_query"));
+ }
+
+ @Test
+ void testSqlSearchTypeBlankSqlQueryFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("search_type", "SQL");
+ config.put("sql_query", " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("sql_query"));
+ }
+
+ @Test
+ void testSqlSearchTypeWithValidSqlQuery() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("search_type", "SQL");
+ config.put("sql_query", "SELECT * FROM test_index");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testUsernameWithoutPasswordFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("username", "admin");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("password"));
+ }
+
+ @Test
+ void testPasswordWithoutUsernameFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("password", "secret");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("password"));
+ }
+
+ @Test
+ void testValidBasicAuth() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("username", "admin");
+ config.put("password", "secret");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testBasicAuthBlankUsernameFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("username", " ");
+ config.put("password", "secret");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("username") ||
ex.getMessage().contains("blank"));
+ }
+
+ @Test
+ void testBasicAuthBlankPasswordFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("username", "admin");
+ config.put("password", " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("password") ||
ex.getMessage().contains("blank"));
+ }
+
+ @Test
+ void testApiKeyAuthMissingKeysFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(
+ ex.getMessage().contains("api_key_id") ||
ex.getMessage().contains("api_key"));
+ }
+
+ @Test
+ void testApiKeyAuthBlankKeyIdFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY");
+ config.put("auth.api_key_id", " ");
+ config.put("auth.api_key", "my_secret");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("api_key_id"));
+ }
+
+ @Test
+ void testApiKeyEncodedAuthMissingFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY_ENCODED");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("api_key_encoded"));
+ }
+
+ @Test
+ void testApiKeyEncodedAuthBlankFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY_ENCODED");
+ config.put("auth.api_key_encoded", " ");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("api_key_encoded"));
+ }
+
+ @Test
+ void testApiKeyEncodedInvalidBase64Fails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY_ENCODED");
+ config.put("auth.api_key_encoded", "not-valid-base64!@#");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("Base64"));
+ }
+
+ @Test
+ void testApiKeyEncodedBase64WithoutColonFails() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY_ENCODED");
+ // "noColonHere" Base64-encoded
+ config.put("auth.api_key_encoded", "bm9Db2xvbkhlcmU=");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ Assertions.assertTrue(ex.getMessage().contains("id:key"));
+ }
+
+ @Test
+ void testApiKeyEncodedValidFormatPasses() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY_ENCODED");
+ // "id:key" Base64-encoded
+ config.put("auth.api_key_encoded", "aWQ6a2V5");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ @Test
+ void testValidApiKeyAuth() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY");
+ config.put("auth.api_key_id", "my_key_id");
+ config.put("auth.api_key", "my_secret");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+
+ /**
+ * Verify the error-aggregation contract: when multiple independent rules
fail in one config,
+ * all errors are reported together in a single exception message instead
of fail-fast on the
+ * first one. Triggers (1) username-without-password (auth_type=BASIC),
(2) search_type=SQL
+ * without sql_query.
+ */
+ @Test
+ void testMultipleValidationErrorsAreAggregated() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("username", "admin");
+ config.put("search_type", "SQL");
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ String msg = ex.getMessage();
+ Assertions.assertTrue(
+ msg.contains("password") || msg.contains("username"),
+ "expected basic-auth pair error in: " + msg);
+ Assertions.assertTrue(msg.contains("sql_query"), "expected sql_query
error in: " + msg);
+ }
+
+ @Test
+ void testApiKeyAuthWithResidualUsernameDoesNotFail() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("hosts", Arrays.asList("localhost:9200"));
+ config.put("index", "test_index");
+ config.put("auth_type", "API_KEY");
+ config.put("auth.api_key_id", "my_key_id");
+ config.put("auth.api_key", "my_secret");
+ config.put("username", "residual_user");
+ Assertions.assertDoesNotThrow(
+ () ->
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule));
+ }
+}
