This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13003-305f655b6903246ad8de9175e2b59d9ca2bc4a1b in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit eee6f703f5e689f17127f65b3eb0bdcb73b6180e Author: WPNG <[email protected]> AuthorDate: Wed Oct 28 16:44:25 2026 +0000 [Fix][Connector-V2] Add opt-in collision-safe Elasticsearch IDs (#13003) Co-authored-by: wpngwang <[email protected]> --- docs/en/connectors/sink/Elasticsearch.md | 7 + docs/zh/connectors/sink/Elasticsearch.md | 7 + .../config/ElasticsearchSinkOptions.java | 7 + .../IndexInfo.java => config/KeyEncoding.java} | 30 +---- .../seatunnel/elasticsearch/dto/IndexInfo.java | 3 + .../serialize/ElasticsearchRowSerializer.java | 5 +- .../elasticsearch/serialize/KeyExtractor.java | 28 +++- .../sink/ElasticsearchSinkFactory.java | 2 + .../elasticsearch/serialize/KeyExtractorTest.java | 150 +++++++++++++++++++++ .../connector/elasticsearch/ElasticsearchIT.java | 23 ++++ .../elasticsearch_composite_key_encoding.conf | 50 +++++++ 11 files changed, 283 insertions(+), 29 deletions(-) diff --git a/docs/en/connectors/sink/Elasticsearch.md b/docs/en/connectors/sink/Elasticsearch.md index a414dde0ab..93e96191d8 100644 --- a/docs/en/connectors/sink/Elasticsearch.md +++ b/docs/en/connectors/sink/Elasticsearch.md @@ -38,6 +38,7 @@ Engine Supported | index_type | string | no | - | Deprecated. Maps to Elasticsearch `_type` for clusters that still require it. Leave unset for modern clusters. | | primary_keys | list | no | - | Primary key fields used to generate the document `_id`. Required for CDC sources that produce update / delete events. | | key_delimiter | string | no | `_` | Delimiter joining composite keys into `_id` (default `_`). Use a different character to avoid clashes with field values. | +| key_encoding | enum | no | LEGACY | `LEGACY` keeps existing document IDs; `LENGTH_PREFIXED` avoids composite-key delimiter collisions. | | auth_type | string | no | basic | Authentication mode: `basic` (HTTP Basic with `username`/`password`) or `api_key` (Elasticsearch API key). | | username | string | no | - | Username for `basic` auth. | | password | string | no | - | Password for `basic` auth. | @@ -78,6 +79,12 @@ Primary key fields used to generate the document `_id`, this is cdc required opt Delimiter for composite keys ("_" by default), e.g., "$" would result in document `_id` "KEY1$KEY2$KEY3". +### key_encoding [enum] + +`LEGACY` (default) keeps existing document IDs. With multiple `primary_keys`, values containing `key_delimiter` can produce the same ID. Set `LENGTH_PREFIXED` to encode each formatted value as `<length>:<value>`; for example, `["1_1", "2"]` becomes `3:1_11:2` and `["1", "1_2"]` becomes `1:13:1_2`. This mode ignores `key_delimiter` and leaves single-key IDs unchanged. + +Changing this option for an existing composite-key index changes its document IDs. Reindex or migrate existing documents before switching; otherwise subsequent writes can create duplicates instead of updating the old IDs. + ### multi_table_sink_replica [int] The replica number of sink writers used for each table in a multi-table sink job. Keep the default value unless one table needs more sink writer parallelism. diff --git a/docs/zh/connectors/sink/Elasticsearch.md b/docs/zh/connectors/sink/Elasticsearch.md index bc38bccb95..d82a44b7dd 100644 --- a/docs/zh/connectors/sink/Elasticsearch.md +++ b/docs/zh/connectors/sink/Elasticsearch.md @@ -32,6 +32,7 @@ import ChangeLog from '../changelog/connector-elasticsearch.md'; | index_type | string | 否 | | | primary_keys | list | 否 | | | key_delimiter | string | 否 | `_` | +| key_encoding | enum | 否 | LEGACY | | auth_type | string | 否 | basic | | username | string | 否 | | | password | string | 否 | | @@ -72,6 +73,12 @@ import ChangeLog from '../changelog/connector-elasticsearch.md'; 设定复合键的分隔符（默认为 `_`），例如，如果使用 `$` 作为分隔符，那么文档的 `_id` 将呈现为 `KEY1$KEY2$KEY3` 的格式 +### key_encoding [enum] + +`LEGACY`（默认）保持现有文档 ID。当存在多个 `primary_keys` 时，如果主键值包含 `key_delimiter`，可能会生成相同的 ID。设为 `LENGTH_PREFIXED` 后，每个格式化后的值会编码为 `<长度>:<值>`；例如 `["1_1", "2"]` 会编码为 `3:1_11:2`，`["1", "1_2"]` 会编码为 `1:13:1_2`。该模式会忽略 `key_delimiter`，且单主键的 ID 保持不变。 + +对已有复合主键索引修改此选项会改变文档 ID，切换前需要重建索引或迁移已有文档，否则后续写入可能产生重复文档而不是更新旧 ID。 + ### multi_table_sink_replica [int] 多表写入时，每张表对应的 Sink Writer 副本数。通常保持默认值即可；只有单表写入压力较大、需要更多写入并行度时再调大。 diff --git a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSinkOptions.java b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSinkOptions.java index 0e8b0d92bf..2424c87f2e 100644 --- a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSinkOptions.java +++ b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/ElasticsearchSinkOptions.java @@ -51,6 +51,13 @@ public class ElasticsearchSinkOptions extends ElasticsearchBaseOptions { .withDescription( "Delimiter for composite keys (\"_\" by default), e.g., \"$\" would result in document `_id` \"KEY1$KEY2$KEY3\"."); + public static final Option<KeyEncoding> KEY_ENCODING = + Options.key("key_encoding") + .enumType(KeyEncoding.class) + .defaultValue(KeyEncoding.LEGACY) + .withDescription( + "LEGACY preserves existing document IDs; LENGTH_PREFIXED encodes each composite key value as `<length>:<value>` to avoid delimiter collisions. LENGTH_PREFIXED ignores `key_delimiter` and leaves single-key IDs unchanged."); + public static final Option<Integer> MAX_BATCH_SIZE = Options.key("max_batch_size") .intType() diff --git a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/IndexInfo.java b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/KeyEncoding.java similarity index 52% copy from seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/IndexInfo.java copy to seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/KeyEncoding.java index 5de741fc10..3311c6e3d7 100644 --- a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/IndexInfo.java +++ b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/KeyEncoding.java @@ -15,28 +15,12 @@ * limitations under the License. */ -package org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto; +package org.apache.seatunnel.connectors.seatunnel.elasticsearch.config; -import org.apache.seatunnel.api.configuration.ReadonlyConfig; -import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions; - -import lombok.Data; - -/** index config by seatunnel */ -@Data -public class IndexInfo { - - private String index; - private String type; - private String[] primaryKeys; - private String keyDelimiter; - - public IndexInfo(String index, ReadonlyConfig config) { - this.index = index; - type = config.get(ElasticsearchSinkOptions.INDEX_TYPE); - if (config.getOptional(ElasticsearchSinkOptions.PRIMARY_KEYS).isPresent()) { - primaryKeys = config.get(ElasticsearchSinkOptions.PRIMARY_KEYS).toArray(new String[0]); - } - keyDelimiter = config.get(ElasticsearchSinkOptions.KEY_DELIMITER); - } +/** Encoding of composite document IDs. Existing indexes use {@link #LEGACY}. */ +public enum KeyEncoding { + /** Join key fields with {@code key_delimiter} (legacy behavior). */ + LEGACY, + /** Encode each key field as {@code <length>:<value>} to avoid delimiter collisions. */ + LENGTH_PREFIXED } diff --git a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/IndexInfo.java b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/IndexInfo.java index 5de741fc10..2ee8f5d8c5 100644 --- a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/IndexInfo.java +++ b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/IndexInfo.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions; +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.KeyEncoding; import lombok.Data; @@ -30,6 +31,7 @@ public class IndexInfo { private String type; private String[] primaryKeys; private String keyDelimiter; + private KeyEncoding keyEncoding; public IndexInfo(String index, ReadonlyConfig config) { this.index = index; @@ -38,5 +40,6 @@ public class IndexInfo { primaryKeys = config.get(ElasticsearchSinkOptions.PRIMARY_KEYS).toArray(new String[0]); } keyDelimiter = config.get(ElasticsearchSinkOptions.KEY_DELIMITER); + keyEncoding = config.get(ElasticsearchSinkOptions.KEY_ENCODING); } } diff --git a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java index 5b97c12c31..34bb6be72d 100644 --- a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java +++ b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java @@ -79,7 +79,10 @@ public class ElasticsearchRowSerializer implements SeaTunnelRowSerializer { this.seaTunnelRowType = seaTunnelRowType; this.keyExtractor = KeyExtractor.createKeyExtractor( - seaTunnelRowType, indexInfo.getPrimaryKeys(), indexInfo.getKeyDelimiter()); + seaTunnelRowType, + indexInfo.getPrimaryKeys(), + indexInfo.getKeyDelimiter(), + indexInfo.getKeyEncoding()); this.vectorizationFields = vectorizationFields; this.vectorDimension = vectorDimension; } diff --git a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractor.java b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractor.java index 31149b43dc..ca587703c8 100644 --- a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractor.java +++ b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractor.java @@ -21,6 +21,7 @@ import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.KeyEncoding; import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorException; import lombok.AllArgsConstructor; @@ -37,22 +38,38 @@ import java.util.function.Function; public class KeyExtractor implements Function<SeaTunnelRow, String>, Serializable { private final FieldFormatter[] fieldFormatters; private final String keyDelimiter; + private final KeyEncoding keyEncoding; @Override public String apply(SeaTunnelRow row) { StringBuilder builder = new StringBuilder(); for (int i = 0; i < fieldFormatters.length; i++) { - if (i > 0) { - builder.append(keyDelimiter); - } String value = fieldFormatters[i].format(row); - builder.append(value); + // LENGTH_PREFIXED is only applied to composite keys (more than one field) so that + // single-key IDs stay stable. value.length() counts UTF-16 code units, which makes + // the encoded ID deterministic across JVMs. + if (keyEncoding == KeyEncoding.LENGTH_PREFIXED && fieldFormatters.length > 1) { + builder.append(value.length()).append(':').append(value); + } else { + if (i > 0) { + builder.append(keyDelimiter); + } + builder.append(value); + } } return builder.toString(); } public static Function<SeaTunnelRow, String> createKeyExtractor( SeaTunnelRowType rowType, String[] primaryKeys, String keyDelimiter) { + return createKeyExtractor(rowType, primaryKeys, keyDelimiter, KeyEncoding.LEGACY); + } + + public static Function<SeaTunnelRow, String> createKeyExtractor( + SeaTunnelRowType rowType, + String[] primaryKeys, + String keyDelimiter, + KeyEncoding keyEncoding) { if (primaryKeys == null) { return row -> null; } @@ -64,7 +81,8 @@ public class KeyExtractor implements Function<SeaTunnelRow, String>, Serializabl FieldFormatter fieldFormatter = createFieldFormatter(fieldIndex, fieldType); fieldFormatters.add(fieldFormatter); } - return new KeyExtractor(fieldFormatters.toArray(new FieldFormatter[0]), keyDelimiter); + return new KeyExtractor( + fieldFormatters.toArray(new FieldFormatter[0]), keyDelimiter, keyEncoding); } private static FieldFormatter createFieldFormatter( 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 0bbab15e35..9aa228ff62 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 @@ -49,6 +49,7 @@ import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.Ela import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions.INDEX; import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions.INDEX_TYPE; import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions.KEY_DELIMITER; +import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions.KEY_ENCODING; import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions.MAX_BATCH_SIZE; import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions.MAX_RETRY_COUNT; import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions.PRIMARY_KEYS; @@ -74,6 +75,7 @@ public class ElasticsearchSinkFactory implements TableSinkFactory { INDEX_TYPE, PRIMARY_KEYS, KEY_DELIMITER, + KEY_ENCODING, USERNAME, PASSWORD, MAX_RETRY_COUNT, diff --git a/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractorTest.java b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractorTest.java new file mode 100644 index 0000000000..73b6299a0b --- /dev/null +++ b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/KeyExtractorTest.java @@ -0,0 +1,150 @@ +/* + * 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.serialize; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.ElasticsearchSinkOptions; +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.KeyEncoding; +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.ElasticsearchClusterInfo; +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.IndexInfo; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.function.Function; + +class KeyExtractorTest { + private static final SeaTunnelRowType ROW_TYPE = + new SeaTunnelRowType( + new String[] {"first", "second"}, + new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE, BasicType.STRING_TYPE}); + + @Test + void lengthPrefixedIdsSeparateDelimiterCollisions() { + Function<SeaTunnelRow, String> legacy = + KeyExtractor.createKeyExtractor(ROW_TYPE, new String[] {"first", "second"}, "_"); + Function<SeaTunnelRow, String> safe = + KeyExtractor.createKeyExtractor( + ROW_TYPE, + new String[] {"first", "second"}, + "_", + KeyEncoding.LENGTH_PREFIXED); + SeaTunnelRow left = new SeaTunnelRow(new Object[] {"1_1", "2"}); + SeaTunnelRow right = new SeaTunnelRow(new Object[] {"1", "1_2"}); + + Assertions.assertEquals("1_1_2", legacy.apply(left)); + Assertions.assertEquals(legacy.apply(left), legacy.apply(right)); + Assertions.assertEquals("3:1_11:2", safe.apply(left)); + Assertions.assertEquals("1:13:1_2", safe.apply(right)); + } + + @Test + void oneKeyKeepsItsExistingIdInBothModes() { + SeaTunnelRow row = new SeaTunnelRow(new Object[] {"1_1", "2"}); + Assertions.assertEquals( + "1_1", + KeyExtractor.createKeyExtractor( + ROW_TYPE, new String[] {"first"}, "_", KeyEncoding.LENGTH_PREFIXED) + .apply(row)); + } + + @Test + void lengthPrefixedIdsHandleThreeKeys() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"first", "second", "third"}, + new SeaTunnelDataType<?>[] { + BasicType.STRING_TYPE, BasicType.STRING_TYPE, BasicType.STRING_TYPE + }); + Function<SeaTunnelRow, String> safe = + KeyExtractor.createKeyExtractor( + rowType, + new String[] {"first", "second", "third"}, + "_", + KeyEncoding.LENGTH_PREFIXED); + Assertions.assertEquals( + "1:a1:b1:c", safe.apply(new SeaTunnelRow(new Object[] {"a", "b", "c"}))); + } + + @Test + void lengthPrefixedIdsHandleEmptyStringValues() { + Function<SeaTunnelRow, String> safe = + KeyExtractor.createKeyExtractor( + ROW_TYPE, + new String[] {"first", "second"}, + "_", + KeyEncoding.LENGTH_PREFIXED); + Assertions.assertEquals("0:1:b", safe.apply(new SeaTunnelRow(new Object[] {"", "b"}))); + } + + @Test + void lengthPrefixedIdsHandleNonStringKeys() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "name"}, + new SeaTunnelDataType<?>[] {BasicType.INT_TYPE, BasicType.STRING_TYPE}); + Function<SeaTunnelRow, String> safe = + KeyExtractor.createKeyExtractor( + rowType, new String[] {"id", "name"}, "_", KeyEncoding.LENGTH_PREFIXED); + Assertions.assertEquals("1:11:2", safe.apply(new SeaTunnelRow(new Object[] {1, "2"}))); + } + + @Test + void nullValuesFailConsistentlyInBothModes() { + SeaTunnelRow row = new SeaTunnelRow(new Object[] {"1_1", null}); + Function<SeaTunnelRow, String> legacy = + KeyExtractor.createKeyExtractor(ROW_TYPE, new String[] {"first", "second"}, "_"); + Function<SeaTunnelRow, String> safe = + KeyExtractor.createKeyExtractor( + ROW_TYPE, + new String[] {"first", "second"}, + "_", + KeyEncoding.LENGTH_PREFIXED); + Assertions.assertThrows(NullPointerException.class, () -> legacy.apply(row)); + Assertions.assertThrows(NullPointerException.class, () -> safe.apply(row)); + } + + @Test + void serializerUsesTheSameSafeIdForUpsertAndDelete() { + Map<String, Object> options = new HashMap<>(); + options.put(ElasticsearchSinkOptions.INDEX.key(), "id_probe"); + options.put(ElasticsearchSinkOptions.PRIMARY_KEYS.key(), Arrays.asList("first", "second")); + options.put(ElasticsearchSinkOptions.KEY_ENCODING.key(), "LENGTH_PREFIXED"); + ElasticsearchClusterInfo clusterInfo = + ElasticsearchClusterInfo.builder().clusterVersion("8.0.0").build(); + ElasticsearchRowSerializer serializer = + new ElasticsearchRowSerializer( + clusterInfo, + new IndexInfo("id_probe", ReadonlyConfig.fromMap(options)), + ROW_TYPE); + SeaTunnelRow row = new SeaTunnelRow(new Object[] {"1_1", "2"}); + + row.setRowKind(RowKind.INSERT); + Assertions.assertTrue(serializer.serializeRow(row).contains("3:1_11:2")); + row.setRowKind(RowKind.DELETE); + Assertions.assertTrue(serializer.serializeRow(row).contains("3:1_11:2")); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java index 39c5cb3997..a3a0380c3b 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java @@ -726,6 +726,29 @@ public class ElasticsearchIT extends TestSuiteBase implements TestResource { }); } + @TestTemplate + public void testCompositeKeyEncodingAvoidsDocumentIdCollisions(TestContainer container) + throws Exception { + Container.ExecResult result = + container.executeJob("/elasticsearch/elasticsearch_composite_key_encoding.conf"); + Assertions.assertEquals(0, result.getExitCode()); + + // Two rows whose legacy delimiter-joined IDs would collide ("1_1_2") must survive as two + // distinct documents when LENGTH_PREFIXED encoding is enabled. + Awaitility.await() + .atMost(INDEX_REFRESH_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .ignoreExceptions() + .untilAsserted( + () -> + Assertions.assertEquals( + 2L, + esRestClient + .getIndexDocsCount("st_id_encoding") + .get(0) + .getDocsCount())); + } + @TestTemplate public void testElasticsearchWithFullType(TestContainer container) throws IOException, InterruptedException { diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_composite_key_encoding.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_composite_key_encoding.conf new file mode 100644 index 0000000000..bba39e5e56 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_composite_key_encoding.conf @@ -0,0 +1,50 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema { + fields { + first = string + second = string + marker = string + } + } + rows = [ + { kind = INSERT, fields = ["1_1", "2", "left"] }, + { kind = INSERT, fields = ["1", "1_2", "right"] } + ] + } +} + +sink { + Elasticsearch { + hosts = ["https://elasticsearch:9200";] + username = "elastic" + password = "elasticsearch" + tls_verify_certificate = false + tls_verify_hostname = false + index = "st_id_encoding" + primary_keys = ["first", "second"] + key_encoding = "LENGTH_PREFIXED" + } +}
