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 5bdeda5fde [Fix][Connector-V2] Support schema evolution in Redis sink
(#11646)
5bdeda5fde is described below
commit 5bdeda5fde41c51b5e719f94f5eb8df494c4576f
Author: limin <[email protected]>
AuthorDate: Sun Aug 23 09:34:57 2026 +0800
[Fix][Connector-V2] Support schema evolution in Redis sink (#11646)
Co-authored-by: limin <[email protected]>
---
docs/en/connectors/sink/Redis.md | 19 +
.../introduction/configuration/schema-evolution.md | 1 +
docs/zh/connectors/sink/Redis.md | 16 +
.../introduction/configuration/schema-evolution.md | 1 +
.../connectors/seatunnel/redis/sink/RedisSink.java | 63 ++-
.../seatunnel/redis/sink/RedisSinkWriter.java | 81 +++-
.../seatunnel/redis/RedisFactoryTest.java | 17 +
.../seatunnel/redis/sink/RedisSinkTest.java | 212 ++++++++++
.../seatunnel/redis/sink/RedisSinkWriterTest.java | 154 ++++++++
.../connector-redis-e2e/pom.xml | 26 ++
.../e2e/connector/redis/RedisSchemaChangeIT.java | 438 +++++++++++++++++++++
.../mysqlcdc_to_redis_with_schema_change.conf | 46 +++
12 files changed, 1063 insertions(+), 11 deletions(-)
diff --git a/docs/en/connectors/sink/Redis.md b/docs/en/connectors/sink/Redis.md
index ea0b18d739..eaa81845b0 100644
--- a/docs/en/connectors/sink/Redis.md
+++ b/docs/en/connectors/sink/Redis.md
@@ -96,6 +96,25 @@ Replica count for multi-table sink writers. It applies when
upstream rows carry
For multi-table jobs, `key` may include `${table_name}` so rows from different
upstream tables are written to separate
Redis keys, for example `key = "redis-result-${table_name}"`.
+## Schema Evolution
+
+Redis Sink supports schema evolution with SeaTunnel Zeta. When the upstream is
a CDC source, enable
+`schema-changes.enabled = true` in the source configuration so schema change
events are sent to the sink.
+
+Redis is schema-less, so schema evolution does not execute DDL in Redis.
Instead, when Redis Sink serializes the whole
+upstream row as JSON or TEXT, it refreshes the serializer after a supported
schema change event. Newly added fields are
+included, and dropped fields are no longer written. See
+[Schema Evolution](../../introduction/configuration/schema-evolution.md) for
the supported event types.
+
+Schema evolution does not rewrite field names configured in `key`, custom key
placeholders, `value_field`,
+`hash_key_field`, or `hash_value_field`. Do not rename or drop a field
referenced by these options while the job is
+running. If a configured field no longer exists, Redis Sink applies the
missing-field behavior described in
+[Write Rules](#write-rules), which can turn the configured field name into a
literal key or value.
+
+Before applying a schema change, Redis Sink flushes rows buffered with the
previous schema. It also stores the latest
+schema in checkpoint state and restores that schema after recovery. Restoring
a job from a checkpoint taken after a DDL
+while increasing the Redis sink parallelism is not currently supported.
+
## Examples
### Write Rows To A Redis List
diff --git a/docs/en/introduction/configuration/schema-evolution.md
b/docs/en/introduction/configuration/schema-evolution.md
index 70af86ea2d..4cd6612059 100644
--- a/docs/en/introduction/configuration/schema-evolution.md
+++ b/docs/en/introduction/configuration/schema-evolution.md
@@ -28,6 +28,7 @@ Schema Evolution means that the schema of a data table can be
changed and the da
[Doris](../../connectors/sink/Doris.md)
[Paimon](../../connectors/sink/Paimon.md#schema-evolution)
[Elasticsearch](../../connectors/sink/Elasticsearch.md#schema-evolution)
+[Redis](../../connectors/sink/Redis.md#schema-evolution)
Note:
* The schema evolution is not support the transform at now. The schema
evolution of different types of databases（Oracle-CDC -> Jdbc-Mysql）is currently
not supported the default value of the column in ddl.
diff --git a/docs/zh/connectors/sink/Redis.md b/docs/zh/connectors/sink/Redis.md
index d08cd96d95..c430f69bb6 100644
--- a/docs/zh/connectors/sink/Redis.md
+++ b/docs/zh/connectors/sink/Redis.md
@@ -93,6 +93,22 @@ Redis 接收器连接器可以在批处理或流处理作业中把上游数据
多表作业中，`key` 可以包含 `${table_name}`，这样不同上游表的数据会写入不同 Redis key，例如
`key = "redis-result-${table_name}"`。
+## 模式演变
+
+Redis Sink 在 SeaTunnel Zeta 引擎中支持模式演变。上游使用 CDC Source 时，需要在 Source 配置中设置
+`schema-changes.enabled = true`，使模式变更事件能够发送到 Sink。
+
+Redis 本身没有表结构，因此模式演变不会在 Redis 中执行 DDL。当 Redis Sink 使用 JSON 或 TEXT
+序列化完整上游行时，它会在收到支持的模式变更事件后更新序列化器：新增字段会写入 Redis，已删除字段不再写入。
+支持的事件类型请参见[模式演变](../../introduction/configuration/schema-evolution.md)。
+
+模式演变不会自动改写 `key`、自定义 Key 占位符、`value_field`、`hash_key_field` 或
`hash_value_field` 中配置的
+字段名。作业运行期间，请勿重命名或删除这些选项引用的字段。如果配置的字段已不存在，Redis Sink 会按
+[写入规则](#写入规则)中的缺失字段行为处理，配置的字段名可能会被当作字面量 Key 或 Value 写入。
+
+应用模式变更前，Redis Sink 会先刷新按旧模式缓存的数据。它还会把最新模式保存到 checkpoint 状态，并在恢复时使用
+该模式。当前不支持从包含 DDL 后状态的 checkpoint 恢复作业时增加 Redis Sink 的并行度。
+
## 示例
### 写入 Redis List
diff --git a/docs/zh/introduction/configuration/schema-evolution.md
b/docs/zh/introduction/configuration/schema-evolution.md
index 7dd6cb8ee0..359d7c4991 100644
--- a/docs/zh/introduction/configuration/schema-evolution.md
+++ b/docs/zh/introduction/configuration/schema-evolution.md
@@ -28,6 +28,7 @@
[Doris](../../connectors/sink/Doris.md)
[Paimon](../../connectors/sink/Paimon.md#模式演变)
[Elasticsearch](../../connectors/sink/Elasticsearch.md#模式演变)
+[Redis](../../connectors/sink/Redis.md#模式演变)
注意:
* 目前模式演进不支持transform。不同类型数据库(Oracle-CDC -> Jdbc-Mysql)的模式演进目前不支持ddl中列的默认值。
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java
index 67774821ed..215be889ca 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java
@@ -18,22 +18,28 @@
package org.apache.seatunnel.connectors.seatunnel.redis.sink;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.serialization.DefaultSerializer;
+import org.apache.seatunnel.api.serialization.Serializer;
import org.apache.seatunnel.api.sink.SinkWriter;
import org.apache.seatunnel.api.sink.SupportMultiTableSink;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSink;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.SchemaChangeType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisBaseOptions;
import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisParameters;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
import java.util.Optional;
-public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, Void>
- implements SupportMultiTableSink {
+public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, TableSchema>
+ implements SupportMultiTableSink, SupportSchemaEvolutionSink {
private final RedisParameters redisParameters = new RedisParameters();
- private final SeaTunnelRowType seaTunnelRowType;
+ private final TableSchema tableSchema;
private final ReadonlyConfig readonlyConfig;
private final CatalogTable catalogTable;
@@ -41,7 +47,7 @@ public class RedisSink extends
AbstractSimpleSink<SeaTunnelRow, Void>
this.readonlyConfig = config;
this.catalogTable = table;
this.redisParameters.buildWithConfig(config);
- this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
+ this.tableSchema = catalogTable.getTableSchema();
}
@Override
@@ -51,11 +57,56 @@ public class RedisSink extends
AbstractSimpleSink<SeaTunnelRow, Void>
@Override
public RedisSinkWriter createWriter(SinkWriter.Context context) throws
IOException {
- return new RedisSinkWriter(seaTunnelRowType, redisParameters);
+ return new RedisSinkWriter(tableSchema, redisParameters);
+ }
+
+ /**
+ * Restores the latest writer schema. Rescaling can provide state from
multiple writers, and
+ * every state must contain the same schema because selecting the widest
schema would restore a
+ * stale definition after a drop-column event.
+ */
+ @Override
+ public RedisSinkWriter restoreWriter(SinkWriter.Context context,
List<TableSchema> states)
+ throws IOException {
+ if (states == null || states.isEmpty()) {
+ return createWriter(context);
+ }
+ TableSchema restoredSchema = states.get(0);
+ for (int stateIndex = 1; stateIndex < states.size(); stateIndex++) {
+ TableSchema state = states.get(stateIndex);
+ if (!restoredSchema.equals(state)) {
+ throw new IOException(
+ String.format(
+ "Redis sink cannot restore writer for table %s
because state 0 fields %s differ from state %d fields %s",
+ catalogTable.getTablePath().getFullName(),
+ schemaFields(restoredSchema),
+ stateIndex,
+ schemaFields(state)));
+ }
+ }
+ return new RedisSinkWriter(restoredSchema, redisParameters);
+ }
+
+ @Override
+ public Optional<Serializer<TableSchema>> getWriterStateSerializer() {
+ return Optional.of(new DefaultSerializer<>());
}
@Override
public Optional<CatalogTable> getWriteCatalogTable() {
return Optional.ofNullable(catalogTable);
}
+
+ @Override
+ public List<SchemaChangeType> supports() {
+ return Arrays.asList(
+ SchemaChangeType.ADD_COLUMN,
+ SchemaChangeType.DROP_COLUMN,
+ SchemaChangeType.RENAME_COLUMN,
+ SchemaChangeType.UPDATE_COLUMN);
+ }
+
+ private static String schemaFields(TableSchema schema) {
+ return Arrays.toString(schema.toPhysicalRowDataType().getFieldNames());
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java
index 21b3233620..f4c5755c5f 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java
@@ -22,7 +22,13 @@ import
org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.serialization.SerializationSchema;
import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSinkWriter;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import
org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher;
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.common.constants.PluginType;
@@ -43,6 +49,7 @@ import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -51,15 +58,25 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
@Slf4j
-public class RedisSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
- implements SupportMultiTableSinkWriter<Void> {
+public class RedisSinkWriter extends AbstractSinkWriter<SeaTunnelRow,
TableSchema>
+ implements SupportMultiTableSinkWriter<Void>,
SupportSchemaEvolutionSinkWriter {
private static final Pattern LEGACY_PLACEHOLDER_PATTERN =
Pattern.compile("(?<!\\$)\\{([^{}]+)\\}");
private static final Pattern PLACEHOLDER_PATTERN =
Pattern.compile("\\$\\{([^}]+)\\}");
- private final SeaTunnelRowType seaTunnelRowType;
+
+ /** Authoritative schema saved in checkpoints and updated by ordered
schema-change events. */
+ private TableSchema tableSchema;
+
+ /** Physical row type derived from {@link #tableSchema} for key and value
field lookup. */
+ private SeaTunnelRowType seaTunnelRowType;
+
private final RedisParameters redisParameters;
- private final SerializationSchema serializationSchema;
+
+ /** Serializer derived from {@link #tableSchema} and refreshed together
with the row type. */
+ private SerializationSchema serializationSchema;
+
private final RedisClient redisClient;
+ private final TableSchemaChangeEventDispatcher schemaChangeEventDispatcher;
private final int batchSize;
@@ -67,11 +84,25 @@ public class RedisSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
private final List<String> keyBuffer;
private final List<String> valueBuffer;
+ /**
+ * Creates a writer from a physical row type.
+ *
+ * @deprecated Use {@link #RedisSinkWriter(TableSchema, RedisParameters)}
so schema evolution
+ * and checkpoint recovery retain the complete table schema.
+ */
+ @Deprecated
public RedisSinkWriter(SeaTunnelRowType seaTunnelRowType, RedisParameters
redisParameters) {
- this.seaTunnelRowType = seaTunnelRowType;
+ this(toTableSchema(seaTunnelRowType), redisParameters);
+ }
+
+ /** Creates a writer whose schema can be updated and restored from
checkpoint state. */
+ public RedisSinkWriter(TableSchema tableSchema, RedisParameters
redisParameters) {
+ this.tableSchema = normalizeTableSchema(tableSchema);
+ this.seaTunnelRowType = this.tableSchema.toPhysicalRowDataType();
this.redisParameters = redisParameters;
this.serializationSchema = createSerializationSchema(redisParameters,
seaTunnelRowType);
this.redisClient = redisParameters.buildRedisClient();
+ this.schemaChangeEventDispatcher = new
TableSchemaChangeEventDispatcher();
this.batchSize = redisParameters.getBatchSize();
this.rowKinds = new ArrayList<>(batchSize);
this.keyBuffer = new ArrayList<>(batchSize);
@@ -93,6 +124,40 @@ public class RedisSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
log.debug("write redis key: {}, value: {}， rowKind: {}", key, value,
element.getRowKind());
}
+ /**
+ * Flushes all rows serialized with the previous schema before refreshing
the writer's
+ * schema-derived views. The engine orders this callback with row
processing, so all three views
+ * are updated before any subsequent row is processed.
+ */
+ @Override
+ public void applySchemaChange(SchemaChangeEvent event) {
+ flush();
+ tableSchema =
schemaChangeEventDispatcher.reset(tableSchema).apply(event);
+ seaTunnelRowType = tableSchema.toPhysicalRowDataType();
+ serializationSchema = createSerializationSchema(redisParameters,
seaTunnelRowType);
+ }
+
+ private static TableSchema normalizeTableSchema(TableSchema tableSchema) {
+ if (tableSchema.getConstraintKeys() != null) {
+ return tableSchema;
+ }
+ return TableSchema.builder()
+ .columns(tableSchema.getColumns())
+ .primaryKey(tableSchema.getPrimaryKey())
+ .build();
+ }
+
+ private static TableSchema toTableSchema(SeaTunnelRowType
seaTunnelRowType) {
+ TableSchema.Builder schemaBuilder = TableSchema.builder();
+ String[] fieldNames = seaTunnelRowType.getFieldNames();
+ SeaTunnelDataType<?>[] fieldTypes = seaTunnelRowType.getFieldTypes();
+ for (int i = 0; i < fieldNames.length; i++) {
+ schemaBuilder.column(
+ PhysicalColumn.of(fieldNames[i], fieldTypes[i], 0L, true,
null, null));
+ }
+ return schemaBuilder.build();
+ }
+
private String getKey(SeaTunnelRow element, List<String> fields) {
String key = redisParameters.getKeyField();
Boolean supportCustomKey = redisParameters.getSupportCustomKey();
@@ -291,6 +356,12 @@ public class RedisSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
return Optional.empty();
}
+ /** Returns an isolated copy of the latest schema for checkpoint and
rescale recovery. */
+ @Override
+ public List<TableSchema> snapshotState(long checkpointId) {
+ return Collections.singletonList(tableSchema.copy());
+ }
+
private synchronized void flush() {
if (!keyBuffer.isEmpty()) {
doBatchWrite();
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
index cf2ca61d79..67fb4a8989 100644
---
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
@@ -21,6 +21,7 @@ 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.apache.seatunnel.api.serialization.Serializer;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
@@ -28,6 +29,7 @@ import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSink;
import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.redis.source.RedisSourceFactory;
@@ -37,6 +39,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -144,6 +147,20 @@ class RedisFactoryTest {
() -> new RedisSinkFactory().createSink(context).createSink());
}
+ @Test
+ void sinkExposesSchemaWriterStateSerializer() throws IOException {
+ RedisSink sink = new
RedisSink(ReadonlyConfig.fromMap(singleSinkConfig()), catalogTable());
+
+ Assertions.assertTrue(sink.getWriterStateSerializer().isPresent());
+ @SuppressWarnings("unchecked")
+ Serializer<TableSchema> serializer =
+ (Serializer<TableSchema>) (Serializer<?>)
sink.getWriterStateSerializer().get();
+ TableSchema tableSchema = catalogTable().getTableSchema();
+
+ Assertions.assertEquals(
+ tableSchema,
serializer.deserialize(serializer.serialize(tableSchema)));
+ }
+
// parameterized-case providers
static Stream<Arguments> invalidSingleConnections() {
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkTest.java
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkTest.java
new file mode 100644
index 0000000000..e387639cb1
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkTest.java
@@ -0,0 +1,212 @@
+/*
+ * 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.redis.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.serialization.Serializer;
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSink;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.SchemaChangeType;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.RowKind;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.seatunnel.redis.client.RedisClient;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisBaseOptions;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisDataType;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisParameters;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.when;
+
+class RedisSinkTest {
+
+ private static final TableIdentifier TABLE_IDENTIFIER =
+ TableIdentifier.of("catalog", "shop", null, "schema_events");
+
+ private RedisClient redisClient;
+ private RedisParameters redisParameters;
+ private RedisSink redisSink;
+
+ @BeforeEach
+ void setUp() throws ReflectiveOperationException {
+ redisClient = Mockito.mock(RedisClient.class);
+ redisParameters = Mockito.mock(RedisParameters.class);
+ when(redisParameters.buildRedisClient()).thenReturn(redisClient);
+ when(redisParameters.getBatchSize()).thenReturn(1);
+
when(redisParameters.getFormat()).thenReturn(RedisBaseOptions.Format.JSON);
+ when(redisParameters.getFieldDelimiter()).thenReturn(",");
+ when(redisParameters.getKeyField()).thenReturn("schema-change:${id}");
+ when(redisParameters.getSupportCustomKey()).thenReturn(true);
+ when(redisParameters.getRedisDataType()).thenReturn(RedisDataType.KEY);
+ when(redisParameters.getExpire()).thenReturn(0L);
+
+ redisSink =
+ new RedisSink(ReadonlyConfig.fromMap(sinkConfig()),
catalogTable(initialSchema()));
+ Field redisParametersField =
RedisSink.class.getDeclaredField("redisParameters");
+ redisParametersField.setAccessible(true);
+ redisParametersField.set(redisSink, redisParameters);
+ }
+
+ @Test
+ void testAdvertisesSupportedSchemaChanges() {
+ Assertions.assertInstanceOf(SupportSchemaEvolutionSink.class,
redisSink);
+ SupportSchemaEvolutionSink schemaEvolutionSink =
(SupportSchemaEvolutionSink) redisSink;
+
+ Assertions.assertEquals(
+ Arrays.asList(
+ SchemaChangeType.ADD_COLUMN,
+ SchemaChangeType.DROP_COLUMN,
+ SchemaChangeType.RENAME_COLUMN,
+ SchemaChangeType.UPDATE_COLUMN),
+ schemaEvolutionSink.supports());
+ }
+
+ @Test
+ void testRestoreWriterUsesConfiguredSchemaWhenStatesAreNull() throws
IOException {
+ RedisSinkWriter writer = redisSink.restoreWriter(context(), null);
+
+ Assertions.assertEquals(initialSchema(),
writer.snapshotState(1L).get(0));
+ }
+
+ @Test
+ void testRestoreWriterUsesConfiguredSchemaWhenStatesAreEmpty() throws
IOException {
+ RedisSinkWriter writer = redisSink.restoreWriter(context(),
Collections.emptyList());
+
+ Assertions.assertEquals(initialSchema(),
writer.snapshotState(1L).get(0));
+ }
+
+ @Test
+ void testRestoreWriterUsesSingleState() throws IOException {
+ TableSchema restoredSchema = evolvedSchema();
+
+ RedisSinkWriter writer =
+ redisSink.restoreWriter(context(),
Collections.singletonList(restoredSchema));
+
+ Assertions.assertEquals(restoredSchema,
writer.snapshotState(1L).get(0));
+ }
+
+ @Test
+ void testRestoreWriterAcceptsIdenticalStates() throws IOException {
+ TableSchema restoredSchema = evolvedSchema();
+
+ RedisSinkWriter writer =
+ redisSink.restoreWriter(
+ context(), Arrays.asList(restoredSchema,
restoredSchema.copy()));
+
+ Assertions.assertEquals(restoredSchema,
writer.snapshotState(1L).get(0));
+ }
+
+ @Test
+ void testRestoreWriterRejectsConflictingStatesWithTableAndFieldContext() {
+ IOException exception =
+ Assertions.assertThrows(
+ IOException.class,
+ () ->
+ redisSink.restoreWriter(
+ context(),
+ Arrays.asList(initialSchema(),
evolvedSchema())));
+
+ Assertions.assertTrue(
+
exception.getMessage().contains(TABLE_IDENTIFIER.toTablePath().getFullName()),
+ exception.getMessage());
+ Assertions.assertTrue(exception.getMessage().contains("state 1"),
exception.getMessage());
+ Assertions.assertTrue(
+ exception.getMessage().contains("[id, name]"),
exception.getMessage());
+ Assertions.assertTrue(
+ exception.getMessage().contains("[id, name, email]"),
exception.getMessage());
+ }
+
+ @Test
+ void testSerializedStateRestoresWriterWithLatestSchema() throws
IOException {
+ List<List<String>> writtenValues = new ArrayList<>();
+ Mockito.doAnswer(
+ invocation -> {
+ List<String> values = invocation.getArgument(2);
+ writtenValues.add(new ArrayList<>(values));
+ return null;
+ })
+ .when(redisClient)
+ .batchWriteString(
+ Mockito.anyList(), Mockito.anyList(),
Mockito.anyList(), Mockito.anyLong());
+ Serializer<TableSchema> serializer =
+
redisSink.getWriterStateSerializer().orElseThrow(AssertionError::new);
+ TableSchema restoredSchema =
serializer.deserialize(serializer.serialize(evolvedSchema()));
+ RedisSinkWriter writer =
+ redisSink.restoreWriter(context(),
Collections.singletonList(restoredSchema));
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {1L, "Alice",
"[email protected]"});
+ row.setRowKind(RowKind.INSERT);
+
+ writer.write(row);
+
+ Assertions.assertEquals(1, writtenValues.size());
+ Assertions.assertEquals(
+
"{\"id\":1,\"name\":\"Alice\",\"email\":\"[email protected]\"}",
+ writtenValues.get(0).get(0));
+ }
+
+ private static SinkWriter.Context context() {
+ return Mockito.mock(SinkWriter.Context.class);
+ }
+
+ private static CatalogTable catalogTable(TableSchema schema) {
+ return CatalogTable.of(
+ TABLE_IDENTIFIER, schema, new HashMap<>(), new ArrayList<>(),
null, "catalog");
+ }
+
+ private static TableSchema initialSchema() {
+ return TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.LONG_TYPE, 0L,
false, null, null))
+ .column(PhysicalColumn.of("name", BasicType.STRING_TYPE, 64L,
true, null, null))
+ .build();
+ }
+
+ private static TableSchema evolvedSchema() {
+ return TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.LONG_TYPE, 0L,
false, null, null))
+ .column(PhysicalColumn.of("name", BasicType.STRING_TYPE, 64L,
true, null, null))
+ .column(PhysicalColumn.of("email", BasicType.STRING_TYPE,
128L, true, null, null))
+ .build();
+ }
+
+ private static Map<String, Object> sinkConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("mode", "SINGLE");
+ config.put("host", "localhost");
+ config.put("port", 6379);
+ config.put("key", "schema-change:${id}");
+ config.put("data_type", "KEY");
+ return config;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriterTest.java
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriterTest.java
index a7adcc928f..9acd43d9d6 100644
---
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriterTest.java
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriterTest.java
@@ -17,6 +17,12 @@
package org.apache.seatunnel.connectors.seatunnel.redis.sink;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSinkWriter;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.RowKind;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
@@ -32,12 +38,18 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
import static org.mockito.Mockito.when;
public class RedisSinkWriterTest {
+ private static final TableIdentifier TABLE_IDENTIFIER =
+ TableIdentifier.of("catalog", "shop", null, "schema_events");
+
private RedisClient mockRedisClient;
private RedisParameters mockRedisParameters;
@@ -157,4 +169,146 @@ public class RedisSinkWriterTest {
Assertions.assertEquals("user:{1}:profile", customKey);
}
+
+ @Test
+ void testAddColumnUsesLatestSchemaForJsonSerialization() throws
IOException {
+ List<List<String>> writtenValues = captureStringWrites();
+ configureJsonKeySink(1);
+ redisSinkWriter = new RedisSinkWriter(initialSchema(),
mockRedisParameters);
+
+ schemaEvolutionWriter().applySchemaChange(addEmailEvent());
+ redisSinkWriter.write(insertRow(1L, "Alice", "[email protected]",
"legacy-value"));
+
+ Assertions.assertEquals(1, writtenValues.size());
+ Assertions.assertEquals(
+
"{\"id\":1,\"name\":\"Alice\",\"email\":\"[email protected]\","
+ + "\"legacy_note\":\"legacy-value\"}",
+ writtenValues.get(0).get(0));
+ }
+
+ @Test
+ void testDropColumnRemovesFieldFromJsonSerialization() throws IOException {
+ List<List<String>> writtenValues = captureStringWrites();
+ configureJsonKeySink(1);
+ redisSinkWriter = new RedisSinkWriter(initialSchema(),
mockRedisParameters);
+
+ SupportSchemaEvolutionSinkWriter schemaEvolutionWriter =
schemaEvolutionWriter();
+ schemaEvolutionWriter.applySchemaChange(addEmailEvent());
+ schemaEvolutionWriter.applySchemaChange(
+ new AlterTableDropColumnEvent(TABLE_IDENTIFIER,
"legacy_note"));
+ redisSinkWriter.write(insertRow(1L, "Alice", "[email protected]"));
+
+ Assertions.assertEquals(1, writtenValues.size());
+ Assertions.assertEquals(
+
"{\"id\":1,\"name\":\"Alice\",\"email\":\"[email protected]\"}",
+ writtenValues.get(0).get(0));
+ }
+
+ @Test
+ void testSchemaChangeFlushesOldRowsBeforeSerializerRefresh() throws
IOException {
+ List<List<String>> writtenValues = captureStringWrites();
+ configureJsonKeySink(2);
+ redisSinkWriter = new RedisSinkWriter(initialSchema(),
mockRedisParameters);
+
+ redisSinkWriter.write(insertRow(1L, "Before", "old-schema-value"));
+ schemaEvolutionWriter().applySchemaChange(addEmailEvent());
+
+ Assertions.assertEquals(1, writtenValues.size());
+ Assertions.assertEquals(
+
"{\"id\":1,\"name\":\"Before\",\"legacy_note\":\"old-schema-value\"}",
+ writtenValues.get(0).get(0));
+
+ redisSinkWriter.write(insertRow(2L, "After", "[email protected]",
"new-schema-value"));
+ redisSinkWriter.prepareCommit();
+
+ Assertions.assertEquals(2, writtenValues.size());
+ Assertions.assertEquals(
+
"{\"id\":2,\"name\":\"After\",\"email\":\"[email protected]\","
+ + "\"legacy_note\":\"new-schema-value\"}",
+ writtenValues.get(1).get(0));
+ }
+
+ @Test
+ void testSnapshotStateRestoresLatestSchema() throws IOException {
+ List<List<String>> writtenValues = captureStringWrites();
+ configureJsonKeySink(1);
+ redisSinkWriter = new RedisSinkWriter(initialSchema(),
mockRedisParameters);
+
+ schemaEvolutionWriter().applySchemaChange(addEmailEvent());
+ List<?> states = redisSinkWriter.snapshotState(1L);
+
+ Assertions.assertEquals(1, states.size());
+ redisSinkWriter = new RedisSinkWriter((TableSchema) states.get(0),
mockRedisParameters);
+ redisSinkWriter.write(insertRow(1L, "Recovered",
"[email protected]", "legacy"));
+
+ Assertions.assertEquals(1, writtenValues.size());
+ Assertions.assertEquals(
+
"{\"id\":1,\"name\":\"Recovered\",\"email\":\"[email protected]\","
+ + "\"legacy_note\":\"legacy\"}",
+ writtenValues.get(0).get(0));
+ }
+
+ @Test
+ void testSnapshotStateNormalizesMissingConstraintKeys() {
+ configureJsonKeySink(1);
+ TableSchema schema = new TableSchema(initialSchema().getColumns(),
null, null);
+ redisSinkWriter = new RedisSinkWriter(schema, mockRedisParameters);
+
+ List<TableSchema> states = redisSinkWriter.snapshotState(1L);
+
+ Assertions.assertEquals(1, states.size());
+ Assertions.assertNotSame(schema, states.get(0));
+ Assertions.assertNotNull(states.get(0).getConstraintKeys());
+ Assertions.assertTrue(states.get(0).getConstraintKeys().isEmpty());
+ }
+
+ private SupportSchemaEvolutionSinkWriter schemaEvolutionWriter() {
+ Assertions.assertInstanceOf(SupportSchemaEvolutionSinkWriter.class,
redisSinkWriter);
+ return (SupportSchemaEvolutionSinkWriter) redisSinkWriter;
+ }
+
+ private List<List<String>> captureStringWrites() {
+ List<List<String>> writtenValues = new ArrayList<>();
+ Mockito.doAnswer(
+ invocation -> {
+ List<String> values = invocation.getArgument(2);
+ writtenValues.add(new ArrayList<>(values));
+ return null;
+ })
+ .when(mockRedisClient)
+ .batchWriteString(
+ Mockito.anyList(), Mockito.anyList(),
Mockito.anyList(), Mockito.anyLong());
+ return writtenValues;
+ }
+
+ private void configureJsonKeySink(int batchSize) {
+ when(mockRedisParameters.getBatchSize()).thenReturn(batchSize);
+
when(mockRedisParameters.getKeyField()).thenReturn("schema-change:${id}");
+ when(mockRedisParameters.getSupportCustomKey()).thenReturn(true);
+
when(mockRedisParameters.getRedisDataType()).thenReturn(RedisDataType.KEY);
+ when(mockRedisParameters.getExpire()).thenReturn(0L);
+ }
+
+ private static SeaTunnelRow insertRow(Object... fields) {
+ SeaTunnelRow row = new SeaTunnelRow(fields);
+ row.setRowKind(RowKind.INSERT);
+ return row;
+ }
+
+ private static AlterTableAddColumnEvent addEmailEvent() {
+ return AlterTableAddColumnEvent.addAfter(
+ TABLE_IDENTIFIER,
+ PhysicalColumn.of("email", BasicType.STRING_TYPE, 128L, true,
null, null),
+ "name");
+ }
+
+ private static TableSchema initialSchema() {
+ return TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.LONG_TYPE, 0L,
false, null, null))
+ .column(PhysicalColumn.of("name", BasicType.STRING_TYPE, 64L,
true, null, null))
+ .column(
+ PhysicalColumn.of(
+ "legacy_note", BasicType.STRING_TYPE, 128L,
true, null, null))
+ .build();
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/pom.xml
index 1f38de643f..c13f97359e 100644
--- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/pom.xml
@@ -25,6 +25,10 @@
<artifactId>connector-redis-e2e</artifactId>
<name>SeaTunnel : E2E : Connector V2 : Redis</name>
+ <properties>
+ <mysql.version>8.0.27</mysql.version>
+ </properties>
+
<dependencies>
<!-- SeaTunnel connectors -->
<dependency>
@@ -41,6 +45,28 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-mysql</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>${mysql.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>mysql</artifactId>
+ <version>${testcontainer.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>dns-cache-manipulator</artifactId>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisSchemaChangeIT.java
new file mode 100644
index 0000000000..2427dd5e09
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisSchemaChangeIT.java
@@ -0,0 +1,438 @@
+/*
+ * 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.e2e.connector.redis;
+
+import
org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.MySQLContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
+
+import com.mysql.cj.jdbc.Driver;
+import lombok.extern.slf4j.Slf4j;
+import redis.clients.jedis.Jedis;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.await;
+
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason = "Only Zeta supports schema evolution for this
streaming CDC case.")
+public class RedisSchemaChangeIT extends TestSuiteBase implements TestResource
{
+ private static final String MYSQL_HOST = "mysql_cdc_redis_e2e";
+ private static final String REDIS_HOST = "redis_cdc_e2e";
+ private static final int REDIS_PORT = 6379;
+ private static final String REDIS_PASSWORD = "SeaTunnel";
+ private static final String REDIS_IMAGE = "redis:7";
+ private static final String MYSQL_IMAGE = "mysql:8.0.43";
+
+ private static final String DATABASE = "shop";
+ private static final String TABLE = "schema_events";
+ private static final String MYSQL_USER_NAME = "root";
+ private static final String MYSQL_USER_PASSWORD = "mysqlpw";
+
+ private static final String KEY_1 = "schema-change:1";
+ private static final String KEY_2 = "schema-change:2";
+ private static final String JOB_CONFIG =
"/mysqlcdc_to_redis_with_schema_change.conf";
+ private static final String INCREMENTAL_READ_MARKER =
+ "Start incremental read task for incremental split";
+ private static final String MYSQL_CDC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/MySQL-CDC/lib";
+
+ private static final MySQLContainer<?> MYSQL_CONTAINER =
createMySqlContainer();
+
+ private GenericContainer<?> redisContainer;
+ private Connection mysqlConnection;
+ private Jedis jedis;
+
+ @TestContainerExtension
+ protected final ContainerExtendedFactory extendedFactory =
+ RedisSchemaChangeIT::copyMySQLDriverToContainer;
+
+ @BeforeAll
+ @Override
+ public void startUp() throws SQLException {
+ redisContainer =
+ new GenericContainer<>(DockerImageName.parse(REDIS_IMAGE))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(REDIS_HOST)
+ .withExposedPorts(REDIS_PORT)
+ .withCommand("redis-server --requirepass " +
REDIS_PASSWORD)
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger(REDIS_IMAGE)))
+ .waitingFor(
+ new HostPortWaitStrategy()
+
.withStartupTimeout(Duration.ofMinutes(2)));
+
+ Startables.deepStart(Stream.of(MYSQL_CONTAINER,
redisContainer)).join();
+
+ mysqlConnection =
+ DriverManager.getConnection(
+ MYSQL_CONTAINER.getJdbcUrl(),
+ MYSQL_CONTAINER.getUsername(),
+ MYSQL_CONTAINER.getPassword());
+ jedis = new Jedis(redisContainer.getHost(),
redisContainer.getFirstMappedPort());
+ jedis.auth(REDIS_PASSWORD);
+ Assertions.assertEquals("PONG", jedis.ping());
+
+ executeSql(
+ "CREATE USER IF NOT EXISTS 'st_user_source'@'%' IDENTIFIED BY
'mysqlpw'",
+ "GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, "
+ + "REPLICATION CLIENT, DROP, LOCK TABLES ON *.* TO
'st_user_source'@'%'",
+ "FLUSH PRIVILEGES",
+ "DROP TABLE IF EXISTS " + DATABASE + "." + TABLE,
+ "CREATE TABLE "
+ + DATABASE
+ + "."
+ + TABLE
+ + " ("
+ + "id BIGINT NOT NULL PRIMARY KEY,"
+ + "name VARCHAR(64),"
+ + "legacy_note VARCHAR(128))",
+ "INSERT INTO "
+ + DATABASE
+ + "."
+ + TABLE
+ + "(id,name,legacy_note) VALUES "
+ + "(1,'before-change','to-be-dropped')");
+ jedis.del(KEY_1, KEY_2);
+ }
+
+ @TestTemplate
+ public void testMysqlCdcToRedisSchemaEvolution(TestContainer container)
throws Exception {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(JOB_CONFIG, jobId);
+ } catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ });
+
+ try {
+ awaitRedisJson(
+ jobFuture,
+ KEY_1,
+ json -> {
+ Assertions.assertEquals(3, json.size());
+ Assertions.assertEquals(1L, json.get("id").asLong());
+ Assertions.assertEquals("before-change",
json.get("name").asText());
+ Assertions.assertEquals("to-be-dropped",
json.get("legacy_note").asText());
+ });
+ waitForIncrementalRead(container, jobFuture);
+
+ executeSql(
+ "ALTER TABLE "
+ + DATABASE
+ + "."
+ + TABLE
+ + " ADD COLUMN email VARCHAR(128) NULL AFTER name",
+ "UPDATE "
+ + DATABASE
+ + "."
+ + TABLE
+ + " SET email='[email protected]' WHERE id=1");
+
+ awaitRedisJson(
+ jobFuture,
+ KEY_1,
+ json -> {
+ Assertions.assertEquals(4, json.size());
+ Assertions.assertEquals("[email protected]",
json.get("email").asText());
+ Assertions.assertEquals("to-be-dropped",
json.get("legacy_note").asText());
+ });
+
+ executeSql(
+ "ALTER TABLE "
+ + DATABASE
+ + "."
+ + TABLE
+ + " ADD COLUMN score INT NULL AFTER email",
+ "UPDATE " + DATABASE + "." + TABLE + " SET score=42 WHERE
id=1",
+ "INSERT INTO "
+ + DATABASE
+ + "."
+ + TABLE
+ + "(id,name,email,score,legacy_note) VALUES "
+ +
"(2,'second-row','[email protected]',84,'second-legacy')");
+
+ awaitRedisJson(
+ jobFuture,
+ KEY_1,
+ json -> {
+ Assertions.assertEquals(5, json.size());
+ Assertions.assertEquals("[email protected]",
json.get("email").asText());
+ Assertions.assertEquals(42, json.get("score").asInt());
+ });
+ awaitRedisJson(
+ jobFuture,
+ KEY_2,
+ json -> {
+ Assertions.assertEquals(5, json.size());
+ Assertions.assertEquals("second-row",
json.get("name").asText());
+ Assertions.assertEquals("[email protected]",
json.get("email").asText());
+ Assertions.assertEquals(84, json.get("score").asInt());
+ Assertions.assertEquals("second-legacy",
json.get("legacy_note").asText());
+ });
+
+ executeSql(
+ "ALTER TABLE " + DATABASE + "." + TABLE + " DROP COLUMN
legacy_note",
+ "UPDATE "
+ + DATABASE
+ + "."
+ + TABLE
+ + " SET name='after-drop' WHERE id IN (1,2)");
+
+ awaitRedisJson(jobFuture, KEY_1,
RedisSchemaChangeIT::assertAfterDrop);
+ awaitRedisJson(jobFuture, KEY_2,
RedisSchemaChangeIT::assertAfterDrop);
+ } finally {
+ Throwable cleanupFailure = stopJobAndAwait(container, jobId,
jobFuture);
+ try {
+ jedis.del(KEY_1, KEY_2);
+ } catch (Throwable failure) {
+ cleanupFailure = mergeFailure(cleanupFailure, failure);
+ }
+ rethrowFailure(cleanupFailure);
+ }
+ }
+
+ private static MySQLContainer<?> createMySqlContainer() {
+ return new MySQLContainer<>(DockerImageName.parse(MYSQL_IMAGE))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MYSQL_HOST)
+ .withDatabaseName(DATABASE)
+ .withUsername(MYSQL_USER_NAME)
+ .withPassword(MYSQL_USER_PASSWORD)
+ .withCommand(
+ "--server-id=223344",
+ "--log-bin=mysql-bin",
+ "--binlog-format=ROW",
+ "--gtid-mode=ON",
+ "--enforce-gtid-consistency=ON")
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
+ }
+
+ private void awaitRedisJson(
+ CompletableFuture<Container.ExecResult> jobFuture,
+ String key,
+ Consumer<ObjectNode> assertion) {
+ await().atMost(180, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertJobHasNotFinished(jobFuture);
+ String value = jedis.get(key);
+ Assertions.assertNotNull(value, "Redis key is not
available: " + key);
+ assertion.accept(JsonUtils.parseObject(value));
+ });
+ }
+
+ private void waitForIncrementalRead(
+ TestContainer container, CompletableFuture<Container.ExecResult>
jobFuture) {
+ await().atMost(180, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertJobHasNotFinished(jobFuture);
+ String serverLogs = container.getServerLogs();
+ Assertions.assertTrue(
+
serverLogs.contains(INCREMENTAL_READ_MARKER),
+ "Incremental reader has not started yet");
+ Assertions.assertTrue(
+ serverLogs.contains(DATABASE + "." +
TABLE),
+ "Incremental reader has not captured the
source table");
+ });
+ }
+
+ private void executeSql(String... sqlStatements) {
+ try (Statement statement = mysqlConnection.createStatement()) {
+ for (String sql : sqlStatements) {
+ statement.execute(sql);
+ String statementType = sql.trim().split("\\s+",
2)[0].toUpperCase(Locale.ROOT);
+ log.info("Executed MySQL {} statement", statementType);
+ }
+ } catch (SQLException e) {
+ throw new IllegalStateException("Failed to execute MySQL schema
evolution SQL", e);
+ }
+ }
+
+ private static void assertAfterDrop(ObjectNode json) {
+ Assertions.assertEquals(4, json.size());
+ Assertions.assertEquals("after-drop", json.get("name").asText());
+ Assertions.assertTrue(json.has("email"));
+ Assertions.assertTrue(json.has("score"));
+ Assertions.assertFalse(json.has("legacy_note"));
+ }
+
+ private static void
assertJobHasNotFinished(CompletableFuture<Container.ExecResult> jobFuture) {
+ if (!jobFuture.isDone()) {
+ return;
+ }
+ try {
+ Container.ExecResult result = jobFuture.getNow(null);
+ Assertions.fail(
+ "Streaming job finished before Redis assertion. exitCode="
+ + result.getExitCode()
+ + ", stderr="
+ + result.getStderr());
+ } catch (CompletionException e) {
+ Assertions.fail("Streaming job failed before Redis assertion",
e.getCause());
+ }
+ }
+
+ private static Throwable stopJobAndAwait(
+ TestContainer container,
+ String jobId,
+ CompletableFuture<Container.ExecResult> jobFuture) {
+ Throwable failure = null;
+ if (!jobFuture.isDone()) {
+ try {
+ Container.ExecResult cancelResult = container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+ } catch (Throwable cancelFailure) {
+ failure = mergeFailure(failure, cancelFailure);
+ }
+ }
+ try {
+ Container.ExecResult result = jobFuture.get(60, TimeUnit.SECONDS);
+ Assertions.assertEquals(0, result.getExitCode(),
result.getStderr());
+ } catch (Throwable jobFailure) {
+ failure = mergeFailure(failure, jobFailure);
+ }
+ return failure;
+ }
+
+ private static Throwable mergeFailure(Throwable previousFailure, Throwable
failure) {
+ if (previousFailure == null) {
+ return failure;
+ }
+ previousFailure.addSuppressed(failure);
+ return previousFailure;
+ }
+
+ private static void rethrowFailure(Throwable failure) throws Exception {
+ if (failure == null) {
+ return;
+ }
+ if (failure instanceof Exception) {
+ throw (Exception) failure;
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw new AssertionError("Failed to stop schema evolution test job",
failure);
+ }
+
+ private static void copyMySQLDriverToContainer(GenericContainer<?>
container)
+ throws IOException, InterruptedException {
+ Container.ExecResult mkdirResult =
+ container.execInContainer("bash", "-c", "mkdir -p " +
MYSQL_CDC_PLUGIN_LIB);
+ Assertions.assertEquals(0, mkdirResult.getExitCode(),
mkdirResult.getStderr());
+
+ Path driverJarPath = mysqlDriverJarPath();
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ MYSQL_CDC_PLUGIN_LIB + "/" + driverJarPath.getFileName());
+ }
+
+ private static Path mysqlDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
+ @AfterAll
+ @Override
+ public void tearDown() throws Exception {
+ Throwable failure = null;
+ failure = closeResource(jedis, failure);
+ failure = closeResource(mysqlConnection, failure);
+ failure = closeResource(redisContainer, failure);
+ failure = closeResource(MYSQL_CONTAINER, failure);
+ if (failure != null) {
+ throw new Exception(
+ "Failed to close one or more schema evolution test
resources", failure);
+ }
+ }
+
+ private static Throwable closeResource(AutoCloseable resource, Throwable
previousFailure) {
+ if (Objects.isNull(resource)) {
+ return previousFailure;
+ }
+ try {
+ resource.close();
+ } catch (Throwable failure) {
+ if (previousFailure == null) {
+ return failure;
+ }
+ previousFailure.addSuppressed(failure);
+ }
+ return previousFailure;
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/mysqlcdc_to_redis_with_schema_change.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/mysqlcdc_to_redis_with_schema_change.conf
new file mode 100644
index 0000000000..30c2890dd7
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/mysqlcdc_to_redis_with_schema_change.conf
@@ -0,0 +1,46 @@
+#
+# 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 = "STREAMING"
+ checkpoint.interval = 2000
+}
+
+source {
+ MySQL-CDC {
+ server-id = 5950-5955
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["shop.schema_events"]
+ url = "jdbc:mysql://mysql_cdc_redis_e2e:3306/shop"
+ schema-changes.enabled = true
+ }
+}
+
+sink {
+ Redis {
+ host = "redis_cdc_e2e"
+ port = 6379
+ auth = "SeaTunnel"
+ key = "schema-change:{id}"
+ support_custom_key = true
+ data_type = key
+ batch_size = 1
+ format = json
+ }
+}
