[Feature][Connector-V2][Elasticsearch] Add Runtime Fields support (Elasticsearch 7.11+)

[CosmosNi]

• add runtime_fields option to the Elasticsearch source to build runtime_mappings and include computed fields in scroll/PIT queries
• document runtime field usage (EN/zh) with configuration details, examples, and performance notes
• add e2e coverage and sample configs for runtime fields to validate query-time computations

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[chl-wxp]

Why modify this line?

[CosmosNi]

This configuration is inconsistent with the actual configuration.

[chl-wxp]

Got it, it seems that the file docs/en/connector-v2/source/Elasticsearch.md also has this problem, please fix it as well.

[chl-wxp]

Is the docker image version es7 more reasonable, because this feature was introduced in version 7.11+?
If you must use version 8.9.0, I think there is no need to add the ElasticsearchRuntimeFieldsIT class. It is also possible to write e2e in ElasticsearchIT.

[CosmosNi]

Get

[LiJie20190102]

thx, please modify tls_verify_hostnames to tls_verify_hostname

[CosmosNi]

get

[LiJie20190102]

LGTM

[chl-wxp]

If runtime_field is determined, there are only three attributes: name, type, and script. I suggest creating a new class RuntimeField. For example:

+ public class RuntimeField {
+    private String name;
+    private String type;
+    private String script;
+}

[CosmosNi]

The required core fields for runtime_field are indeed name, type, and script. They form the minimal valid configuration. However, the current implementation also supports optional fields such as script_lang and script_params, which are added into the script block. So it’s not limited to only those three.

[chl-wxp]

delete

[chl-wxp]

@zhangshenghang cc

[CosmosNi]

@zhangshenghang PTAL

[chl-wxp]

if ci passes

[corgy-w]

+1

[DanielCarter-stack]

Issue 1: Missing Elasticsearch version compatibility check

Location: ElasticsearchSource.java:188-194

// 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:

// 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

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:

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

@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:

@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:

### 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):

{
  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 ...
    }
}

并在文档中添加说明：

**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

---

[davidzollo]

+1
LGTM