You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2026/09/23 07:00:56 UTC
(flink-cdc) branch master updated: [FLINK-40552][runtime] Support asynchronous transform execution (#4526)
This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 7d560004b [FLINK-40552][runtime] Support asynchronous transform execution (#4526)
7d560004b is described below
commit 7d560004bcbbd644ff2411b693e4f87f1646080a
Author: haruki <je...@outlook.com>
AuthorDate: Wed Sep 23 15:00:51 2026 +0800
[FLINK-40552][runtime] Support asynchronous transform execution (#4526)
Co-authored-by: 春栖 <ch...@U-3L66HJF6-0148.local>
Co-authored-by: yux <34...@users.noreply.github.com>
---
docs/content.zh/docs/core-concept/data-pipeline.md | 8 +
docs/content/docs/core-concept/data-pipeline.md | 8 +
.../flink/cdc/common/model/AiModelClient.java | 4 +
.../flink/cdc/common/pipeline/PipelineOptions.java | 28 +
.../flink/cdc/common/udf/UserDefinedFunction.java | 4 +
.../cdc/composer/flink/FlinkPipelineComposer.java | 47 +-
.../flink/translator/TransformTranslator.java | 70 ++
.../cdc/composer/flink/FlinkPipelineUdfITCase.java | 286 ++++++
.../flink/translator/TransformTranslatorTest.java | 79 ++
.../cdc/composer/specs/TransformSpecsITCase.java | 17 +
.../tests/{stage3 => stage2}/RouteE2eITCase.java | 2 +-
.../pipeline/tests/stage3/TransformE2eITCase.java | 297 ++++++-
.../tests/utils/PipelineTestEnvironment.java | 36 +
.../operators/AsyncWaitOperatorAdapter.java | 61 ++
.../operators/AsyncWaitOperatorAdapter.java | 61 ++
.../java/SkewedThrottlerFunctionClass.java | 44 +
.../async/AsyncPostTransformFunction.java | 959 +++++++++++++++++++++
.../async/AsyncPostTransformFunctionBuilder.java | 131 +++
.../async/AsyncPostTransformOperatorFactory.java | 64 ++
.../async/AsyncPostTransformFunctionTest.java | 386 +++++++++
20 files changed, 2558 insertions(+), 34 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/data-pipeline.md b/docs/content.zh/docs/core-concept/data-pipeline.md
index ffa6f0d5f..2e228574c 100644
--- a/docs/content.zh/docs/core-concept/data-pipeline.md
+++ b/docs/content.zh/docs/core-concept/data-pipeline.md
@@ -125,5 +125,13 @@ under the License.
| `operator.uid.prefix` | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的 UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。 [...]
| `sink.partitioning.strategy` | Sink 写入数据时使用的分区策略。数据类型：String。默认值：`SINK_DEFINED`。备注：可配置的值如下：`SINK_DEFINED`：使用 Sink 定义的分区策略；`PRIMARY_KEY`：按表 ID 和主键分区；`TABLE_ID`：仅按表 ID 分区。 | optional |
| `transform.decimal.precision.mode` | transform 表达式求值中 DECIMAL 类型的最大精度模式。可选值：`UP_TO_19`（默认，使用 Calcite 默认类型系统）或 `UP_TO_38`（允许 DECIMAL 精度最高为 38 位）。 | optional |
+| `transform.async-execution.enabled` | 是否为 PostTransform 开启有序异步执行，默认值为 `false`。 | op [...]
+| `transform.async-execution.timeout` | 有序异步执行中每个 PostTransform 事件的超时时间，默认值为 5 分钟。 | option [...]
+| `transform.async-execution.capacity` | 异步执行中最多允许同时处理的 PostTransform 事件数，默认值为 100。 | option [...]
+| `transform.async-execution.worker-threads` | 每个异步 PostTransform 任务使用的工作线程数，默认值为 16。 | option [...]
+
+> **实验性且不稳定：** 异步 PostTransform 执行目前是实验性功能，其配置和行为可能会在未来版本中发生变化。
+
+异步 PostTransform 适用于 AI 模型调用等 I/O 密集型表达式。DataChangeEvent 可能并发调用 UDF 和 AI 模型客户端，但输出和所有 SchemaChangeEvent 仍保持有序，因此 UDF 和 AI 模型客户端实现必须是线程安全的。为保证 schema 状态一致，checkpoint 或 savepoint 前会等待尚未完成的异步请求，长时间运行的请求可能会延长 checkpoint 时间。savepoint 只支持在并行度不变时恢复，并且不能在从已有 savepoint 恢复时开启或关闭此选项。
注意：虽然上述参数都是可选的，但至少需要指定其中一个。`pipeline` 部分是必需的，不能为空。
diff --git a/docs/content/docs/core-concept/data-pipeline.md b/docs/content/docs/core-concept/data-pipeline.md
index 4b35ad6ca..664c5c5db 100644
--- a/docs/content/docs/core-concept/data-pipeline.md
+++ b/docs/content/docs/core-concept/data-pipeline.md
@@ -127,5 +127,13 @@ Note that whilst the parameters are each individually optional, at least one of
| `operator.uid.prefix` | The prefix to use for all pipeline operator UIDs. If not set, all pipeline operator UIDs will be generated by Flink. It is recommended to set this parameter to ensure stable and recognizable operator UIDs, which can help with stateful upgrades, troubleshooting, and Flink UI diagnostics. [...]
| `sink.partitioning.strategy` | The partitioning strategy used when writing data to the sink. Data type: String. Default value: `SINK_DEFINED`. Available values: `SINK_DEFINED`: uses the partitioning strategy defined by the sink; `PRIMARY_KEY`: partitions by table ID and primary key; `TABLE_ID`: partitions only by table ID. [...]
| `transform.decimal.precision.mode` | Maximum precision mode for DECIMAL type in transform expression evaluation. One of: `UP_TO_19` (default, match Calcite's default type system) or `UP_TO_38` (allow DECIMAL precision up to 38 digits). [...]
+| `transform.async-execution.enabled` | Whether to enable ordered asynchronous execution for post-transform. Defaults to `false`. [...]
+| `transform.async-execution.timeout` | The timeout for each post-transform event in ordered asynchronous execution. Defaults to 5 minutes. [...]
+| `transform.async-execution.capacity` | The maximum number of in-flight post-transform events. Defaults to 100. [...]
+| `transform.async-execution.worker-threads` | The number of worker threads used by each asynchronous post-transform task. Defaults to 16. [...]
+
+> **Experimental and unstable:** Asynchronous post-transform execution is an experimental feature. Its configuration and behavior may change in future releases.
+
+Asynchronous post-transform execution is intended for I/O-bound expressions such as AI model calls. Data change events may invoke UDFs and AI model clients concurrently, while their output and all schema changes remain ordered. UDF and AI model client implementations must therefore be thread-safe. Pending asynchronous requests are completed before a checkpoint or savepoint to keep schema state consistent, so long-running requests may extend checkpoint duration. Savepoint restore is suppo [...]
NOTE: Whilst the above parameters are each individually optional, at least one of them must be specified. The `pipeline` section is mandatory and cannot be empty.
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java
index 0e46508bb..512c684ac 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java
@@ -27,6 +27,10 @@ import java.io.Serializable;
*
* <p>Implementations must be {@link Serializable} so that they can be distributed across Flink task
* managers together with the operator that holds them.
+ *
+ * <p>When {@code transform.async-execution.enabled} is enabled, the same client instance may be
+ * invoked concurrently. Implementations used in asynchronous transforms must therefore be
+ * thread-safe.
*/
@Experimental
public interface AiModelClient extends Serializable, AutoCloseable {
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
index d46c94c65..3b39645d0 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
@@ -175,5 +175,33 @@ public class PipelineOptions {
"UP_TO_38: Allows DECIMAL precision up to 38 digits, matching Flink CDC's extended type system.")))
.build());
+ public static final ConfigOption<Boolean> PIPELINE_TRANSFORM_ASYNC_EXECUTION_ENABLED =
+ ConfigOptions.key("transform.async-execution.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to enable ordered async execution for post-transform.");
+
+ public static final ConfigOption<Duration> PIPELINE_TRANSFORM_ASYNC_EXECUTION_TIMEOUT =
+ ConfigOptions.key("transform.async-execution.timeout")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(5))
+ .withDescription(
+ "The timeout for each post-transform event in ordered async execution.");
+
+ public static final ConfigOption<Integer> PIPELINE_TRANSFORM_ASYNC_EXECUTION_CAPACITY =
+ ConfigOptions.key("transform.async-execution.capacity")
+ .intType()
+ .defaultValue(100)
+ .withDescription(
+ "The maximum number of in-flight post-transform events in ordered async execution.");
+
+ public static final ConfigOption<Integer> PIPELINE_TRANSFORM_ASYNC_EXECUTION_WORKER_THREADS =
+ ConfigOptions.key("transform.async-execution.worker-threads")
+ .intType()
+ .defaultValue(16)
+ .withDescription(
+ "The number of worker threads used by each async post-transform task.");
+
private PipelineOptions() {}
}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
index 0fa5d78c8..f20343b13 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
@@ -23,6 +23,10 @@ import org.apache.flink.cdc.common.types.DataType;
/**
* Base interface for creating a UDF in transform projection and filtering expressions. You should
* define at least one {@code eval} method.
+ *
+ * <p>When {@code transform.async-execution.enabled} is enabled, the same UDF instance may be
+ * invoked concurrently. Implementations used in asynchronous transforms must therefore be
+ * thread-safe.
*/
@PublicEvolving
public interface UserDefinedFunction {
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
index c0d693ed2..3b73a283e 100644
--- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
+++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
@@ -192,19 +192,40 @@ public class FlinkPipelineComposer implements PipelineComposer {
dataSource.supportedMetadataColumns());
// PreTransform ---> PostTransform
- stream =
- transformTranslator.translatePostTransform(
- stream,
- pipelineDef.getTransforms(),
- pipelineDef.getConfig().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
- pipelineDef
- .getConfig()
- .get(PipelineOptions.PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE),
- pipelineDef.getUdfs(),
- pipelineDef.getModels(),
- dataSource.supportedMetadataColumns(),
- operatorUidGenerator,
- env);
+ if (pipelineDefConfig.get(PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_ENABLED)) {
+ stream =
+ transformTranslator.translateAsyncPostTransform(
+ stream,
+ pipelineDef.getTransforms(),
+ pipelineDefConfig.get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
+ pipelineDefConfig.get(
+ PipelineOptions.PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE),
+ pipelineDef.getUdfs(),
+ pipelineDef.getModels(),
+ dataSource.supportedMetadataColumns(),
+ operatorUidGenerator,
+ pipelineDefConfig.get(
+ PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_TIMEOUT),
+ pipelineDefConfig.get(
+ PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_CAPACITY),
+ pipelineDefConfig.get(
+ PipelineOptions
+ .PIPELINE_TRANSFORM_ASYNC_EXECUTION_WORKER_THREADS),
+ env);
+ } else {
+ stream =
+ transformTranslator.translatePostTransform(
+ stream,
+ pipelineDef.getTransforms(),
+ pipelineDefConfig.get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
+ pipelineDefConfig.get(
+ PipelineOptions.PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE),
+ pipelineDef.getUdfs(),
+ pipelineDef.getModels(),
+ dataSource.supportedMetadataColumns(),
+ operatorUidGenerator,
+ env);
+ }
if (isParallelMetadataSource) {
// Translate a distributed topology for sources with distributed tables
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
index 99be446f2..a8c650c07 100644
--- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
+++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
@@ -35,16 +35,22 @@ import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator;
import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperatorBuilder;
import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperator;
import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperatorBuilder;
+import org.apache.flink.cdc.runtime.operators.transform.async.AsyncPostTransformFunction;
+import org.apache.flink.cdc.runtime.operators.transform.async.AsyncPostTransformFunctionBuilder;
+import org.apache.flink.cdc.runtime.operators.transform.async.AsyncPostTransformOperatorFactory;
import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import java.time.Duration;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import static org.apache.flink.cdc.common.utils.Preconditions.checkArgument;
+
/**
* Translator used to build {@link PreTransformOperator} and {@link PostTransformOperator} for event
* transform.
@@ -148,6 +154,70 @@ public class TransformTranslator {
.uid(operatorUidGenerator.generateUid("post-transform"));
}
+ public DataStream<Event> translateAsyncPostTransform(
+ DataStream<Event> input,
+ List<TransformDef> transforms,
+ String timezone,
+ DecimalPrecisionMode decimalPrecisionMode,
+ List<UdfDef> udfFunctions,
+ List<ModelDef> models,
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ OperatorUidGenerator operatorUidGenerator,
+ Duration asyncTransformTimeout,
+ int asyncTransformCapacity,
+ int asyncTransformWorkerThreads,
+ StreamExecutionEnvironment env) {
+ if (transforms.isEmpty()) {
+ return input;
+ }
+ checkArgument(
+ asyncTransformTimeout.toMillis() > 0,
+ "Async transform timeout must be greater than 0.");
+ checkArgument(
+ asyncTransformCapacity > 0, "Async transform capacity must be greater than 0.");
+ checkArgument(
+ asyncTransformWorkerThreads > 0,
+ "Async transform worker threads must be greater than 0.");
+
+ AsyncPostTransformFunctionBuilder asyncPostTransformFunctionBuilder =
+ AsyncPostTransformFunction.newBuilder();
+ for (TransformDef transform : transforms) {
+ asyncPostTransformFunctionBuilder.addTransform(
+ transform.getSourceTable(),
+ transform.getProjection(),
+ transform.getFilter(),
+ transform.getPrimaryKeys(),
+ transform.getPartitionKeys(),
+ transform.getTableOptions(),
+ transform.getTableOptionsDelimiter(),
+ transform.getPostTransformConverter(),
+ supportedMetadataColumns);
+ }
+ asyncPostTransformFunctionBuilder.addTimezone(timezone);
+ asyncPostTransformFunctionBuilder.addDecimalPrecisionMode(decimalPrecisionMode);
+ asyncPostTransformFunctionBuilder.addUdfFunctions(
+ udfFunctions.stream().map(this::udfDefToUDFTuple).collect(Collectors.toList()));
+ asyncPostTransformFunctionBuilder.addUdfFunctions(
+ models.stream()
+ .filter(ModelDef::isLegacy)
+ .map(this::modelToUDFTuple)
+ .collect(Collectors.toList()));
+ Map<String, AiModelClient> modelClients = loadModelClients(models, env);
+ asyncPostTransformFunctionBuilder.addModelClients(modelClients);
+ asyncPostTransformFunctionBuilder.addAsyncWorkerThreads(asyncTransformWorkerThreads);
+
+ long timeoutMillis = asyncTransformTimeout.toMillis();
+ return input.transform(
+ "Transform:Data",
+ new EventTypeInfo(),
+ new AsyncPostTransformOperatorFactory(
+ asyncPostTransformFunctionBuilder.build(),
+ timeoutMillis,
+ asyncTransformCapacity))
+ .name("Transform:Data")
+ .uid(operatorUidGenerator.generateUid("post-transform"));
+ }
+
private Tuple3<String, String, Map<String, String>> modelToUDFTuple(ModelDef model) {
return Tuple3.of(
model.getModelName(),
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
index 9d5a8b22e..2bb4b14f3 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
@@ -18,8 +18,19 @@
package org.apache.flink.cdc.composer.flink;
import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.udf.UserDefinedFunction;
+import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
import org.apache.flink.cdc.composer.PipelineExecution;
import org.apache.flink.cdc.composer.definition.ModelDef;
import org.apache.flink.cdc.composer.definition.PipelineDef;
@@ -33,6 +44,7 @@ import org.apache.flink.cdc.connectors.values.sink.ValuesDataSink;
import org.apache.flink.cdc.connectors.values.sink.ValuesDataSinkOptions;
import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper;
import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceOptions;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.test.junit5.MiniClusterExtension;
@@ -41,6 +53,7 @@ import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -48,19 +61,41 @@ import org.junit.jupiter.params.provider.MethodSource;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
+import static org.apache.flink.cdc.common.types.DataTypes.BIGINT;
+import static org.apache.flink.cdc.common.types.DataTypes.INT;
+import static org.apache.flink.cdc.common.types.DataTypes.STRING;
import static org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.params.provider.Arguments.arguments;
/** Integration test for UDFs. */
class FlinkPipelineUdfITCase {
private static final int MAX_PARALLELISM = 4;
+ private static final TableId ASYNC_TRANSFORM_TABLE_ID =
+ TableId.tableId("foo", "bar", "async_transform");
+ private static final Schema ASYNC_TRANSFORM_SCHEMA =
+ Schema.newBuilder()
+ .physicalColumn("id_", BIGINT().notNull())
+ .physicalColumn("name_", STRING())
+ .build();
+ private static final Schema ASYNC_TRANSFORM_SCHEMA_WITH_REGION =
+ Schema.newBuilder()
+ .physicalColumn("id_", BIGINT().notNull())
+ .physicalColumn("name_", STRING())
+ .physicalColumn("region_", STRING())
+ .build();
// Always use parent-first classloader for CDC classes.
// The reason is that ValuesDatabase uses static field for holding data, we need to make sure
@@ -106,6 +141,113 @@ class FlinkPipelineUdfITCase {
// ----------------------
// CDC pipeline UDF tests
// ----------------------
+ @Test
+ void testAsyncTransformExecutesDataConcurrentlyInOrderWithSchemaBarrier() throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.CUSTOM_SOURCE_EVENTS);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(ValuesDataSinkOptions.PRINT_ENABLED, true);
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+
+ ValuesDataSourceHelper.setSourceEvents(
+ Collections.singletonList(createAsyncTransformEvents()));
+
+ TransformDef transformDef =
+ new TransformDef(
+ ASYNC_TRANSFORM_TABLE_ID.toString(),
+ "*, completion_order(id_) AS completion_order_",
+ null,
+ null,
+ null,
+ null,
+ "async transform ordering",
+ null);
+ UdfDef udfDef = new UdfDef("completion_order", CompletionOrderUdf.class.getName());
+
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
+ pipelineConfig.set(PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_ENABLED, true);
+ pipelineConfig.set(PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_WORKER_THREADS, 2);
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.singletonList(transformDef),
+ Collections.singletonList(udfDef),
+ pipelineConfig);
+
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+
+ assertThat(outCaptor.toString().trim().split("\n"))
+ .containsExactly(
+ "CreateTableEvent{tableId=foo.bar.async_transform, schema=columns={`id_` BIGINT NOT NULL,`name_` STRING,`completion_order_` INT}, primaryKeys=, options=()}",
+ "DataChangeEvent{tableId=foo.bar.async_transform, before=[], after=[1, name-1, 2], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=foo.bar.async_transform, before=[], after=[2, name-2, 1], op=INSERT, meta=()}",
+ "AddColumnEvent{tableId=foo.bar.async_transform, addedColumns=[ColumnWithPosition{column=`region_` STRING, position=AFTER, existedColumnName=name_}]}",
+ "DataChangeEvent{tableId=foo.bar.async_transform, before=[], after=[3, name-3, region-3, 3], op=INSERT, meta=()}");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("asyncTransformFailureCases")
+ void testAsyncTransformPropagatesFailures(
+ String testName, String udfClassName, Duration timeout, String expectedMessage) {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.CUSTOM_SOURCE_EVENTS);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+
+ Configuration sinkConfig = new Configuration();
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+
+ ValuesDataSourceHelper.setSourceEvents(
+ Collections.singletonList(createSingleAsyncTransformRecord()));
+
+ TransformDef transformDef =
+ new TransformDef(
+ ASYNC_TRANSFORM_TABLE_ID.toString(),
+ "*, test_failure(name_) AS result_",
+ null,
+ null,
+ null,
+ null,
+ testName,
+ null);
+ UdfDef udfDef = new UdfDef("test_failure", udfClassName);
+
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
+ pipelineConfig.set(PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_ENABLED, true);
+ pipelineConfig.set(PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_TIMEOUT, timeout);
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.singletonList(transformDef),
+ Collections.singletonList(udfDef),
+ pipelineConfig);
+
+ assertThatThrownBy(() -> composer.compose(pipelineDef).execute())
+ .hasStackTraceContaining(expectedMessage);
+ }
+
@ParameterizedTest
@MethodSource("testParams")
void testTransformWithUdf(ValuesDataSink.SinkApi sinkApi, String language) throws Exception {
@@ -1095,4 +1237,148 @@ class FlinkPipelineUdfITCase {
arguments(ValuesDataSink.SinkApi.SINK_FUNCTION, "scala"),
arguments(ValuesDataSink.SinkApi.SINK_V2, "scala"));
}
+
+ private static List<Event> createAsyncTransformEvents() {
+ BinaryRecordDataGenerator initialSchemaGenerator =
+ new BinaryRecordDataGenerator(
+ ASYNC_TRANSFORM_SCHEMA.getColumnDataTypes().toArray(new DataType[0]));
+ BinaryRecordDataGenerator schemaWithRegionGenerator =
+ new BinaryRecordDataGenerator(
+ ASYNC_TRANSFORM_SCHEMA_WITH_REGION
+ .getColumnDataTypes()
+ .toArray(new DataType[0]));
+ List<Event> events = new ArrayList<>();
+ events.add(new CreateTableEvent(ASYNC_TRANSFORM_TABLE_ID, ASYNC_TRANSFORM_SCHEMA));
+ events.add(
+ DataChangeEvent.insertEvent(
+ ASYNC_TRANSFORM_TABLE_ID,
+ initialSchemaGenerator.generate(
+ new Object[] {1L, BinaryStringData.fromString("name-1")})));
+ events.add(
+ DataChangeEvent.insertEvent(
+ ASYNC_TRANSFORM_TABLE_ID,
+ initialSchemaGenerator.generate(
+ new Object[] {2L, BinaryStringData.fromString("name-2")})));
+ events.add(
+ new AddColumnEvent(
+ ASYNC_TRANSFORM_TABLE_ID,
+ Collections.singletonList(
+ AddColumnEvent.last(Column.physicalColumn("region_", STRING())))));
+ events.add(
+ DataChangeEvent.insertEvent(
+ ASYNC_TRANSFORM_TABLE_ID,
+ schemaWithRegionGenerator.generate(
+ new Object[] {
+ 3L,
+ BinaryStringData.fromString("name-3"),
+ BinaryStringData.fromString("region-3")
+ })));
+ return events;
+ }
+
+ private static List<Event> createSingleAsyncTransformRecord() {
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+ ASYNC_TRANSFORM_SCHEMA.getColumnDataTypes().toArray(new DataType[0]));
+ return Arrays.asList(
+ new CreateTableEvent(ASYNC_TRANSFORM_TABLE_ID, ASYNC_TRANSFORM_SCHEMA),
+ DataChangeEvent.insertEvent(
+ ASYNC_TRANSFORM_TABLE_ID,
+ generator.generate(
+ new Object[] {1L, BinaryStringData.fromString("name-1")})));
+ }
+
+ private static Stream<Arguments> asyncTransformFailureCases() {
+ return Stream.of(
+ arguments(
+ "UDF exception",
+ FailingUdf.class.getName(),
+ Duration.ofSeconds(30),
+ "expected async transform failure"),
+ arguments(
+ "timeout",
+ TimeoutUdf.class.getName(),
+ Duration.ofMillis(100),
+ "Async post-transform timed out for event"));
+ }
+
+ public static class CompletionOrderUdf implements UserDefinedFunction {
+
+ private transient CountDownLatch secondRecordCompleted;
+ private transient CountDownLatch firstRecordCompleted;
+ private transient AtomicInteger completionOrder;
+
+ @Override
+ public DataType getReturnType(UserDefinedFunctionContext context) {
+ return INT();
+ }
+
+ @Override
+ public void open(UserDefinedFunctionContext context) {
+ secondRecordCompleted = new CountDownLatch(1);
+ firstRecordCompleted = new CountDownLatch(1);
+ completionOrder = new AtomicInteger();
+ }
+
+ public Integer eval(Long value) {
+ if (value == 1L) {
+ await(secondRecordCompleted);
+ int order = completionOrder.incrementAndGet();
+ firstRecordCompleted.countDown();
+ return order;
+ }
+ if (value == 2L) {
+ int order = completionOrder.incrementAndGet();
+ secondRecordCompleted.countDown();
+ return order;
+ }
+ if (firstRecordCompleted.getCount() != 0) {
+ throw new IllegalStateException(
+ "Data after the schema change ran before preceding data completed.");
+ }
+ return completionOrder.incrementAndGet();
+ }
+
+ private void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException(
+ "Later data was not executed concurrently with earlier data.");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for concurrent data.", e);
+ }
+ }
+ }
+
+ public static class FailingUdf implements UserDefinedFunction {
+
+ @Override
+ public DataType getReturnType(UserDefinedFunctionContext context) {
+ return STRING();
+ }
+
+ public String eval(String value) {
+ throw new IllegalStateException("expected async transform failure");
+ }
+ }
+
+ public static class TimeoutUdf implements UserDefinedFunction {
+
+ @Override
+ public DataType getReturnType(UserDefinedFunctionContext context) {
+ return STRING();
+ }
+
+ public String eval(String value) {
+ try {
+ Thread.sleep(Duration.ofSeconds(30).toMillis());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for timeout.", e);
+ }
+ return value;
+ }
+ }
}
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslatorTest.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslatorTest.java
new file mode 100644
index 000000000..12aa4c3ca
--- /dev/null
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslatorTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.flink.cdc.composer.flink.translator;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.composer.definition.TransformDef;
+import org.apache.flink.cdc.runtime.operators.transform.async.AsyncPostTransformOperatorFactory;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TransformTranslator}. */
+class TransformTranslatorTest {
+
+ @Test
+ void testTranslateAsyncPostTransformUsesStateConsistentOperator() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ TableId tableId = TableId.tableId("ns", "schema", "customers");
+ Schema schema = Schema.newBuilder().physicalColumn("id", DataTypes.INT()).build();
+ DataStream<Event> input =
+ env.fromCollection(
+ Collections.singletonList((Event) new CreateTableEvent(tableId, schema)),
+ new EventTypeInfo());
+ TransformDef transform =
+ new TransformDef(tableId.identifier(), "*", null, null, null, null, null, null);
+
+ DataStream<Event> result =
+ new TransformTranslator()
+ .translateAsyncPostTransform(
+ input,
+ Collections.singletonList(transform),
+ "UTC",
+ DecimalPrecisionMode.UP_TO_19,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0],
+ new OperatorUidGenerator("test"),
+ Duration.ofSeconds(30),
+ 10,
+ 2,
+ env);
+
+ assertThat(result.getTransformation())
+ .isInstanceOfSatisfying(
+ OneInputTransformation.class,
+ transformation ->
+ assertThat(transformation.getOperatorFactory())
+ .isInstanceOf(AsyncPostTransformOperatorFactory.class));
+ }
+}
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java
index 337df690b..7c5f22e23 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java
@@ -57,9 +57,11 @@ import org.assertj.core.api.Assumptions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
@@ -118,6 +120,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Fail.fail;
/** Spec based transform module IT cases. */
+@ParameterizedClass
+@ValueSource(booleans = {false, true})
class TransformSpecsITCase {
private static final TableId testTableId = TableId.tableId("foo", "bar", "baz");
@@ -418,6 +422,11 @@ class TransformSpecsITCase {
private final PrintStream standardOut = System.out;
private final ByteArrayOutputStream outCaptor = new ByteArrayOutputStream();
+ private final boolean asyncTransform;
+
+ TransformSpecsITCase(boolean asyncTransform) {
+ this.asyncTransform = asyncTransform;
+ }
@BeforeEach
void takeOverStdOut() {
@@ -450,6 +459,12 @@ class TransformSpecsITCase {
.as("Test case %s is ignored until we close %s", spec.name, spec.ignore)
.isNull();
FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+ if (asyncTransform) {
+ // Keep the value source in a separate task. Otherwise, a transform failure in the
+ // chained async task may trigger irrelevant exceptions that hide the original
+ // exception.
+ composer.getEnv().disableOperatorChaining();
+ }
Configuration sourceConfig = new Configuration();
sourceConfig.set(EVENT_SET_ID, CUSTOM_SOURCE_EVENTS);
@@ -483,6 +498,8 @@ class TransformSpecsITCase {
pipelineConfig.set(
PipelineOptions.PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE,
spec.decimalPrecisionMode);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_TRANSFORM_ASYNC_EXECUTION_ENABLED, asyncTransform);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/RouteE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/RouteE2eITCase.java
similarity index 99%
rename from flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/RouteE2eITCase.java
rename to flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/RouteE2eITCase.java
index bb5a2c97f..e99a2e008 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/RouteE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/RouteE2eITCase.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.pipeline.tests.stage3;
+package org.apache.flink.cdc.pipeline.tests.stage2;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/TransformE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/TransformE2eITCase.java
index a844dae14..43a2c6a06 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/TransformE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage3/TransformE2eITCase.java
@@ -20,20 +20,24 @@ package org.apache.flink.cdc.pipeline.tests.stage3;
import org.apache.flink.api.common.JobID;
import org.apache.flink.cdc.common.data.DateData;
import org.apache.flink.cdc.common.data.TimeData;
+import org.apache.flink.cdc.common.test.utils.TestUtils;
import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator;
import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperator;
+import org.assertj.core.api.Assumptions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.nio.file.Path;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
@@ -53,24 +57,36 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
/** E2e tests for the {@link PreTransformOperator} and {@link PostTransformOperator}. */
-@ParameterizedClass
-@ValueSource(ints = {1, 4})
+@ParameterizedClass(name = "asyncTransform: {0} parallelism: {1}")
+@CsvSource(
+ value = {
+ "false,1", "true,1", "false,4", "true,4",
+ })
class TransformE2eITCase extends PipelineTestEnvironment {
private static final Logger LOG = LoggerFactory.getLogger(TransformE2eITCase.class);
+ private static final Duration ASYNC_RESTORE_TIMEOUT = Duration.ofMinutes(3);
protected final UniqueDatabase transformTestDatabase =
new UniqueDatabase(MYSQL, "transform_test", MYSQL_TEST_USER, MYSQL_TEST_PASSWORD);
- TransformE2eITCase(int parallelism) {
- super(parallelism);
- }
-
private final Function<String, String> dbNameFormatter =
(s) -> {
String databaseName = transformTestDatabase.getDatabaseName();
return String.format(s, databaseName, databaseName, databaseName);
};
+ private final boolean asyncTransform;
+
+ TransformE2eITCase(boolean asyncTransform, int parallelism) {
+ super(parallelism);
+ this.asyncTransform = asyncTransform;
+ }
+
+ private String withAsyncTransform(String pipelineJob) {
+ return pipelineJob
+ + String.format("%n transform.async-execution.enabled: %s", asyncTransform);
+ }
+
@BeforeEach
public void before() throws Exception {
super.before();
@@ -83,6 +99,129 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.dropDatabase();
}
+ @Test
+ void testAsyncTransformRestoreAfterInflightEvents() throws Exception {
+ Assumptions.assumeThat(asyncTransform).isTrue();
+
+ int slowRecordId = 4000;
+ int lastRecordId = 4007;
+ int slowRecordDelaySeconds = 15;
+ String databaseName = transformTestDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " scan.startup.mode: initial\n"
+ + " tables: %s.TABLEALPHA\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "transform:\n"
+ + " - source-table: %s.TABLEALPHA\n"
+ + " projection: \\*, throttle(ID, %d, %d) AS THROTTLED\n"
+ + "\n"
+ + "pipeline:\n"
+ + " transform.async-execution.enabled: true\n"
+ + " transform.async-execution.timeout: 1m\n"
+ + " transform.async-execution.capacity: 16\n"
+ + " transform.async-execution.worker-threads: 4\n"
+ + " parallelism: %d\n"
+ + " user-defined-function:\n"
+ + " - name: throttle\n"
+ + " classpath: org.apache.flink.cdc.udf.examples.java.SkewedThrottlerFunctionClass\n",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ databaseName,
+ databaseName,
+ slowRecordId,
+ slowRecordDelaySeconds,
+ parallelism);
+ Path udfJar = TestUtils.getResource("udf-examples.jar");
+
+ JobID jobId = submitPipelineJob(pipelineJob, udfJar);
+ waitUntilJobRunning(jobId, Duration.ofSeconds(30));
+ waitUntilStreamSplitReady(jobId, parallelism);
+
+ String mysqlJdbcUrl =
+ String.format(
+ "jdbc:mysql://%s:%s/%s",
+ MYSQL.getHost(), MYSQL.getDatabasePort(), databaseName);
+ int incrementalOutputOffset = taskManagerConsumer.toUtf8String().length();
+ addRegionColumnToTableAlpha(mysqlJdbcUrl);
+ insertTableAlphaRows(mysqlJdbcUrl, slowRecordId, lastRecordId);
+
+ waitUntilEventAppearsBeforeAnother(
+ incrementalOutputOffset,
+ ASYNC_RESTORE_TIMEOUT,
+ "SkewedThrottlerFunctionClass finished " + (slowRecordId + 1),
+ "SkewedThrottlerFunctionClass finished " + slowRecordId);
+
+ String savepointPath = stopJobWithSavepoint(jobId);
+ if (parallelism == 1) {
+ waitUntilSpecificEventsAfter(
+ incrementalOutputOffset,
+ ASYNC_RESTORE_TIMEOUT,
+ tableAlphaInsertEvent(databaseName, slowRecordId),
+ tableAlphaInsertEvent(databaseName, lastRecordId));
+ } else {
+ waitUntilEventsAfter(
+ incrementalOutputOffset,
+ ASYNC_RESTORE_TIMEOUT,
+ tableAlphaInsertEvent(databaseName, slowRecordId),
+ tableAlphaInsertEvent(databaseName, lastRecordId));
+ }
+
+ int restoredOutputOffset = taskManagerConsumer.toUtf8String().length();
+ JobID restoredJobId = submitPipelineJob(pipelineJob, savepointPath, false, udfJar);
+ waitUntilJobRunning(restoredJobId, Duration.ofSeconds(30));
+
+ insertTableAlphaRows(mysqlJdbcUrl, 5000, 5001);
+ String restoredCreateTableEvent =
+ String.format(
+ "CreateTableEvent{tableId=%s.TABLEALPHA, schema=columns={`ID` INT NOT NULL,`VERSION` VARCHAR(17),`PRICEALPHA` INT,`AGEALPHA` INT,`NAMEALPHA` VARCHAR(128),`REGION` VARCHAR(17),`THROTTLED` STRING}, primaryKeys=ID, options=()}",
+ databaseName);
+ if (parallelism == 1) {
+ waitUntilSpecificEventsAfter(
+ restoredOutputOffset,
+ ASYNC_RESTORE_TIMEOUT,
+ restoredCreateTableEvent,
+ tableAlphaInsertEvent(databaseName, 5000),
+ tableAlphaInsertEvent(databaseName, 5001));
+ } else {
+ waitUntilEventsAfter(
+ restoredOutputOffset,
+ ASYNC_RESTORE_TIMEOUT,
+ restoredCreateTableEvent,
+ tableAlphaInsertEvent(databaseName, 5000),
+ tableAlphaInsertEvent(databaseName, 5001));
+ }
+
+ String restoredOutput =
+ taskManagerConsumer
+ .toUtf8String()
+ .substring(
+ Math.min(
+ restoredOutputOffset,
+ taskManagerConsumer.toUtf8String().length()));
+ if (parallelism == 1) {
+ assertThat(restoredOutput).containsOnlyOnce(restoredCreateTableEvent);
+ } else {
+ assertThat(restoredOutput).contains(restoredCreateTableEvent);
+ }
+ for (int id = slowRecordId; id <= lastRecordId; id++) {
+ assertThat(restoredOutput).doesNotContain(tableAlphaInsertEvent(databaseName, id));
+ }
+ cancelJob(restoredJobId);
+ }
+
@ParameterizedTest(name = "batchMode: {0}")
@ValueSource(booleans = {true, false})
void testHeteroSchemaTransform(boolean batchMode) throws Exception {
@@ -125,7 +264,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -209,7 +348,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
testParallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -300,7 +439,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -382,7 +521,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -472,7 +611,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
testParallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -555,7 +694,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -641,7 +780,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
testParallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -722,7 +861,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -802,7 +941,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
if (batchMode) {
@@ -884,7 +1023,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
runtimeMode,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -919,7 +1058,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
transformTestDatabase.getDatabaseName(),
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -1014,7 +1153,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
transformTestDatabase.getDatabaseName(),
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -1117,7 +1256,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
transformTestDatabase.getDatabaseName(),
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -1250,7 +1389,7 @@ class TransformE2eITCase extends PipelineTestEnvironment {
transformTestDatabase.getDatabaseName(),
projectionExpression,
parallelism);
- JobID jobId = submitPipelineJob(pipelineJob);
+ JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob));
waitUntilJobRunning(Duration.ofSeconds(30));
LOG.info("Pipeline job is running");
@@ -1327,6 +1466,124 @@ class TransformE2eITCase extends PipelineTestEnvironment {
"DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[Beginning, 3010, 10, 10, 97, Lemon, ascii test!?, 大五, 测试数据, ひびぴ, 죠주쥬, ÀÆÉ, ÓÔŐÖ, αβγδε, בבקשה, твой, ภาษาไทย, piedzimst brīvi], op=INSERT, meta=()}");
}
+ private static void addRegionColumnToTableAlpha(String mysqlJdbcUrl) throws SQLException {
+ try (Connection conn =
+ DriverManager.getConnection(
+ mysqlJdbcUrl, MYSQL_TEST_USER, MYSQL_TEST_PASSWORD);
+ Statement stat = conn.createStatement()) {
+ stat.execute("ALTER TABLE TABLEALPHA ADD COLUMN REGION VARCHAR(17);");
+ }
+ }
+
+ private static void insertTableAlphaRows(
+ String mysqlJdbcUrl, int startInclusive, int endInclusive) throws SQLException {
+ try (Connection conn =
+ DriverManager.getConnection(
+ mysqlJdbcUrl, MYSQL_TEST_USER, MYSQL_TEST_PASSWORD);
+ Statement stat = conn.createStatement()) {
+ for (int id = startInclusive; id <= endInclusive; id++) {
+ stat.execute(
+ String.format(
+ "INSERT INTO TABLEALPHA(ID, VERSION, PRICEALPHA, AGEALPHA, NAMEALPHA) VALUES (%d, '%d', %d, %d, 'Bulk%d')",
+ id, id, id, id % 100, id));
+ }
+ }
+ }
+
+ private static String tableAlphaInsertEvent(String databaseName, int id) {
+ return String.format(
+ "DataChangeEvent{tableId=%s.TABLEALPHA, before=[], after=[%d, %d, %d, %d, Bulk%d, null, throttled_%d], op=INSERT, meta=()}",
+ databaseName, id, id, id, id % 100, id, id);
+ }
+
+ private void waitUntilSpecificEventsAfter(int offset, Duration timeout, String... events)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + timeout.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ String stdout = taskManagerConsumer.toUtf8String();
+ int searchOffset = Math.min(offset, stdout.length());
+ boolean matched = true;
+ for (String event : events) {
+ int eventOffset = stdout.indexOf(event, searchOffset);
+ if (eventOffset < 0) {
+ matched = false;
+ break;
+ }
+ searchOffset = eventOffset + event.length();
+ }
+ if (matched) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ throw new TimeoutException(
+ "Failed to get events after offset "
+ + offset
+ + ": "
+ + Arrays.toString(events)
+ + " from stdout: "
+ + taskManagerConsumer.toUtf8String());
+ }
+
+ private void waitUntilEventsAfter(int offset, Duration timeout, String... events)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + timeout.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ String stdout = taskManagerConsumer.toUtf8String();
+ int searchOffset = Math.min(offset, stdout.length());
+ boolean matched = true;
+ for (String event : events) {
+ if (stdout.indexOf(event, searchOffset) < 0) {
+ matched = false;
+ break;
+ }
+ }
+ if (matched) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ throw new TimeoutException(
+ "Failed to get events after offset "
+ + offset
+ + ": "
+ + Arrays.toString(events)
+ + " from stdout: "
+ + taskManagerConsumer.toUtf8String());
+ }
+
+ private void waitUntilEventAppearsBeforeAnother(
+ int offset, Duration timeout, String expectedEvent, String unexpectedEarlierEvent)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + timeout.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ String stdout = taskManagerConsumer.toUtf8String();
+ String outputAfterOffset = stdout.substring(Math.min(offset, stdout.length()));
+ int expectedOffset = outputAfterOffset.indexOf(expectedEvent);
+ int unexpectedOffset = outputAfterOffset.indexOf(unexpectedEarlierEvent);
+ if (unexpectedOffset >= 0
+ && (expectedOffset < 0 || unexpectedOffset < expectedOffset)) {
+ throw new AssertionError(
+ unexpectedEarlierEvent
+ + " appeared before "
+ + expectedEvent
+ + " in stdout: "
+ + stdout);
+ }
+ if (expectedOffset >= 0) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ throw new TimeoutException(
+ "Failed to get event after offset "
+ + offset
+ + ": "
+ + expectedEvent
+ + " from stdout: "
+ + taskManagerConsumer.toUtf8String());
+ }
+
private void validateEventsWithPattern(String... patterns) throws Exception {
for (String pattern : patterns) {
waitUntilSpecificEventWithPattern(
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
index f4db8e320..a86903f1a 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
@@ -446,6 +446,10 @@ public abstract class PipelineTestEnvironment extends TestLogger {
waitUntilJobState(timeout, JobStatus.RUNNING);
}
+ public void waitUntilJobRunning(JobID jobID, Duration timeout) {
+ waitUntilJobState(jobID, timeout, JobStatus.RUNNING);
+ }
+
public void waitUntilJobFinished(Duration timeout) {
waitUntilJobState(timeout, JobStatus.FINISHED);
}
@@ -478,6 +482,38 @@ public abstract class PipelineTestEnvironment extends TestLogger {
}
}
+ public void waitUntilJobState(JobID jobID, Duration timeout, JobStatus expectedStatus) {
+ RestClusterClient<?> clusterClient = getRestClusterClient();
+ Deadline deadline = Deadline.fromNow(timeout);
+ while (deadline.hasTimeLeft()) {
+ Collection<JobStatusMessage> jobStatusMessages;
+ try {
+ jobStatusMessages = clusterClient.listJobs().get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ LOG.warn("Error when fetching job status.", e);
+ continue;
+ }
+ if (jobStatusMessages != null && !jobStatusMessages.isEmpty()) {
+ for (JobStatusMessage message : jobStatusMessages) {
+ if (!jobID.equals(message.getJobId())) {
+ continue;
+ }
+ JobStatus jobStatus = message.getJobState();
+ if (!expectedStatus.isTerminalState() && jobStatus.isTerminalState()) {
+ throw new ValidationException(
+ String.format(
+ "Job has been terminated! JobName: %s, JobID: %s, Status: %s",
+ message.getJobName(),
+ message.getJobId(),
+ message.getJobState()));
+ } else if (jobStatus == expectedStatus) {
+ return;
+ }
+ }
+ }
+ }
+ }
+
protected String getFlinkDockerImageTag() {
if (System.getProperty("java.specification.version").equals("17")) {
return String.format("flink:%s-scala_2.12-java17", flinkVersion);
diff --git a/flink-cdc-flink1-compat/src/main/java/org/apache/flink/cdc/runtime/operators/AsyncWaitOperatorAdapter.java b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/cdc/runtime/operators/AsyncWaitOperatorAdapter.java
new file mode 100644
index 000000000..0b8dc06f6
--- /dev/null
+++ b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/cdc/runtime/operators/AsyncWaitOperatorAdapter.java
@@ -0,0 +1,61 @@
+/*
+ * 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.flink.cdc.runtime.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.streaming.api.datastream.AsyncDataStream;
+import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.async.AsyncWaitOperator;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.util.retryable.AsyncRetryStrategies;
+
+/** Flink 1.20 adapter for an ordered async operator with state-consistent checkpoints. */
+@Internal
+public class AsyncWaitOperatorAdapter<IN, OUT> extends AsyncWaitOperator<IN, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ @SuppressWarnings("unchecked")
+ public AsyncWaitOperatorAdapter(
+ StreamOperatorParameters<OUT> parameters,
+ AsyncFunction<IN, OUT> asyncFunction,
+ long timeout,
+ int capacity,
+ ProcessingTimeService processingTimeService,
+ MailboxExecutor mailboxExecutor) {
+ super(
+ asyncFunction,
+ timeout,
+ capacity,
+ AsyncDataStream.OutputMode.ORDERED,
+ AsyncRetryStrategies.NO_RETRY_STRATEGY,
+ processingTimeService,
+ mailboxExecutor);
+ setup(parameters.getContainingTask(), parameters.getStreamConfig(), parameters.getOutput());
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ // The async function owns schema state, so it must not advance beyond records retained in
+ // AsyncWaitOperator's recovery queue.
+ endInput();
+ super.prepareSnapshotPreBarrier(checkpointId);
+ }
+}
diff --git a/flink-cdc-flink2-compat/src/main/java/org/apache/flink/cdc/runtime/operators/AsyncWaitOperatorAdapter.java b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/cdc/runtime/operators/AsyncWaitOperatorAdapter.java
new file mode 100644
index 000000000..0e2a54e1f
--- /dev/null
+++ b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/cdc/runtime/operators/AsyncWaitOperatorAdapter.java
@@ -0,0 +1,61 @@
+/*
+ * 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.flink.cdc.runtime.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.streaming.api.datastream.AsyncDataStream;
+import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.async.AsyncWaitOperator;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.util.retryable.AsyncRetryStrategies;
+
+/** Flink 2.2 adapter for an ordered async operator with state-consistent checkpoints. */
+@Internal
+public class AsyncWaitOperatorAdapter<IN, OUT> extends AsyncWaitOperator<IN, OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ @SuppressWarnings("unchecked")
+ public AsyncWaitOperatorAdapter(
+ StreamOperatorParameters<OUT> parameters,
+ AsyncFunction<IN, OUT> asyncFunction,
+ long timeout,
+ int capacity,
+ ProcessingTimeService processingTimeService,
+ MailboxExecutor mailboxExecutor) {
+ super(
+ parameters,
+ asyncFunction,
+ timeout,
+ capacity,
+ AsyncDataStream.OutputMode.ORDERED,
+ AsyncRetryStrategies.NO_RETRY_STRATEGY,
+ processingTimeService,
+ mailboxExecutor);
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ // The async function owns schema state, so it must not advance beyond records retained in
+ // AsyncWaitOperator's recovery queue.
+ endInput();
+ super.prepareSnapshotPreBarrier(checkpointId);
+ }
+}
diff --git a/flink-cdc-pipeline-udf-examples/src/main/java/org/apache/flink/cdc/udf/examples/java/SkewedThrottlerFunctionClass.java b/flink-cdc-pipeline-udf-examples/src/main/java/org/apache/flink/cdc/udf/examples/java/SkewedThrottlerFunctionClass.java
new file mode 100644
index 000000000..034db7444
--- /dev/null
+++ b/flink-cdc-pipeline-udf-examples/src/main/java/org/apache/flink/cdc/udf/examples/java/SkewedThrottlerFunctionClass.java
@@ -0,0 +1,44 @@
+/*
+ * 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.flink.cdc.udf.examples.java;
+
+import org.apache.flink.cdc.common.udf.UserDefinedFunction;
+
+/** A UDF that makes one record deliberately slower than later records. */
+public class SkewedThrottlerFunctionClass implements UserDefinedFunction {
+
+ public String eval(Object value, int slowValue, int slowSeconds) {
+ if (asLong(value) == slowValue) {
+ try {
+ Thread.sleep(slowSeconds * 1000L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while throttling a record.", e);
+ }
+ }
+ System.out.println("SkewedThrottlerFunctionClass finished " + value);
+ return "throttled_" + value;
+ }
+
+ private long asLong(Object value) {
+ if (value instanceof Number) {
+ return ((Number) value).longValue();
+ }
+ return Long.parseLong(String.valueOf(value));
+ }
+}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunction.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunction.java
new file mode 100644
index 000000000..531fde01b
--- /dev/null
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunction.java
@@ -0,0 +1,959 @@
+/*
+ * 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.flink.cdc.runtime.operators.transform.async;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.converter.JavaObjectConverter;
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.event.ChangeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.schema.Selectors;
+import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
+import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.common.utils.SchemaUtils;
+import org.apache.flink.cdc.runtime.operators.transform.PostTransformChangeInfo;
+import org.apache.flink.cdc.runtime.operators.transform.PostTransformer;
+import org.apache.flink.cdc.runtime.operators.transform.ProjectionColumn;
+import org.apache.flink.cdc.runtime.operators.transform.TransformContext;
+import org.apache.flink.cdc.runtime.operators.transform.TransformExpressionCompiler;
+import org.apache.flink.cdc.runtime.operators.transform.TransformFilter;
+import org.apache.flink.cdc.runtime.operators.transform.TransformFilterProcessor;
+import org.apache.flink.cdc.runtime.operators.transform.TransformProjection;
+import org.apache.flink.cdc.runtime.operators.transform.TransformProjectionProcessor;
+import org.apache.flink.cdc.runtime.operators.transform.TransformRule;
+import org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor;
+import org.apache.flink.cdc.runtime.operators.transform.converter.PostTransformConverters;
+import org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException;
+import org.apache.flink.cdc.runtime.parser.TransformParser;
+import org.apache.flink.cdc.runtime.serializer.TableIdSerializer;
+import org.apache.flink.cdc.runtime.serializer.event.CreateTableEventSerializer;
+import org.apache.flink.cdc.runtime.serializer.schema.SchemaSerializer;
+import org.apache.flink.cdc.runtime.typeutils.BinaryInternalObjectConverter;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder;
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheLoader;
+import org.apache.flink.shaded.guava31.com.google.common.cache.LoadingCache;
+import org.apache.flink.shaded.guava31.com.google.common.collect.HashBasedTable;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Table;
+import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.cdc.common.utils.Preconditions.checkNotNull;
+
+/**
+ * An async post-transform function for ordered async execution.
+ *
+ * <p>{@link SchemaChangeEvent}s are handled as barriers. The function waits for all pending {@link
+ * DataChangeEvent} futures submitted before the schema change, applies the schema change, and only
+ * then allows following data changes to run against the updated schema.
+ */
+public class AsyncPostTransformFunction extends RichAsyncFunction<Event, Event>
+ implements CheckpointedFunction, Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final String TABLE_STATE_NAME = "async-post-transform-table-state";
+ private static final long EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS = 30L;
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncPostTransformFunction.class);
+
+ private static final int TABLE_STATE_VERSION = 1;
+
+ private final String timezone;
+ private final DecimalPrecisionMode decimalPrecisionMode;
+ private final List<TransformRule> transformRules;
+ private final Map<TableId, PostTransformTableInfo> tableInfoMap;
+
+ // Tuple3 items are: function name, class path, and extra options.
+ private final List<Tuple3<String, String, Map<String, String>>> udfFunctions;
+
+ // Serializable AI model clients keyed by model name, e.g. myModel.
+ private final Map<String, AiModelClient> modelClients;
+
+ private transient List<PostTransformer> transformers;
+ private transient List<UserDefinedFunctionDescriptor> udfDescriptors;
+ private transient List<Object> udfFunctionInstances;
+ private transient ThreadLocal<Table<TableId, PostTransformer, TransformProjectionProcessor>>
+ projectionProcessors;
+ private transient ThreadLocal<Table<TableId, PostTransformer, TransformFilterProcessor>>
+ filterProcessors;
+ private transient Queue<Table<TableId, PostTransformer, TransformProjectionProcessor>>
+ projectionProcessorCaches;
+ private transient Queue<Table<TableId, PostTransformer, TransformFilterProcessor>>
+ filterProcessorCaches;
+ private transient LoadingCache<TableId, Optional<PostTransformer>> transformersCache;
+
+ private final int asyncWorkerThreads;
+
+ private transient ExecutorService executorService;
+ private transient Set<CompletableFuture<List<Event>>> pendingDataFutures;
+ private transient CompletableFuture<Void> schemaBarrierFuture;
+ private transient ListState<byte[]> tableState;
+ private transient Set<TableId> emittedCreateTableEventTables;
+
+ public static AsyncPostTransformFunctionBuilder newBuilder() {
+ return new AsyncPostTransformFunctionBuilder();
+ }
+
+ AsyncPostTransformFunction(
+ List<TransformRule> transformRules,
+ String timezone,
+ DecimalPrecisionMode decimalPrecisionMode,
+ List<Tuple3<String, String, Map<String, String>>> udfFunctions,
+ Map<String, AiModelClient> modelClients,
+ int asyncWorkerThreads) {
+ Preconditions.checkArgument(
+ asyncWorkerThreads > 0, "Async worker threads must be greater than 0.");
+ this.timezone = timezone;
+ this.decimalPrecisionMode = decimalPrecisionMode;
+ this.transformRules = transformRules;
+ this.tableInfoMap = new ConcurrentHashMap<>();
+ this.udfFunctions = udfFunctions;
+ this.modelClients = modelClients;
+ this.asyncWorkerThreads = asyncWorkerThreads;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+ this.pendingDataFutures = ConcurrentHashMap.newKeySet();
+ this.schemaBarrierFuture = CompletableFuture.completedFuture(null);
+ this.emittedCreateTableEventTables = ConcurrentHashMap.newKeySet();
+
+ this.projectionProcessorCaches = new ConcurrentLinkedQueue<>();
+ this.filterProcessorCaches = new ConcurrentLinkedQueue<>();
+ this.projectionProcessors =
+ ThreadLocal.withInitial(
+ () -> {
+ Table<TableId, PostTransformer, TransformProjectionProcessor>
+ processors = HashBasedTable.create();
+ projectionProcessorCaches.add(processors);
+ return processors;
+ });
+ this.filterProcessors =
+ ThreadLocal.withInitial(
+ () -> {
+ Table<TableId, PostTransformer, TransformFilterProcessor> processors =
+ HashBasedTable.create();
+ filterProcessorCaches.add(processors);
+ return processors;
+ });
+
+ initializeAiModelClients();
+ initializeUdf();
+
+ this.transformers = createTransformers();
+ this.transformersCache =
+ CacheBuilder.newBuilder()
+ .maximumSize(1024)
+ .build(
+ new CacheLoader<>() {
+ @Override
+ public Optional<PostTransformer> load(TableId tableId) {
+ return getEffectiveTransformer(tableId);
+ }
+ });
+ this.executorService =
+ Executors.newFixedThreadPool(
+ asyncWorkerThreads,
+ new ThreadFactoryBuilder()
+ .setNameFormat(
+ "post-transform-async-"
+ + getRuntimeContext()
+ .getTaskInfo()
+ .getIndexOfThisSubtask()
+ + "-%d")
+ .build());
+ }
+
+ @Override
+ public void close() throws Exception {
+ try {
+ boolean executorTerminated = true;
+ if (executorService != null) {
+ executorService.shutdownNow();
+ executorTerminated =
+ executorService.awaitTermination(
+ EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
+ if (executorTerminated) {
+ TransformExpressionCompiler.cleanUp();
+ destroyUdf();
+ destroyAiModelClients();
+ if (transformersCache != null) {
+ transformersCache.invalidateAll();
+ }
+ } else {
+ LOG.warn(
+ "Async post-transform workers did not terminate within {} seconds; "
+ + "processor resources will remain open to avoid concurrent close.",
+ EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS);
+ }
+ } finally {
+ super.close();
+ }
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ tableState.clear();
+ for (byte[] serializedTableState : serializeTableStates()) {
+ tableState.add(serializedTableState);
+ }
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ tableState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>(TABLE_STATE_NAME, byte[].class));
+ if (context.isRestored()) {
+ for (byte[] serializedTableState : tableState.get()) {
+ restoreTableState(serializedTableState);
+ }
+ }
+ }
+
+ @Override
+ public void asyncInvoke(Event event, ResultFuture<Event> resultFuture) {
+ if (event instanceof DataChangeEvent) {
+ // Resolve a restored CreateTableEvent on the mailbox thread so ORDERED output
+ // deterministically prepends it to the first data event for the table.
+ TableId tableId = ((DataChangeEvent) event).tableId();
+ List<Event> prependedEvents = prependCreateTableEventIfNeeded(tableId);
+ asyncInvokeDataChangeEvent(event, resultFuture, prependedEvents);
+ } else {
+ asyncInvokeBarrierEvent(event, resultFuture);
+ }
+ }
+
+ @Override
+ public void timeout(Event event, ResultFuture<Event> resultFuture) {
+ resultFuture.completeExceptionally(
+ new FlinkRuntimeException("Async post-transform timed out for event: " + event));
+ }
+
+ private void asyncInvokeDataChangeEvent(
+ Event event, ResultFuture<Event> resultFuture, List<Event> prependedEvents) {
+ CompletableFuture<List<Event>> dataFuture =
+ schemaBarrierFuture.thenCompose(
+ ignored ->
+ CompletableFuture.supplyAsync(
+ () -> processSafely(event), executorService));
+ pendingDataFutures.add(dataFuture);
+ dataFuture.whenComplete(
+ (result, error) -> {
+ pendingDataFutures.remove(dataFuture);
+ if (error != null) {
+ completeResultFuture(event, resultFuture, null, error);
+ } else {
+ completeResultFuture(
+ event, resultFuture, prependEvents(prependedEvents, result), null);
+ }
+ });
+ }
+
+ private void asyncInvokeBarrierEvent(Event event, ResultFuture<Event> resultFuture) {
+ CompletableFuture<Void> previousSchemaBarrierFuture = schemaBarrierFuture;
+ CompletableFuture<Void> previousDataFutures = waitForPendingDataFutures();
+ CompletableFuture<List<Event>> schemaFuture =
+ CompletableFuture.allOf(previousSchemaBarrierFuture, previousDataFutures)
+ .thenApply(ignored -> processSafely(event));
+ schemaBarrierFuture = schemaFuture.thenApply(ignored -> null);
+ schemaFuture.whenComplete(
+ (result, error) -> completeResultFuture(event, resultFuture, result, error));
+ }
+
+ private CompletableFuture<Void> waitForPendingDataFutures() {
+ CompletableFuture<?>[] futures = pendingDataFutures.toArray(new CompletableFuture<?>[0]);
+ return CompletableFuture.allOf(futures);
+ }
+
+ private List<Event> processSafely(Event event) {
+ try {
+ Optional<Event> result = process(event);
+ return result.map(Collections::singletonList).orElseGet(Collections::emptyList);
+ } catch (Exception e) {
+ throw wrapTransformException("async post-transform", event, e);
+ }
+ }
+
+ private void completeResultFuture(
+ Event event,
+ ResultFuture<Event> resultFuture,
+ @Nullable List<Event> result,
+ @Nullable Throwable error) {
+ if (error != null) {
+ resultFuture.completeExceptionally(
+ wrapTransformException("async post-transform", event, error));
+ } else {
+ resultFuture.complete(result);
+ }
+ }
+
+ private List<Event> prependCreateTableEventIfNeeded(TableId tableId) {
+ if (!emittedCreateTableEventTables.add(tableId)) {
+ return Collections.emptyList();
+ }
+
+ CreateTableEvent outputCreateTableEvent = getOutputCreateTableEvent(tableId);
+ if (outputCreateTableEvent == null) {
+ emittedCreateTableEventTables.remove(tableId);
+ return Collections.emptyList();
+ }
+ return Collections.singletonList(outputCreateTableEvent);
+ }
+
+ private static List<Event> prependEvents(
+ List<Event> prependedEvents, @Nullable List<Event> result) {
+ if (prependedEvents.isEmpty()) {
+ return result;
+ }
+ List<Event> output = new ArrayList<>(prependedEvents);
+ if (result != null) {
+ output.addAll(result);
+ }
+ return output;
+ }
+
+ Optional<Event> process(Event event) {
+ if (event == null) {
+ return Optional.empty();
+ }
+
+ if (!(event instanceof ChangeEvent)) {
+ throw new UnsupportedOperationException("Unexpected stream record event: " + event);
+ }
+
+ ChangeEvent changeEvent = (ChangeEvent) event;
+ TableId tableId = changeEvent.tableId();
+ Optional<PostTransformer> transformer = transformersCache.getUnchecked(tableId);
+
+ if (transformer.isEmpty()) {
+ cachePassthroughSchemaEvent(event);
+ if (event instanceof SchemaChangeEvent) {
+ emittedCreateTableEventTables.add(tableId);
+ }
+ return Optional.of(event);
+ }
+
+ if (event instanceof CreateTableEvent) {
+ Optional<Event> result =
+ processCreateTableEvent((CreateTableEvent) event, transformer.get());
+ emittedCreateTableEventTables.add(tableId);
+ invalidateCache(tableId);
+ return result;
+ } else if (event instanceof SchemaChangeEvent) {
+ Optional<Event> result =
+ processSchemaChangeEvent((SchemaChangeEvent) event, transformer.get());
+ emittedCreateTableEventTables.add(tableId);
+ invalidateCache(tableId);
+ return result;
+ } else if (event instanceof DataChangeEvent) {
+ return processDataChangeEvent((DataChangeEvent) event, transformer.get());
+ } else {
+ throw new UnsupportedOperationException("Unexpected stream record event: " + event);
+ }
+ }
+
+ TransformException wrapTransformException(String command, Event event, Throwable throwable) {
+ Throwable cause = throwable;
+ if (cause instanceof CompletionException && cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ if (cause instanceof TransformException) {
+ return (TransformException) cause;
+ }
+
+ TableId tableId = null;
+ Schema schemaBefore = null;
+ Schema schemaAfter = null;
+ if (event instanceof ChangeEvent) {
+ tableId = ((ChangeEvent) event).tableId();
+ PostTransformTableInfo tableInfo = tableInfoMap.get(tableId);
+ if (tableInfo != null) {
+ schemaBefore = tableInfo.changeInfo.getPreTransformedSchema();
+ schemaAfter = tableInfo.changeInfo.getPostTransformedSchema();
+ }
+ }
+ return new TransformException(command, event, tableId, schemaBefore, schemaAfter, cause);
+ }
+
+ @Nullable
+ CreateTableEvent getOutputCreateTableEvent(TableId tableId) {
+ PostTransformTableInfo tableInfo = tableInfoMap.get(tableId);
+ return tableInfo == null ? null : tableInfo.outputCreateTableEvent;
+ }
+
+ List<byte[]> serializeTableStates() throws IOException {
+ List<byte[]> result = new ArrayList<>(tableInfoMap.size());
+ for (PostTransformTableInfo tableInfo : tableInfoMap.values()) {
+ result.add(serializeTableState(tableInfo));
+ }
+ return result;
+ }
+
+ void restoreTableState(byte[] serializedTableState) throws IOException {
+ TableIdSerializer tableIdSerializer = TableIdSerializer.INSTANCE;
+ SchemaSerializer schemaSerializer = SchemaSerializer.INSTANCE;
+ CreateTableEventSerializer createTableEventSerializer = CreateTableEventSerializer.INSTANCE;
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(serializedTableState);
+ DataInputStream in = new DataInputStream(bais)) {
+ int version = in.readInt();
+ if (version != TABLE_STATE_VERSION) {
+ throw new IOException(
+ "Unrecognized async post-transform table state version " + version);
+ }
+ TableId tableId = tableIdSerializer.deserialize(new DataInputViewStreamWrapper(in));
+ Schema preTransformedSchema =
+ schemaSerializer.deserialize(new DataInputViewStreamWrapper(in));
+ Schema postTransformedSchema =
+ schemaSerializer.deserialize(new DataInputViewStreamWrapper(in));
+ CreateTableEvent outputCreateTableEvent = null;
+ if (in.readBoolean()) {
+ outputCreateTableEvent =
+ createTableEventSerializer.deserialize(new DataInputViewStreamWrapper(in));
+ }
+ cacheTableState(
+ tableId, preTransformedSchema, postTransformedSchema, outputCreateTableEvent);
+ }
+ }
+
+ private byte[] serializeTableState(PostTransformTableInfo tableInfo) throws IOException {
+ TableIdSerializer tableIdSerializer = TableIdSerializer.INSTANCE;
+ SchemaSerializer schemaSerializer = SchemaSerializer.INSTANCE;
+ CreateTableEventSerializer createTableEventSerializer = CreateTableEventSerializer.INSTANCE;
+ PostTransformChangeInfo changeInfo = tableInfo.changeInfo;
+
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ out.writeInt(TABLE_STATE_VERSION);
+ tableIdSerializer.serialize(
+ changeInfo.getTableId(), new DataOutputViewStreamWrapper(out));
+ schemaSerializer.serialize(
+ changeInfo.getPreTransformedSchema(), new DataOutputViewStreamWrapper(out));
+ schemaSerializer.serialize(
+ changeInfo.getPostTransformedSchema(), new DataOutputViewStreamWrapper(out));
+ out.writeBoolean(tableInfo.outputCreateTableEvent != null);
+ if (tableInfo.outputCreateTableEvent != null) {
+ createTableEventSerializer.serialize(
+ tableInfo.outputCreateTableEvent, new DataOutputViewStreamWrapper(out));
+ }
+ return baos.toByteArray();
+ }
+ }
+
+ private void cachePassthroughSchemaEvent(Event event) {
+ if (event instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) event;
+ cacheTableState(
+ createTableEvent.tableId(),
+ createTableEvent.getSchema(),
+ createTableEvent.getSchema(),
+ createTableEvent);
+ } else if (event instanceof SchemaChangeEvent) {
+ SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
+ PostTransformTableInfo tableInfo = tableInfoMap.get(schemaChangeEvent.tableId());
+ if (tableInfo != null) {
+ Schema nextSchema =
+ SchemaUtils.applySchemaChangeEvent(
+ tableInfo.changeInfo.getPreTransformedSchema(), schemaChangeEvent);
+ CreateTableEvent nextOutputCreateTableEvent =
+ applySchemaChangeEventToOutputCreateTableEvent(
+ tableInfo.outputCreateTableEvent, schemaChangeEvent);
+ cacheTableState(
+ schemaChangeEvent.tableId(),
+ nextSchema,
+ nextSchema,
+ tableInfo.hasAsterisk,
+ nextOutputCreateTableEvent);
+ }
+ }
+ }
+
+ private Optional<Event> processCreateTableEvent(
+ CreateTableEvent event, PostTransformer effectiveTransformer) {
+ TableId tableId = event.tableId();
+ Schema preSchema = event.getSchema();
+ Schema postSchema =
+ SchemaUtils.ensurePkNonNull(transformSchema(preSchema, effectiveTransformer));
+ CreateTableEvent outputCreateTableEvent = new CreateTableEvent(tableId, postSchema);
+
+ cacheTableState(
+ tableId,
+ preSchema,
+ postSchema,
+ hasAsterisk(effectiveTransformer),
+ outputCreateTableEvent);
+ return Optional.of(outputCreateTableEvent);
+ }
+
+ private Optional<Event> processSchemaChangeEvent(
+ SchemaChangeEvent event, PostTransformer effectiveTransformer) {
+ TableId tableId = event.tableId();
+ PostTransformTableInfo tableInfo = checkNotNull(tableInfoMap.get(tableId));
+ PostTransformChangeInfo info = tableInfo.changeInfo;
+
+ Schema prevPreSchema = info.getPreTransformedSchema();
+ Schema nextPreSchema = SchemaUtils.applySchemaChangeEvent(prevPreSchema, event);
+ Schema nextPostSchema =
+ SchemaUtils.ensurePkNonNull(transformSchema(nextPreSchema, effectiveTransformer));
+
+ Schema prevPostSchema = info.getPostTransformedSchema();
+ List<String> columnNamesBeforeChange = prevPostSchema.getColumnNames();
+ Optional<SchemaChangeEvent> outputEvent;
+ if (tableInfo.hasAsterisk) {
+ // See comments in PreTransformOperator#cacheChangeSchema method.
+ outputEvent =
+ SchemaUtils.transformSchemaChangeEvent(true, columnNamesBeforeChange, event);
+ } else {
+ outputEvent =
+ SchemaUtils.transformSchemaChangeEvent(
+ false, tableInfo.projectedColumns, event);
+ }
+
+ CreateTableEvent nextOutputCreateTableEvent =
+ outputEvent
+ .map(
+ transformedEvent ->
+ applySchemaChangeEventToOutputCreateTableEvent(
+ tableInfo.outputCreateTableEvent, transformedEvent))
+ .orElse(tableInfo.outputCreateTableEvent);
+ cacheTableState(
+ tableId,
+ nextPreSchema,
+ nextPostSchema,
+ tableInfo.hasAsterisk,
+ nextOutputCreateTableEvent);
+ return outputEvent.map(Event.class::cast);
+ }
+
+ @Nullable
+ private CreateTableEvent applySchemaChangeEventToOutputCreateTableEvent(
+ @Nullable CreateTableEvent outputCreateTableEvent, SchemaChangeEvent event) {
+ if (outputCreateTableEvent == null) {
+ return null;
+ }
+ Schema schema =
+ SchemaUtils.applySchemaChangeEvent(outputCreateTableEvent.getSchema(), event);
+ return new CreateTableEvent(event.tableId(), schema);
+ }
+
+ private Optional<Event> processDataChangeEvent(
+ DataChangeEvent event, PostTransformer effectiveTransformer) {
+ TableId tableId = event.tableId();
+ PostTransformChangeInfo info = checkNotNull(tableInfoMap.get(tableId)).changeInfo;
+
+ TransformContext context = new TransformContext();
+ context.epochTime = System.currentTimeMillis();
+ context.meta = event.meta();
+
+ String beforeOp = event.opTypeString(false);
+ String afterOp = event.opTypeString(true);
+ TransformProjectionProcessor projectionProcessor =
+ getProjectionProcessor(tableId, effectiveTransformer);
+ TransformFilterProcessor filterProcessor =
+ getFilterProcessor(tableId, effectiveTransformer);
+
+ BinaryRecordData beforeRow = null;
+ BinaryRecordData afterRow = null;
+ boolean beforeFilterPassed = false;
+ boolean afterFilterPassed = false;
+
+ if (event.before() != null) {
+ context.opType = beforeOp;
+ Tuple2<BinaryRecordData, Boolean> result =
+ transformRecord(
+ event.before(), info, projectionProcessor, filterProcessor, context);
+ beforeRow = result.f0;
+ beforeFilterPassed = result.f1;
+ }
+ if (event.after() != null) {
+ context.opType = afterOp;
+ Tuple2<BinaryRecordData, Boolean> result =
+ transformRecord(
+ event.after(), info, projectionProcessor, filterProcessor, context);
+ afterRow = result.f0;
+ afterFilterPassed = result.f1;
+ }
+
+ DataChangeEvent finalEvent;
+ switch (event.op()) {
+ case INSERT:
+ case REPLACE:
+ if (!afterFilterPassed) {
+ return Optional.empty();
+ }
+ finalEvent = DataChangeEvent.projectRecords(event, beforeRow, afterRow);
+ break;
+ case DELETE:
+ if (!beforeFilterPassed) {
+ return Optional.empty();
+ }
+ finalEvent = DataChangeEvent.projectRecords(event, beforeRow, afterRow);
+ break;
+ case UPDATE:
+ if (beforeFilterPassed && afterFilterPassed) {
+ finalEvent = DataChangeEvent.projectRecords(event, beforeRow, afterRow);
+ } else if (beforeFilterPassed) {
+ finalEvent = DataChangeEvent.deleteEvent(tableId, beforeRow, event.meta());
+ } else if (afterFilterPassed) {
+ finalEvent = DataChangeEvent.insertEvent(tableId, afterRow, event.meta());
+ } else {
+ return Optional.empty();
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported operation type: " + event.op());
+ }
+
+ if (effectiveTransformer.getPostTransformConverter().isPresent()) {
+ return effectiveTransformer
+ .getPostTransformConverter()
+ .get()
+ .convert(finalEvent)
+ .map(Event.class::cast);
+ }
+ return Optional.of(finalEvent);
+ }
+
+ private Schema transformSchema(Schema preSchema, PostTransformer transformer) {
+ List<ProjectionColumn> projectionColumns =
+ TransformParser.generateProjectionColumns(
+ transformer
+ .getProjection()
+ .map(TransformProjection::getProjection)
+ .orElse(null),
+ preSchema.getColumns(),
+ udfDescriptors,
+ transformer.getSupportedMetadataColumns(),
+ decimalPrecisionMode);
+ return preSchema.copy(
+ projectionColumns.stream()
+ .map(ProjectionColumn::getColumn)
+ .collect(Collectors.toList()));
+ }
+
+ private Tuple2<BinaryRecordData, Boolean> transformRecord(
+ RecordData recordData,
+ PostTransformChangeInfo info,
+ @Nullable TransformProjectionProcessor projectionProcessor,
+ @Nullable TransformFilterProcessor filterProcessor,
+ TransformContext context) {
+ RecordData.FieldGetter[] preFieldGetters = info.getPreTransformedFieldGetters();
+ Schema preSchema = info.getPreTransformedSchema();
+ Schema postSchema = info.getPostTransformedSchema();
+ BinaryRecordDataGenerator postGenerator = info.getPostTransformedRecordDataGenerator();
+
+ Object[] preRow = new Object[preFieldGetters.length];
+ for (int i = 0; i < preFieldGetters.length; i++) {
+ preRow[i] =
+ JavaObjectConverter.convertToJava(
+ preFieldGetters[i].getFieldOrNull(recordData),
+ preSchema.getColumnDataTypes().get(i));
+ }
+
+ Object[] postRow =
+ projectionProcessor != null ? projectionProcessor.project(preRow, context) : preRow;
+ boolean filterPassed =
+ filterProcessor == null || filterProcessor.test(preRow, postRow, context);
+
+ Object[] postRowBinary = new Object[postSchema.getColumnCount()];
+ for (int i = 0; i < postRow.length; i++) {
+ postRowBinary[i] =
+ BinaryInternalObjectConverter.convertToInternal(
+ postRow[i], postSchema.getColumnDataTypes().get(i));
+ }
+ synchronized (postGenerator) {
+ return Tuple2.of(postGenerator.generate(postRowBinary), filterPassed);
+ }
+ }
+
+ private Optional<PostTransformer> getEffectiveTransformer(TableId tableId) {
+ for (PostTransformer transformer : transformers) {
+ if (transformer.getSelectors().isMatch(tableId)) {
+ return Optional.of(transformer);
+ }
+ }
+ return Optional.empty();
+ }
+
+ private TransformProjectionProcessor getProjectionProcessor(
+ TableId tableId, PostTransformer postTransformer) {
+ Table<TableId, PostTransformer, TransformProjectionProcessor> processors =
+ projectionProcessors.get();
+ if (!processors.contains(tableId, postTransformer)) {
+ PostTransformChangeInfo changeInfo = checkNotNull(tableInfoMap.get(tableId)).changeInfo;
+ processors.put(
+ tableId,
+ postTransformer,
+ new TransformProjectionProcessor(
+ changeInfo,
+ postTransformer
+ .getProjection()
+ .map(TransformProjection::getProjection)
+ .orElse(null),
+ timezone,
+ decimalPrecisionMode,
+ udfDescriptors,
+ udfFunctionInstances,
+ postTransformer.getSupportedMetadataColumns(),
+ modelClients));
+ }
+ return processors.get(tableId, postTransformer);
+ }
+
+ private TransformFilterProcessor getFilterProcessor(
+ TableId tableId, PostTransformer postTransformer) {
+ Table<TableId, PostTransformer, TransformFilterProcessor> processors =
+ filterProcessors.get();
+ if (!processors.contains(tableId, postTransformer)) {
+ if (!postTransformer.getFilter().isPresent()) {
+ processors.put(
+ tableId,
+ postTransformer,
+ TransformFilterProcessor.ofNoOp(decimalPrecisionMode));
+ } else {
+ PostTransformChangeInfo changeInfo =
+ checkNotNull(tableInfoMap.get(tableId)).changeInfo;
+ processors.put(
+ tableId,
+ postTransformer,
+ TransformFilterProcessor.of(
+ changeInfo,
+ postTransformer.getFilter().orElse(null),
+ timezone,
+ decimalPrecisionMode,
+ udfDescriptors,
+ udfFunctionInstances,
+ postTransformer.getSupportedMetadataColumns(),
+ modelClients));
+ }
+ }
+ return processors.get(tableId, postTransformer);
+ }
+
+ private void invalidateCache(TableId tableId) {
+ projectionProcessorCaches.forEach(processors -> processors.row(tableId).clear());
+ filterProcessorCaches.forEach(processors -> processors.row(tableId).clear());
+ }
+
+ private List<PostTransformer> createTransformers() {
+ List<PostTransformer> list = new ArrayList<>();
+ for (TransformRule rule : transformRules) {
+ Selectors selectors =
+ new Selectors.SelectorsBuilder()
+ .includeTables(rule.getTableInclusions())
+ .build();
+ list.add(
+ new PostTransformer(
+ selectors,
+ TransformProjection.of(rule.getProjection()).orElse(null),
+ TransformFilter.of(rule.getFilter()).orElse(null),
+ PostTransformConverters.of(rule.getPostTransformConverter())
+ .orElse(null),
+ rule.getSupportedMetadataColumns()));
+ }
+ return list;
+ }
+
+ private void initializeUdf() {
+ this.udfDescriptors =
+ udfFunctions.stream()
+ .map(UserDefinedFunctionDescriptor::new)
+ .collect(Collectors.toList());
+ this.udfFunctionInstances = new ArrayList<>();
+
+ for (UserDefinedFunctionDescriptor udf : udfDescriptors) {
+ try {
+ Class<?> clazz = Class.forName(udf.getClasspath());
+ Object udfInstance = clazz.getDeclaredConstructor().newInstance();
+ udfFunctionInstances.add(udfInstance);
+
+ if (udf.isCdcPipelineUdf()) {
+ UserDefinedFunctionContext userDefinedFunctionContext =
+ () -> Configuration.fromMap(udf.getParameters());
+ udfInstance
+ .getClass()
+ .getMethod("open", UserDefinedFunctionContext.class)
+ .invoke(udfInstance, userDefinedFunctionContext);
+ }
+ } catch (ReflectiveOperationException e) {
+ throw new RuntimeException("Failed to instantiate UDF function " + udf, e);
+ }
+ }
+ }
+
+ private void destroyUdf() {
+ if (udfDescriptors == null || udfFunctionInstances == null) {
+ return;
+ }
+ for (int i = 0; i < udfDescriptors.size(); i++) {
+ UserDefinedFunctionDescriptor udf = udfDescriptors.get(i);
+ try {
+ if (udf.isCdcPipelineUdf()) {
+ Object udfInstance = udfFunctionInstances.get(i);
+ udfInstance.getClass().getMethod("close").invoke(udfInstance);
+ }
+ } catch (ReflectiveOperationException e) {
+ throw new RuntimeException("Failed to destroy UDF " + udf, e);
+ }
+ }
+ udfDescriptors.clear();
+ udfFunctionInstances.clear();
+ }
+
+ private void initializeAiModelClients() {
+ for (Map.Entry<String, AiModelClient> entry : modelClients.entrySet()) {
+ try {
+ entry.getValue().open();
+ LOG.info("Successfully opened AI model client '{}'.", entry.getKey());
+ } catch (Exception e) {
+ LOG.error("Failed to open AI model client '{}'.", entry.getKey(), e);
+ throw new FlinkRuntimeException(
+ "Failed to initialize AI model: " + entry.getKey(), e);
+ }
+ }
+ }
+
+ private void destroyAiModelClients() {
+ for (Map.Entry<String, AiModelClient> entry : modelClients.entrySet()) {
+ try {
+ entry.getValue().close();
+ LOG.info("Successfully closed AI model client '{}'.", entry.getKey());
+ } catch (Exception e) {
+ LOG.warn("Failed to close AI model client '{}'.", entry.getKey(), e);
+ }
+ }
+ }
+
+ private void cacheTableState(
+ TableId tableId,
+ Schema preSchema,
+ Schema postSchema,
+ @Nullable CreateTableEvent outputCreateTableEvent) {
+ cacheTableState(
+ tableId, preSchema, postSchema, hasAsterisk(tableId), outputCreateTableEvent);
+ }
+
+ private void cacheTableState(
+ TableId tableId,
+ Schema preSchema,
+ Schema postSchema,
+ boolean hasAsterisk,
+ @Nullable CreateTableEvent outputCreateTableEvent) {
+ tableInfoMap.put(
+ tableId,
+ new PostTransformTableInfo(
+ PostTransformChangeInfo.of(tableId, preSchema, postSchema),
+ outputCreateTableEvent,
+ hasAsterisk,
+ projectedColumns(preSchema, postSchema)));
+ }
+
+ private boolean hasAsterisk(TableId tableId) {
+ for (TransformRule rule : transformRules) {
+ Selectors selectors =
+ new Selectors.SelectorsBuilder()
+ .includeTables(rule.getTableInclusions())
+ .build();
+ if (selectors.isMatch(tableId)) {
+ return rule.getProjection() != null
+ && TransformParser.hasAsterisk(rule.getProjection());
+ }
+ }
+ return false;
+ }
+
+ private boolean hasAsterisk(PostTransformer transformer) {
+ return transformer.getProjection().isPresent()
+ && TransformParser.hasAsterisk(transformer.getProjection().get().getProjection());
+ }
+
+ private List<String> projectedColumns(Schema preSchema, Schema postSchema) {
+ return preSchema.getColumnNames().stream()
+ .filter(postSchema.getColumnNames()::contains)
+ .collect(Collectors.toList());
+ }
+
+ private static final class PostTransformTableInfo {
+
+ private final PostTransformChangeInfo changeInfo;
+ @Nullable private final CreateTableEvent outputCreateTableEvent;
+ private final boolean hasAsterisk;
+ private final List<String> projectedColumns;
+
+ private PostTransformTableInfo(
+ PostTransformChangeInfo changeInfo,
+ @Nullable CreateTableEvent outputCreateTableEvent,
+ boolean hasAsterisk,
+ List<String> projectedColumns) {
+ this.changeInfo = changeInfo;
+ this.outputCreateTableEvent = outputCreateTableEvent;
+ this.hasAsterisk = hasAsterisk;
+ this.projectedColumns = Collections.unmodifiableList(new ArrayList<>(projectedColumns));
+ }
+ }
+}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunctionBuilder.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunctionBuilder.java
new file mode 100644
index 000000000..affbe9cd7
--- /dev/null
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunctionBuilder.java
@@ -0,0 +1,131 @@
+/*
+ * 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.flink.cdc.runtime.operators.transform.async;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.runtime.operators.transform.TransformRule;
+
+import javax.annotation.Nullable;
+
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Builder of {@link AsyncPostTransformFunction}. */
+public class AsyncPostTransformFunctionBuilder {
+
+ private final List<TransformRule> transformRules = new ArrayList<>();
+ private String timezone;
+ private DecimalPrecisionMode decimalPrecisionMode = DecimalPrecisionMode.UP_TO_19;
+ private final List<Tuple3<String, String, Map<String, String>>> udfFunctions =
+ new ArrayList<>();
+ private final Map<String, AiModelClient> modelClients = new LinkedHashMap<>();
+ private int asyncWorkerThreads = 16;
+
+ public AsyncPostTransformFunctionBuilder addTransform(
+ String tableInclusions,
+ @Nullable String projection,
+ @Nullable String filter,
+ String primaryKey,
+ String partitionKey,
+ String tableOptions,
+ String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ return addTransform(
+ tableInclusions,
+ projection,
+ filter,
+ primaryKey,
+ partitionKey,
+ tableOptions,
+ ",",
+ postTransformConverter,
+ supportedMetadataColumns);
+ }
+
+ public AsyncPostTransformFunctionBuilder addTransform(
+ String tableInclusions,
+ @Nullable String projection,
+ @Nullable String filter,
+ String primaryKey,
+ String partitionKey,
+ String tableOptions,
+ String tableOptionsDelimiter,
+ String postTransformConverter,
+ SupportedMetadataColumn[] supportedMetadataColumns) {
+ transformRules.add(
+ new TransformRule(
+ tableInclusions,
+ projection,
+ filter,
+ primaryKey,
+ partitionKey,
+ tableOptions,
+ tableOptionsDelimiter,
+ postTransformConverter,
+ supportedMetadataColumns));
+ return this;
+ }
+
+ public AsyncPostTransformFunctionBuilder addTimezone(String timezone) {
+ if (PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue().equals(timezone)) {
+ this.timezone = ZoneId.systemDefault().toString();
+ } else {
+ this.timezone = timezone;
+ }
+ return this;
+ }
+
+ public AsyncPostTransformFunctionBuilder addDecimalPrecisionMode(
+ DecimalPrecisionMode decimalPrecisionMode) {
+ this.decimalPrecisionMode = decimalPrecisionMode;
+ return this;
+ }
+
+ public AsyncPostTransformFunctionBuilder addUdfFunctions(
+ List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
+ this.udfFunctions.addAll(udfFunctions);
+ return this;
+ }
+
+ public AsyncPostTransformFunctionBuilder addModelClients(Map<String, AiModelClient> clients) {
+ this.modelClients.putAll(clients);
+ return this;
+ }
+
+ public AsyncPostTransformFunctionBuilder addAsyncWorkerThreads(int asyncWorkerThreads) {
+ this.asyncWorkerThreads = asyncWorkerThreads;
+ return this;
+ }
+
+ public AsyncPostTransformFunction build() {
+ return new AsyncPostTransformFunction(
+ transformRules,
+ timezone,
+ decimalPrecisionMode,
+ udfFunctions,
+ modelClients,
+ asyncWorkerThreads);
+ }
+}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformOperatorFactory.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformOperatorFactory.java
new file mode 100644
index 000000000..cff11ed3f
--- /dev/null
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformOperatorFactory.java
@@ -0,0 +1,64 @@
+/*
+ * 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.flink.cdc.runtime.operators.transform.async;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.runtime.operators.AsyncWaitOperatorAdapter;
+import org.apache.flink.streaming.api.datastream.AsyncDataStream;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorFactory;
+
+/** Factory for an ordered async post-transform operator with state-consistent checkpoints. */
+@Internal
+public class AsyncPostTransformOperatorFactory extends AsyncWaitOperatorFactory<Event, Event> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final AsyncPostTransformFunction asyncFunction;
+ private final long timeout;
+ private final int capacity;
+
+ public AsyncPostTransformOperatorFactory(
+ AsyncPostTransformFunction asyncFunction, long timeout, int capacity) {
+ super(asyncFunction, timeout, capacity, AsyncDataStream.OutputMode.ORDERED);
+ this.asyncFunction = asyncFunction;
+ this.timeout = timeout;
+ this.capacity = capacity;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<Event>> T createStreamOperator(
+ StreamOperatorParameters<Event> parameters) {
+ return (T)
+ new AsyncWaitOperatorAdapter<>(
+ parameters,
+ asyncFunction,
+ timeout,
+ capacity,
+ processingTimeService,
+ getMailboxExecutor());
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ return AsyncWaitOperatorAdapter.class;
+ }
+}
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunctionTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunctionTest.java
new file mode 100644
index 000000000..b2b0b54a5
--- /dev/null
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunctionTest.java
@@ -0,0 +1,386 @@
+/*
+ * 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.flink.cdc.runtime.operators.transform.async;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.common.udf.UserDefinedFunction;
+import org.apache.flink.cdc.runtime.serializer.event.EventSerializer;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.mailbox.Mail;
+import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailbox;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link AsyncPostTransformFunction}. */
+class AsyncPostTransformFunctionTest {
+
+ private static final TableId TABLE_ID = TableId.tableId("ns", "schema", "customers");
+ private static final Schema SCHEMA =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+ private static final Schema SCHEMA_AFTER_ADD_COLUMN =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("region", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+
+ @AfterEach
+ void releaseBlockingInvocation() {
+ BlockingFunction.releaseFirstInvocation();
+ }
+
+ @Test
+ void testDataChangesExecuteConcurrentlyAndEmitInOrder() throws Exception {
+ BlockingFunction.reset(true);
+ try (OneInputStreamOperatorTestHarness<Event, Event> harness =
+ createHarness(TABLE_ID.identifier(), "*, block(id) AS blocked", 10_000L, 2)) {
+ harness.setup(EventSerializer.INSTANCE);
+ harness.open();
+ harness.processElement(new StreamRecord<>(new CreateTableEvent(TABLE_ID, SCHEMA)));
+ waitUntilOutputSize(harness, 1);
+
+ DataChangeEvent first = insert(SCHEMA, 1, "Alice");
+ DataChangeEvent second = insert(SCHEMA, 2, "Bob");
+ harness.processElement(new StreamRecord<>(first));
+ assertThat(BlockingFunction.awaitFirstInvocation()).isTrue();
+ harness.processElement(new StreamRecord<>(second));
+
+ assertThat(BlockingFunction.awaitSecondInvocation()).isTrue();
+ drainMailbox(harness);
+ assertThat(harness.getOutput()).hasSize(1);
+
+ BlockingFunction.releaseFirstInvocation();
+ waitUntilOutputSize(harness, 3);
+
+ Schema outputSchema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("blocked", DataTypes.INT())
+ .primaryKey("id")
+ .build();
+ assertThat(harness.extractOutputValues())
+ .containsExactly(
+ new CreateTableEvent(TABLE_ID, outputSchema),
+ insert(outputSchema, 1, "Alice", 1),
+ insert(outputSchema, 2, "Bob", 2));
+ }
+ }
+
+ @Test
+ void testSchemaChangeWaitsForPreviousDataAndBlocksFollowingData() throws Exception {
+ BlockingFunction.reset(true);
+ try (OneInputStreamOperatorTestHarness<Event, Event> harness =
+ createHarness(TABLE_ID.identifier(), "*, block(id) AS blocked", 10_000L, 2)) {
+ harness.setup(EventSerializer.INSTANCE);
+ harness.open();
+ harness.processElement(new StreamRecord<>(new CreateTableEvent(TABLE_ID, SCHEMA)));
+ waitUntilOutputSize(harness, 1);
+
+ harness.processElement(new StreamRecord<>(insert(SCHEMA, 1, "Alice")));
+ assertThat(BlockingFunction.awaitFirstInvocation()).isTrue();
+ AddColumnEvent addColumnEvent = addRegionColumnEvent();
+ harness.processElement(new StreamRecord<>(addColumnEvent));
+ harness.processElement(
+ new StreamRecord<>(insert(SCHEMA_AFTER_ADD_COLUMN, 2, "Bob", "Berlin")));
+
+ assertThat(BlockingFunction.awaitSecondInvocation(Duration.ofMillis(100))).isFalse();
+ BlockingFunction.releaseFirstInvocation();
+ assertThat(BlockingFunction.awaitSecondInvocation()).isTrue();
+ waitUntilOutputSize(harness, 4);
+
+ assertThat(harness.extractOutputValues().get(1)).isInstanceOf(DataChangeEvent.class);
+ assertThat(harness.extractOutputValues().get(2))
+ .isEqualTo(
+ new AddColumnEvent(
+ TABLE_ID,
+ Collections.singletonList(
+ AddColumnEvent.after(
+ Column.physicalColumn(
+ "region", DataTypes.STRING()),
+ "blocked"))));
+ assertThat(harness.extractOutputValues().get(3)).isInstanceOf(DataChangeEvent.class);
+ }
+ }
+
+ @Test
+ void testSchemaChangeAfterRestoreDoesNotEmitStaleCreateTableEvent() throws Exception {
+ OperatorSubtaskState snapshot;
+ try (OneInputStreamOperatorTestHarness<Event, Event> harness =
+ createHarness(TABLE_ID.identifier(), "*", 10_000L, 1)) {
+ harness.setup(EventSerializer.INSTANCE);
+ harness.open();
+ harness.processElement(new StreamRecord<>(new CreateTableEvent(TABLE_ID, SCHEMA)));
+ waitUntilOutputSize(harness, 1);
+ snapshot = harness.snapshot(1L, 1L);
+ }
+
+ try (OneInputStreamOperatorTestHarness<Event, Event> restoredHarness =
+ createHarness(TABLE_ID.identifier(), "*", 10_000L, 1)) {
+ restoredHarness.setup(EventSerializer.INSTANCE);
+ restoredHarness.initializeState(snapshot);
+ restoredHarness.open();
+ AddColumnEvent addColumnEvent = addRegionColumnEvent();
+ restoredHarness.processElement(new StreamRecord<>(addColumnEvent));
+ restoredHarness.processElement(
+ new StreamRecord<>(insert(SCHEMA_AFTER_ADD_COLUMN, 2, "Bob", "Berlin")));
+ waitUntilOutputSize(restoredHarness, 2);
+
+ assertThat(restoredHarness.extractOutputValues())
+ .containsExactly(
+ new AddColumnEvent(
+ TABLE_ID,
+ Collections.singletonList(
+ AddColumnEvent.after(
+ Column.physicalColumn(
+ "region", DataTypes.STRING()),
+ "name"))),
+ insert(SCHEMA_AFTER_ADD_COLUMN, 2, "Bob", "Berlin"));
+ }
+ }
+
+ @Test
+ void testTimeoutIsPropagated() throws Exception {
+ BlockingFunction.reset(true);
+ try (OneInputStreamOperatorTestHarness<Event, Event> harness =
+ createHarness(TABLE_ID.identifier(), "*, block(id) AS blocked", 50L, 1)) {
+ harness.setup(EventSerializer.INSTANCE);
+ harness.getEnvironment().setExpectedExternalFailureCause(Throwable.class);
+ harness.open();
+ harness.processElement(new StreamRecord<>(new CreateTableEvent(TABLE_ID, SCHEMA)));
+ waitUntilOutputSize(harness, 1);
+ DataChangeEvent event = insert(SCHEMA, 1, "Alice");
+ harness.processElement(new StreamRecord<>(event));
+ assertThat(BlockingFunction.awaitFirstInvocation()).isTrue();
+
+ harness.setProcessingTime(100L);
+ assertThat(waitUntilExternalFailure(harness))
+ .rootCause()
+ .hasMessageContaining("Async post-transform timed out for event");
+ BlockingFunction.releaseFirstInvocation();
+ }
+ }
+
+ @Test
+ void testTransformExceptionIsPropagated() throws Exception {
+ try (OneInputStreamOperatorTestHarness<Event, Event> harness =
+ createHarness(TABLE_ID.identifier(), "*, fail(id) AS failed", 10_000L, 1)) {
+ harness.setup(EventSerializer.INSTANCE);
+ harness.getEnvironment().setExpectedExternalFailureCause(Throwable.class);
+ harness.open();
+ harness.processElement(new StreamRecord<>(new CreateTableEvent(TABLE_ID, SCHEMA)));
+ waitUntilOutputSize(harness, 1);
+ harness.processElement(new StreamRecord<>(insert(SCHEMA, 1, "Alice")));
+
+ assertThat(waitUntilExternalFailure(harness))
+ .rootCause()
+ .hasMessage("expected transform failure");
+ }
+ }
+
+ private OneInputStreamOperatorTestHarness<Event, Event> createHarness(
+ String tableInclusion, String projection, long timeout, int workerThreads)
+ throws Exception {
+ return new OneInputStreamOperatorTestHarness<>(
+ new AsyncPostTransformOperatorFactory(
+ createFunction(tableInclusion, projection, workerThreads), timeout, 10),
+ EventSerializer.INSTANCE);
+ }
+
+ private AsyncPostTransformFunction createFunction(
+ String tableInclusion, String projection, int workerThreads) {
+ AsyncPostTransformFunctionBuilder builder =
+ AsyncPostTransformFunction.newBuilder()
+ .addTransform(
+ tableInclusion,
+ projection,
+ null,
+ null,
+ null,
+ null,
+ null,
+ new SupportedMetadataColumn[0])
+ .addTimezone("UTC")
+ .addAsyncWorkerThreads(workerThreads);
+ if (projection.contains("block(")) {
+ builder.addUdfFunctions(
+ Collections.singletonList(
+ Tuple3.of(
+ "block",
+ BlockingFunction.class.getName(),
+ Collections.emptyMap())));
+ } else if (projection.contains("fail(")) {
+ builder.addUdfFunctions(
+ Collections.singletonList(
+ Tuple3.of(
+ "fail",
+ FailingFunction.class.getName(),
+ Collections.emptyMap())));
+ }
+ return builder.build();
+ }
+
+ private static AddColumnEvent addRegionColumnEvent() {
+ return new AddColumnEvent(
+ TABLE_ID,
+ Collections.singletonList(
+ AddColumnEvent.last(Column.physicalColumn("region", DataTypes.STRING()))));
+ }
+
+ private static DataChangeEvent insert(Schema schema, Object... values) {
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator((RowType) schema.toRowDataType());
+ Object[] binaryValues = new Object[values.length];
+ for (int i = 0; i < values.length; i++) {
+ binaryValues[i] =
+ values[i] instanceof String
+ ? BinaryStringData.fromString((String) values[i])
+ : values[i];
+ }
+ BinaryRecordData record = generator.generate(binaryValues);
+ return DataChangeEvent.insertEvent(TABLE_ID, record);
+ }
+
+ private static void waitUntilOutputSize(
+ OneInputStreamOperatorTestHarness<Event, Event> harness, int expectedSize)
+ throws Exception {
+ long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos();
+ while (System.nanoTime() < deadline) {
+ drainMailbox(harness);
+ if (harness.getOutput().size() >= expectedSize) {
+ return;
+ }
+ Thread.sleep(10L);
+ }
+ assertThat(harness.getOutput()).hasSize(expectedSize);
+ }
+
+ private static Throwable waitUntilExternalFailure(
+ OneInputStreamOperatorTestHarness<Event, Event> harness) throws Exception {
+ long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos();
+ Optional<? extends Throwable> failure;
+ while (System.nanoTime() < deadline) {
+ drainMailbox(harness);
+ failure = harness.getEnvironment().getActualExternalFailureCause();
+ if (failure.isPresent()) {
+ return failure.get();
+ }
+ Thread.sleep(10L);
+ }
+ failure = harness.getEnvironment().getActualExternalFailureCause();
+ assertThat(failure).isPresent();
+ return failure.get();
+ }
+
+ private static void drainMailbox(OneInputStreamOperatorTestHarness<Event, Event> harness)
+ throws Exception {
+ while (true) {
+ Mail mail = harness.getTaskMailbox().tryTake(TaskMailbox.MIN_PRIORITY).orElse(null);
+ if (mail == null) {
+ return;
+ }
+ mail.run();
+ }
+ }
+
+ /** Test UDF that allows assertions about worker execution order. */
+ public static class BlockingFunction implements UserDefinedFunction {
+
+ private static volatile boolean blockFirst;
+ private static volatile CountDownLatch firstInvocationStarted = new CountDownLatch(1);
+ private static volatile CountDownLatch secondInvocationStarted = new CountDownLatch(1);
+ private static volatile CountDownLatch firstInvocationRelease = new CountDownLatch(1);
+
+ public Integer eval(Integer value) {
+ if (value == 1) {
+ firstInvocationStarted.countDown();
+ if (blockFirst) {
+ try {
+ firstInvocationRelease.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ } else if (value == 2) {
+ secondInvocationStarted.countDown();
+ }
+ return value;
+ }
+
+ private static void reset(boolean shouldBlockFirst) {
+ blockFirst = shouldBlockFirst;
+ firstInvocationStarted = new CountDownLatch(1);
+ secondInvocationStarted = new CountDownLatch(1);
+ firstInvocationRelease = new CountDownLatch(1);
+ }
+
+ private static boolean awaitFirstInvocation() throws InterruptedException {
+ return firstInvocationStarted.await(10, TimeUnit.SECONDS);
+ }
+
+ private static boolean awaitSecondInvocation() throws InterruptedException {
+ return awaitSecondInvocation(Duration.ofSeconds(10));
+ }
+
+ private static boolean awaitSecondInvocation(Duration timeout) throws InterruptedException {
+ return secondInvocationStarted.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ private static void releaseFirstInvocation() {
+ firstInvocationRelease.countDown();
+ }
+ }
+
+ /** Test UDF that always fails. */
+ public static class FailingFunction implements UserDefinedFunction {
+
+ public Integer eval(Integer value) {
+ throw new IllegalStateException("expected transform failure");
+ }
+ }
+}