diff --git a/docs/content.zh/docs/core-concept/data-pipeline.md b/docs/content.zh/docs/core-concept/data-pipeline.md index bf4f3212e12..62d9c023434 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 上的诊断。 | 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 恢复时开启或关闭此选项。 注意：虽然上述参数都是可选的，但至少需要指定其中一个。`pipeline` 部分是必需的，不能为空。 diff --git a/docs/content/docs/core-concept/data-pipeline.md b/docs/content/docs/core-concept/data-pipeline.md index 4b35ad6ca3c..664c5c5db4e 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. | 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. 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 0e46508bb2e..512c684ac4e 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 @@ * *

Implementations must be {@link Serializable} so that they can be distributed across Flink task * managers together with the operator that holds them. + * + *

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 d46c94c6530..3b39645d096 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 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 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 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 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 0fa5d78c8c9..f20343b138a 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 @@ /** * Base interface for creating a UDF in transform projection and filtering expressions. You should * define at least one {@code eval} method. + * + *

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 c0d693ed201..3b73a283e50 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 @@ 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 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 99be446f200..a8c650c07b2 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.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 DataStream translatePostTransform( .uid(operatorUidGenerator.generateUid("post-transform")); } + public DataStream translateAsyncPostTransform( + DataStream input, + List transforms, + String timezone, + DecimalPrecisionMode decimalPrecisionMode, + List udfFunctions, + List 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 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> 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 9d5a8b22e23..2bb4b14f38e 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.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.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 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 @@ void cleanup() { // ---------------------- // 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 @@ private static Stream testParams() { arguments(ValuesDataSink.SinkApi.SINK_FUNCTION, "scala"), arguments(ValuesDataSink.SinkApi.SINK_V2, "scala")); } + + private static List 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 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 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 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 00000000000..12aa4c3ca81 --- /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 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 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-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java index 4f9c7bd19af..3fe6e85c280 100644 --- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java +++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java @@ -20,19 +20,23 @@ 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.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; @@ -52,8 +56,11 @@ import static org.assertj.core.api.Assertions.assertThat; /** E2e tests for the {@link PreTransformOperator} and {@link PostTransformOperator}. */ +@ParameterizedClass(name = "asyncTransform: {0}") +@ValueSource(booleans = {false, true}) 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); @@ -64,6 +71,17 @@ class TransformE2eITCase extends PipelineTestEnvironment { return String.format(s, databaseName, databaseName, databaseName); }; + private final boolean asyncTransform; + + TransformE2eITCase(boolean asyncTransform) { + 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(); @@ -76,6 +94,107 @@ public void after() { 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: 1\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); + Path udfJar = TestUtils.getResource("udf-examples.jar"); + + JobID jobId = submitPipelineJob(pipelineJob, udfJar); + waitUntilJobRunning(Duration.ofSeconds(30)); + waitUntilStreamSplitReady(jobId, 1); + + 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 " + lastRecordId, + "SkewedThrottlerFunctionClass finished " + slowRecordId); + + String savepointPath = stopJobWithSavepoint(jobId); + waitUntilSpecificEventsAfter( + incrementalOutputOffset, + ASYNC_RESTORE_TIMEOUT, + tableAlphaInsertEvent(databaseName, slowRecordId), + tableAlphaInsertEvent(databaseName, lastRecordId)); + + int restoredOutputOffset = taskManagerConsumer.toUtf8String().length(); + JobID restoredJobId = submitPipelineJob(pipelineJob, savepointPath, false, udfJar); + waitUntilJobRunning(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); + waitUntilSpecificEventsAfter( + restoredOutputOffset, + ASYNC_RESTORE_TIMEOUT, + restoredCreateTableEvent, + tableAlphaInsertEvent(databaseName, 5000), + tableAlphaInsertEvent(databaseName, 5001)); + + String restoredOutput = + taskManagerConsumer + .toUtf8String() + .substring( + Math.min( + restoredOutputOffset, + taskManagerConsumer.toUtf8String().length())); + assertThat(restoredOutput).containsOnlyOnce(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 { @@ -118,7 +237,7 @@ void testHeteroSchemaTransform(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -202,7 +321,7 @@ void testMultipleTransformRule(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, testParallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -293,7 +412,7 @@ void testAssortedSchemaTransform(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -375,7 +494,7 @@ void testWildcardSchemaTransform(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -465,7 +584,7 @@ void testWildcardSchemaTransformWithFilter(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, testParallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -548,7 +667,7 @@ void testWildcardWithMetadataColumnTransform(boolean batchMode) throws Exception transformTestDatabase.getDatabaseName(), runtimeMode, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -634,7 +753,7 @@ void testMultipleHittingTable(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, testParallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -715,7 +834,7 @@ void testMultipleTransformWithDiffRefColumn(boolean batchMode) throws Exception transformTestDatabase.getDatabaseName(), runtimeMode, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -795,7 +914,7 @@ void testTransformWithCast(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); if (batchMode) { @@ -877,7 +996,7 @@ void testTemporalFunctions(boolean batchMode) throws Exception { transformTestDatabase.getDatabaseName(), runtimeMode, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -912,7 +1031,7 @@ void testTransformWithSchemaEvolution() throws Exception { transformTestDatabase.getDatabaseName(), transformTestDatabase.getDatabaseName(), parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -1007,7 +1126,7 @@ void testTransformWildcardPrefixWithSchemaEvolution() throws Exception { transformTestDatabase.getDatabaseName(), transformTestDatabase.getDatabaseName(), parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -1110,7 +1229,7 @@ void testTransformWildcardSuffixWithSchemaEvolution() throws Exception { transformTestDatabase.getDatabaseName(), transformTestDatabase.getDatabaseName(), parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -1243,7 +1362,7 @@ void testTransformWithUnicodeLiterals() throws Exception { transformTestDatabase.getDatabaseName(), projectionExpression, parallelism); - JobID jobId = submitPipelineJob(pipelineJob); + JobID jobId = submitPipelineJob(withAsyncTransform(pipelineJob)); waitUntilJobRunning(Duration.ofSeconds(30)); LOG.info("Pipeline job is running"); @@ -1320,6 +1439,97 @@ void testTransformWithUnicodeLiterals() throws Exception { "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 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-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 00000000000..0b8dc06f6a1 --- /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 extends AsyncWaitOperator { + + private static final long serialVersionUID = 1L; + + @SuppressWarnings("unchecked") + public AsyncWaitOperatorAdapter( + StreamOperatorParameters parameters, + AsyncFunction 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 00000000000..0e2a54e1f4f --- /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 extends AsyncWaitOperator { + + private static final long serialVersionUID = 1L; + + @SuppressWarnings("unchecked") + public AsyncWaitOperatorAdapter( + StreamOperatorParameters parameters, + AsyncFunction 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 00000000000..034db744464 --- /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 00000000000..524ca68ea53 --- /dev/null +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunction.java @@ -0,0 +1,958 @@ +/* + * 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. + * + *

{@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 + 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 = 2; + + private final String timezone; + private final DecimalPrecisionMode decimalPrecisionMode; + private final List transformRules; + private final Map tableInfoMap; + + // Tuple3 items are: function name, class path, and extra options. + private final List>> udfFunctions; + + // Serializable AI model clients keyed by model name, e.g. myModel. + private final Map modelClients; + + private transient List transformers; + private transient List udfDescriptors; + private transient List udfFunctionInstances; + private transient ThreadLocal> + projectionProcessors; + private transient ThreadLocal> + filterProcessors; + private transient Queue> + projectionProcessorCaches; + private transient Queue> + filterProcessorCaches; + private transient LoadingCache> transformersCache; + + private final int asyncWorkerThreads; + + private transient ExecutorService executorService; + private transient Set>> pendingDataFutures; + private transient CompletableFuture schemaBarrierFuture; + private transient ListState tableState; + private transient Set emittedCreateTableEventTables; + + public static AsyncPostTransformFunctionBuilder newBuilder() { + return new AsyncPostTransformFunctionBuilder(); + } + + AsyncPostTransformFunction( + List transformRules, + String timezone, + DecimalPrecisionMode decimalPrecisionMode, + List>> udfFunctions, + Map 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 + processors = HashBasedTable.create(); + projectionProcessorCaches.add(processors); + return processors; + }); + this.filterProcessors = + ThreadLocal.withInitial( + () -> { + Table 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 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 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 prependedEvents = prependCreateTableEventIfNeeded(tableId); + asyncInvokeDataChangeEvent(event, resultFuture, prependedEvents); + } else { + asyncInvokeBarrierEvent(event, resultFuture); + } + } + + @Override + public void timeout(Event event, ResultFuture resultFuture) { + resultFuture.completeExceptionally( + new FlinkRuntimeException("Async post-transform timed out for event: " + event)); + } + + private void asyncInvokeDataChangeEvent( + Event event, ResultFuture resultFuture, List prependedEvents) { + CompletableFuture> 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 resultFuture) { + CompletableFuture previousSchemaBarrierFuture = schemaBarrierFuture; + CompletableFuture previousDataFutures = waitForPendingDataFutures(); + CompletableFuture> 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 waitForPendingDataFutures() { + CompletableFuture[] futures = pendingDataFutures.toArray(new CompletableFuture[0]); + return CompletableFuture.allOf(futures); + } + + private List processSafely(Event event) { + try { + Optional 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 resultFuture, + @Nullable List result, + @Nullable Throwable error) { + if (error != null) { + resultFuture.completeExceptionally( + wrapTransformException("async post-transform", event, error)); + } else { + resultFuture.complete(result); + } + } + + private List 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 prependEvents( + List prependedEvents, @Nullable List result) { + if (prependedEvents.isEmpty()) { + return result; + } + List output = new ArrayList<>(prependedEvents); + if (result != null) { + output.addAll(result); + } + return output; + } + + Optional 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 transformer = transformersCache.getUnchecked(tableId); + + if (transformer.isEmpty()) { + cachePassthroughSchemaEvent(event); + if (event instanceof CreateTableEvent) { + emittedCreateTableEventTables.add(tableId); + } + return Optional.of(event); + } + + if (event instanceof CreateTableEvent) { + Optional result = + processCreateTableEvent((CreateTableEvent) event, transformer.get()); + emittedCreateTableEventTables.add(tableId); + invalidateCache(tableId); + return result; + } else if (event instanceof SchemaChangeEvent) { + Optional result = + processSchemaChangeEvent((SchemaChangeEvent) event, transformer.get()); + 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 serializeTableStates() throws IOException { + List 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 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 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 columnNamesBeforeChange = prevPostSchema.getColumnNames(); + Optional 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 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 result = + transformRecord( + event.before(), info, projectionProcessor, filterProcessor, context); + beforeRow = result.f0; + beforeFilterPassed = result.f1; + } + if (event.after() != null) { + context.opType = afterOp; + Tuple2 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 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 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 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 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 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 createTransformers() { + List 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 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 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 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 projectedColumns; + + private PostTransformTableInfo( + PostTransformChangeInfo changeInfo, + @Nullable CreateTableEvent outputCreateTableEvent, + boolean hasAsterisk, + List 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 00000000000..affbe9cd76c --- /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 transformRules = new ArrayList<>(); + private String timezone; + private DecimalPrecisionMode decimalPrecisionMode = DecimalPrecisionMode.UP_TO_19; + private final List>> udfFunctions = + new ArrayList<>(); + private final Map 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>> udfFunctions) { + this.udfFunctions.addAll(udfFunctions); + return this; + } + + public AsyncPostTransformFunctionBuilder addModelClients(Map 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 00000000000..cff11ed3ff4 --- /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 { + + 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 createStreamOperator( + StreamOperatorParameters parameters) { + return (T) + new AsyncWaitOperatorAdapter<>( + parameters, + asyncFunction, + timeout, + capacity, + processingTimeService, + getMailboxExecutor()); + } + + @Override + public Class 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 00000000000..630c0a9e04b --- /dev/null +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/async/AsyncPostTransformFunctionTest.java @@ -0,0 +1,349 @@ +/* + * 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.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 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 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 testTimeoutIsPropagated() throws Exception { + BlockingFunction.reset(true); + try (OneInputStreamOperatorTestHarness 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 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 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 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 harness) throws Exception { + long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos(); + Optional 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 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"); + } + } +}