Skip to content

[FLINK-40552][runtime] Support asynchronous transform execution #4526

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
haruki-830 wants to merge 5 commits into
base: master
Choose a base branch
from
Open

[FLINK-40552][runtime] Support asynchronous transform execution #4526

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
29ff5de
[FLINK-40552][runtime] Support asynchronous transform execution
Sep 4, 2026
f42742a
[FLINK-40552][runtime] Improve asynchronous transform test coverage
Sep 10, 2026
c4cfde7
[FLINK-40552][runtime] Fix async transform compilation after rebase
Sep 11, 2026
60d9239
[FLINK-40552][runtime] Isolate asynchronous transform implementation
Sep 11, 2026
18c532d
[FLINK-40552][runtime] Make asynchronous transform function self-cont…
Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/content.zh/docs/core-concept/data-pipeline.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,5 +125,13 @@ under the License.
| `operator.uid.prefix` | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的 UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。 | optional |
| `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`。 | optional |
| `transform.async-execution.timeout` | 有序异步执行中每个 PostTransform 事件的超时时间，默认值为 5 分钟。 | optional |
| `transform.async-execution.capacity` | 异步执行中最多允许同时处理的 PostTransform 事件数，默认值为 100。 | optional |
| `transform.async-execution.worker-threads` | 每个异步 PostTransform 任务使用的工作线程数，默认值为 16。 | optional |

> **实验性且不稳定：** 异步 PostTransform 执行目前是实验性功能，其配置和行为可能会在未来版本中发生变化。

异步 PostTransform 适用于 AI 模型调用等 I/O 密集型表达式。DataChangeEvent 可能并发调用 UDF 和 AI 模型客户端，但输出和所有 SchemaChangeEvent 仍保持有序，因此 UDF 和 AI 模型客户端实现必须是线程安全的。为保证 schema 状态一致，checkpoint 或 savepoint 前会等待尚未完成的异步请求，长时间运行的请求可能会延长 checkpoint 时间。savepoint 只支持在并行度不变时恢复，并且不能在从已有 savepoint 恢复时开启或关闭此选项。
Comment thread
yuxiqian marked this conversation as resolved.

注意：虽然上述参数都是可选的，但至少需要指定其中一个。`pipeline` 部分是必需的，不能为空。
8 changes: 8 additions & 0 deletions docs/content/docs/core-concept/data-pipeline.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. | optional |
| `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. | optional |
| `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). | optional |
| `transform.async-execution.enabled` | Whether to enable ordered asynchronous execution for post-transform. Defaults to `false`. | optional |
| `transform.async-execution.timeout` | The timeout for each post-transform event in ordered asynchronous execution. Defaults to 5 minutes. | optional |
| `transform.async-execution.capacity` | The maximum number of in-flight post-transform events. Defaults to 100. | optional |
| `transform.async-execution.worker-threads` | The number of worker threads used by each asynchronous post-transform task. Defaults to 16. | optional |

> **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 supported only with unchanged parallelism, and this option must not be enabled or disabled when restoring an existing savepoint.

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.
4 changes: 4 additions & 0 deletions flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,10 @@
*
* <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 {
Expand Down
28 changes: 28 additions & 0 deletions flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {}
}
4 changes: 4 additions & 0 deletions flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,10 @@
/**
* 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 {
Expand Down
47 changes: 34 additions & 13 deletions ...cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,19 +192,40 @@ private void translate(StreamExecutionEnvironment env, PipelineDef pipelineDef)
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
Expand Down
70 changes: 70 additions & 0 deletions ...ser/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,16 +35,22 @@
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.
Expand Down Expand Up @@ -148,6 +154,70 @@ public DataStream<Event> translatePostTransform(
.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(),
Expand Down
Loading
Loading