DanielCarter-stack commented on PR #10201:
URL: https://github.com/apache/seatunnel/pull/10201#issuecomment-3847051327
<!-- code-pr-reviewer -->
<!-- cpr:pr_reply_v2_parts {"group": "apache/seatunnel#10201", "part": 1,
"total": 1} -->
### Issue 1: Missing Elasticsearch version compatibility check
**Location**: `ElasticsearchSource.java:188-194`
```java
// Parse runtime fields configuration
Map<String, Object> runtimeFields = null;
if
(readonlyConfig.getOptional(ElasticsearchSourceOptions.RUNTIME_FIELDS).isPresent())
{
runtimeFields =
parseRuntimeFields(
readonlyConfig.get(ElasticsearchSourceOptions.RUNTIME_FIELDS));
}
```
**Related context**:
- Caller: `ElasticsearchSource` constructor
- Dependency: `EsRestClient.getClusterInfo()` can obtain cluster version
information
- Documentation explicitly requires Elasticsearch 7.11+
**Problem description**:
When a user configures `runtime_fields` but the connected Elasticsearch
version is lower than 7.11, the job will only fail when it reaches the query
stage (may receive a 400 error or field does not exist error). This leads to:
1. Delayed error detection, wasting resources
2. Error messages may not be clear enough, making it difficult for users to
identify the problem
**Potential risks**:
- Users configure runtime fields on older versions of Elasticsearch, causing
job failures
- Error messages may not be clear enough, increasing troubleshooting
difficulty
- Impacts user experience
**Impact scope**:
- Direct impact: Elasticsearch Source tasks configured with `runtime_fields`
- Indirect impact: Data sync jobs that depend on this functionality
- Affected area: Single Connector (Elasticsearch)
**Severity**: MAJOR
**Improvement suggestions**:
```java
// Add version check in parseOneIndexQueryConfig method
if
(readonlyConfig.getOptional(ElasticsearchSourceOptions.RUNTIME_FIELDS).isPresent())
{
// Check Elasticsearch version
try (EsRestClient esRestClient =
EsRestClient.createInstance(connectionConfig)) {
ElasticsearchClusterInfo clusterInfo = esRestClient.getClusterInfo();
String version = clusterInfo.getClusterVersion();
if (!isRuntimeFieldsSupported(version)) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR,
String.format("Runtime fields require Elasticsearch 7.11 or
higher, " +
"but current cluster version is: %s", version));
}
runtimeFields = parseRuntimeFields(
readonlyConfig.get(ElasticsearchSourceOptions.RUNTIME_FIELDS));
}
}
// Add version comparison helper method
private boolean isRuntimeFieldsSupported(String version) {
String[] parts = version.split("\\.");
int major = Integer.parseInt(parts[0]);
int minor = parts.length > 1 ? Integer.parseInt(parts[1]) : 0;
return major > 7 || (major == 7 && minor >= 11);
}
```
**Rationale**:
1. Perform version check during configuration parsing phase to fail fast
2. Provide clear error messages, explicitly informing users of version
requirements
3. Avoid users encountering difficult-to-understand errors at runtime
---
### Issue 2: Runtime Fields configuration validation is not strict enough
**Location**: `ElasticsearchSource.java:220-257`
```java
private Map<String, Object> parseRuntimeFields(List<Map<String, Object>>
runtimeFieldsList) {
if (runtimeFieldsList == null || runtimeFieldsList.isEmpty()) {
return null;
}
Map<String, Object> runtimeMappings = new java.util.LinkedHashMap<>();
for (Map<String, Object> fieldConfig : runtimeFieldsList) {
String name = (String) fieldConfig.get("name");
String type = (String) fieldConfig.get("type");
String script = (String) fieldConfig.get("script");
if (name == null || type == null || script == null) {
log.warn("Invalid runtime field configuration: {}, skipping",
fieldConfig);
continue; // Just skip, don't throw exception
}
// ...
}
return runtimeMappings.isEmpty() ? null : runtimeMappings;
}
```
**Related context**:
- Documentation explicitly lists supported types: boolean, date, double,
geo_point, ip, keyword, long
- Users may configure unsupported types or invalid field names
**Problem description**:
Current implementation only checks whether `name`, `type`, `script` are
null, but does not:
1. Validate whether `type` is in the supported type list
2. Validate whether `name` complies with Elasticsearch field naming
conventions
3. Validate whether `script` is an empty string
4. Simply skip invalid configurations, which may cause users to configure
multiple fields with partial failures without realizing it
**Potential risks**:
- Users configure unsupported types (e.g., "text"), errors only occur during
query
- Field name conflicts or naming errors cause query failures
- Partial field configuration failures but job continues running, leading to
incomplete data
**Impact scope**:
- Direct impact: Users using Runtime Fields
- Indirect impact: Downstream jobs that depend on runtime field calculation
results
- Affected area: Single Connector
**Severity**: MAJOR
**Improvement suggestions**:
```java
private static final Set<String> SUPPORTED_RUNTIME_FIELD_TYPES =
new HashSet<>(Arrays.asList(
"boolean", "date", "double", "geo_point", "ip", "keyword", "long"
));
private Map<String, Object> parseRuntimeFields(List<Map<String, Object>>
runtimeFieldsList) {
if (runtimeFieldsList == null || runtimeFieldsList.isEmpty()) {
return null;
}
Map<String, Object> runtimeMappings = new java.util.LinkedHashMap<>();
Set<String> fieldNames = new HashSet<>();
for (int i = 0; i < runtimeFieldsList.size(); i++) {
Map<String, Object> fieldConfig = runtimeFieldsList.get(i);
String name = (String) fieldConfig.get("name");
String type = (String) fieldConfig.get("type");
String script = (String) fieldConfig.get("script");
// Validate required fields
if (name == null || name.trim().isEmpty()) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR,
String.format("Runtime field at index %d: 'name' cannot be
null or empty", i));
}
if (type == null || type.trim().isEmpty()) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR,
String.format("Runtime field '%s': 'type' cannot be null or
empty", name));
}
if (script == null || script.trim().isEmpty()) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR,
String.format("Runtime field '%s': 'script' cannot be null
or empty", name));
}
// Validate if type is supported
if (!SUPPORTED_RUNTIME_FIELD_TYPES.contains(type.toLowerCase())) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR,
String.format("Runtime field '%s': unsupported type '%s'. " +
"Supported types are: %s",
name, type, SUPPORTED_RUNTIME_FIELD_TYPES));
}
// Validate field name uniqueness
if (fieldNames.contains(name)) {
log.warn("Duplicate runtime field name: {}, skipping", name);
continue;
}
fieldNames.add(name);
// Validate field name format (Elasticsearch field name rules)
if (!name.matches("^[a-zA-Z_][a-zA-Z0-9_]*$")) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SOURCE_CONFIG_ERROR,
String.format("Runtime field name '%s' contains invalid
characters. " +
"Field names must start with a letter or
underscore " +
"and contain only letters, digits, and
underscores", name));
}
// Build configuration
Map<String, Object> fieldDef = new java.util.LinkedHashMap<>();
fieldDef.put("type", type);
Map<String, Object> scriptDef = new java.util.LinkedHashMap<>();
scriptDef.put("source", script);
if (fieldConfig.containsKey("script_lang")) {
scriptDef.put("lang", fieldConfig.get("script_lang"));
}
if (fieldConfig.containsKey("script_params")) {
scriptDef.put("params", fieldConfig.get("script_params"));
}
fieldDef.put("script", scriptDef);
runtimeMappings.put(name, fieldDef);
}
return runtimeMappings.isEmpty() ? null : runtimeMappings;
}
```
**Rationale**:
1. Perform comprehensive validation during configuration parsing phase to
fail fast
2. Provide clear error messages to guide users in correcting configuration
3. Prevent incomplete data issues caused by partial field configuration
failures
4. Validate field name uniqueness to avoid conflicts
---
### Issue 3: Insufficient E2E test coverage
**Location**: `ElasticsearchIT.java:416-425`
```java
@TestTemplate
public void testElasticsearchSourceWithRuntimeFields(TestContainer container)
throws IOException, InterruptedException {
Container.ExecResult execResult =
container.executeJob(
"/elasticsearch/elasticsearch_source_with_runtime_fields.conf");
Assertions.assertEquals(0, execResult.getExitCode(), "Job should
complete successfully");
log.info("Runtime fields test completed successfully");
log.info("Job output: {}", execResult.getStdout());
}
```
**Related context**:
- Test configuration file contains 3 runtime fields
- Only validates job exit code, does not verify actual output data
**Problem description**:
Current E2E tests have the following issues:
1. Only validates that the job completes successfully, does not verify
whether runtime field calculation results are correct
2. Does not cover edge cases (empty configuration, invalid configuration)
3. Does not test error scenarios (script errors, field conflicts, etc.)
4. Does not test integration with existing functionality (e.g., using
runtime_fields and regular fields simultaneously)
**Potential risks**:
- Runtime field calculation logic errors cannot be detected by tests
- Functional regressions caused by Elasticsearch API changes cannot be
detected in time
- Users may encounter issues not covered by tests
**Impact scope**:
- Direct impact: Code quality and stability
- Indirect impact: User experience and maintenance cost
- Affected area: E2E test suite
**Severity**: MAJOR
**Improvement suggestions**:
```java
@TestTemplate
public void testElasticsearchSourceWithRuntimeFields(TestContainer container)
throws IOException, InterruptedException {
// Prepare test data
String indexName = "st_index_runtime";
createTestIndexWithDataForRuntimeFields(indexName);
// Execute job
Container.ExecResult execResult =
container.executeJob("/elasticsearch/elasticsearch_source_with_runtime_fields.conf");
Assertions.assertEquals(0, execResult.getExitCode(), "Job should
complete successfully");
// Verify runtime field calculation results
List<Map<String, Object>> docs = getDocumentsFromIndex(indexName);
Assertions.assertFalse(docs.isEmpty(), "Should have at least one
document");
Map<String, Object> doc = docs.get(0);
// Verify day_of_week calculation is correct
Assertions.assertEquals("MONDAY", doc.get("day_of_week"));
// Verify c_int_doubled calculation is correct
Assertions.assertEquals(20, doc.get("c_int_doubled"));
// Verify full_name calculation is correct
Assertions.assertEquals("test_1_computed", doc.get("full_name"));
log.info("Runtime fields test completed successfully with data
validation");
}
// Add additional test cases
@TestTemplate
public void testElasticsearchSourceWithInvalidRuntimeFieldType(TestContainer
container)
throws IOException, InterruptedException {
// This test should fail, verify type checking logic
Container.ExecResult execResult = container.executeJob(
"/elasticsearch/elasticsearch_source_with_invalid_runtime_field_type.conf");
Assertions.assertNotEquals(0, execResult.getExitCode(),
"Job should fail with invalid runtime field type");
Assertions.assertTrue(execResult.getStderr().contains("unsupported
type"),
"Error message should mention unsupported type");
}
@TestTemplate
public void testElasticsearchSourceWithEmptyRuntimeFields(TestContainer
container)
throws IOException, InterruptedException {
// Test empty list configuration
Container.ExecResult execResult = container.executeJob(
"/elasticsearch/elasticsearch_source_with_empty_runtime_fields.conf");
Assertions.assertEquals(0, execResult.getExitCode(),
"Job should succeed with empty runtime fields list");
}
```
**Rationale**:
1. Enhance test credibility to ensure functionality truly works as expected
2. Detect potential functional regression issues early
3. Improve code quality and user confidence
4. Cover more edge cases and error scenarios
---
### Issue 4: Documentation lacks detailed explanation of script security and
performance impact
**Location**: `docs/en/connectors/source/Elasticsearch.md:218-248`
**Related context**:
- Documentation has a "Performance Considerations" section, but the content
is brief
- No mention of script injection risks and best practices
**Problem description**:
Current documentation has the following shortcomings:
1. Does not explicitly explain the security risks of Painless scripts
2. Does not provide best practices for script writing
3. Does not explain in which scenarios runtime fields should be avoided
4. Does not provide suggestions for script performance optimization
**Potential risks**:
- Users may write inefficient scripts, affecting overall cluster performance
- Users may not understand the limitations of runtime fields and use them
inappropriately in production environments
- Lack of security warnings may lead to sensitive data leaks
**Impact scope**:
- Direct impact: User experience and cluster performance
- Indirect impact: Support cost
- Affected area: Documentation
**Severity**: MINOR
**Improvement suggestions**:
Add the following content to the documentation:
```markdown
### runtime_fields [array]
**Security Considerations:**
Runtime fields execute Painless scripts on your Elasticsearch cluster. Keep
these security best practices in mind:
1. **Script Injection**: Avoid using user-provided input directly in scripts
without proper validation
2. **Resource Limits**: Complex scripts can consume significant CPU
resources and impact query performance
3. **Access Control**: Ensure only authorized users can configure runtime
fields
4. **Data Exposure**: Runtime fields can access all document fields, be
cautious about sensitive data
**Performance Best Practices:**
1. **Keep Scripts Simple**: Prefer simple arithmetic and string operations
over complex logic
2. **Avoid Heavy Computations**: Scripts like regex, date parsing with
complex formats, or loops can be slow
3. **Use Indexed Fields for Frequent Queries**: If a runtime field is used
frequently, consider indexing it
4. **Test with Real Data**: Always test runtime fields with a realistic
dataset before production use
5. **Monitor Query Performance**: Use Elasticsearch's profiling API to
identify slow runtime fields
**When to Avoid Runtime Fields:**
- High-throughput pipelines (more than 10,000 docs/second)
- Complex aggregations on runtime fields
- Frequently executed queries
- When deterministic performance is required
**Example of Efficient vs Inefficient Scripts:**
Good (simple arithmetic):
```hocon
{
name = "total_amount"
type = "double"
script = "emit(doc['quantity'].value * doc['price'].value)"
}
```
Avoid (complex logic):
```hocon
{
name = "complex_calculation"
type = "double"
script = "double result = 0; for (int i = 0; i <
doc['array_field'].length; i++) { result += doc['array_field'][i]; }
emit(result)"
}
```
```
** Rationale**:
1. 帮助用户理解运行时字段的限制和风险
2. 提供实用的性能优化建议
3. 减少因不当使用导致的性能问题和支持请求
4. 提升文档的专业性和完整性
---
# ## Issue 5: Unhandled conflicts between Runtime Fields and Source fields
** Location**:`ElasticsearchSource.java` and documentation
** Related context**:
- 用户可以在 `source` 列表中同时指定普通字段和运行时字段
- 运行时字段和索引中的字段可能有同名冲突
** Problem description**:
当前实现没有处理以下场景：
1. 用户定义的运行时字段名与索引中已存在的字段名相同
2. Elasticsearch 会优先返回索引中的字段值，而不是运行时字段的计算值
3. 用户可能误以为使用了运行时字段的计算结果，实际使用的是索引字段
** Potential risks**:
- 用户得到意外的计算结果
- 数据不一致，难以排查
- 可能导致业务逻辑错误
** Scope of impact**:
- 直接影响：配置了运行时字段但与索引字段重名的用户
- 间接影响：数据准确性和业务逻辑
- 影响面：单个 Connector
** Severity**:MINOR
** Improvement suggestions**:
```java
private Map<String, Object> parseRuntimeFields(
List<Map<String, Object>> runtimeFieldsList,
List<String> sourceFields,
String index) {
// ... existing code ...
for (Map<String, Object> fieldConfig : runtimeFieldsList) {
String name = (String) fieldConfig.get("name");
// Check for conflicts with index fields
if (sourceFields != null && sourceFields.contains(name)) {
log.warn("Runtime field '{}' conflicts with existing index
field. " +
"The index field value will be used. " +
"Consider renaming the runtime field.", name);
// Options: 1) skip this runtime field, 2) throw exception, 3)
log warning and continue
// Recommended: log warning and continue, let user decide
}
// ... remaining code ...
}
}
```
并在文档中添加说明：
```markdown
**Important Notes:**
- If a runtime field has the same name as an existing field in the index,
the index field value takes precedence. Be careful to avoid naming
conflicts.
- To avoid conflicts, use a prefix like `runtime_` for your runtime fields.
```
**Rationale**:
1. Avoid users getting unexpected results due to field conflicts
2. Provide clear warnings to help users identify configuration issues
3. Documentation helps users understand the behavior
---
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
