lvyanquan commented on code in PR #4526: URL: https://github.com/apache/flink-cdc/pull/4526#discussion_r4069878068
########## 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. + * + * <p>{@link SchemaChangeEvent}s are handled as barriers. The function waits for all pending {@link + * DataChangeEvent} futures submitted before the schema change, applies the schema change, and only + * then allows following data changes to run against the updated schema. + */ +public class AsyncPostTransformFunction extends RichAsyncFunction<Event, Event> + implements CheckpointedFunction, Serializable { + + private static final long serialVersionUID = 1L; + private static final String TABLE_STATE_NAME = "async-post-transform-table-state"; + private static final long EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS = 30L; + private static final Logger LOG = LoggerFactory.getLogger(AsyncPostTransformFunction.class); + + private static final int TABLE_STATE_VERSION = 1; + + private final String timezone; + private final DecimalPrecisionMode decimalPrecisionMode; + private final List<TransformRule> transformRules; + private final Map<TableId, PostTransformTableInfo> tableInfoMap; + + // Tuple3 items are: function name, class path, and extra options. + private final List<Tuple3<String, String, Map<String, String>>> udfFunctions; + + // Serializable AI model clients keyed by model name, e.g. myModel. + private final Map<String, AiModelClient> modelClients; + + private transient List<PostTransformer> transformers; + private transient List<UserDefinedFunctionDescriptor> udfDescriptors; + private transient List<Object> udfFunctionInstances; + private transient ThreadLocal<Table<TableId, PostTransformer, TransformProjectionProcessor>> + projectionProcessors; + private transient ThreadLocal<Table<TableId, PostTransformer, TransformFilterProcessor>> + filterProcessors; + private transient Queue<Table<TableId, PostTransformer, TransformProjectionProcessor>> + projectionProcessorCaches; + private transient Queue<Table<TableId, PostTransformer, TransformFilterProcessor>> + filterProcessorCaches; + private transient LoadingCache<TableId, Optional<PostTransformer>> transformersCache; + + private final int asyncWorkerThreads; + + private transient ExecutorService executorService; + private transient Set<CompletableFuture<List<Event>>> pendingDataFutures; + private transient CompletableFuture<Void> schemaBarrierFuture; + private transient ListState<byte[]> tableState; + private transient Set<TableId> emittedCreateTableEventTables; + + public static AsyncPostTransformFunctionBuilder newBuilder() { + return new AsyncPostTransformFunctionBuilder(); + } + + AsyncPostTransformFunction( + List<TransformRule> transformRules, + String timezone, + DecimalPrecisionMode decimalPrecisionMode, + List<Tuple3<String, String, Map<String, String>>> udfFunctions, + Map<String, AiModelClient> modelClients, + int asyncWorkerThreads) { + Preconditions.checkArgument( + asyncWorkerThreads > 0, "Async worker threads must be greater than 0."); + this.timezone = timezone; + this.decimalPrecisionMode = decimalPrecisionMode; + this.transformRules = transformRules; + this.tableInfoMap = new ConcurrentHashMap<>(); + this.udfFunctions = udfFunctions; + this.modelClients = modelClients; + this.asyncWorkerThreads = asyncWorkerThreads; + } + + @Override + public void open(OpenContext openContext) throws Exception { + super.open(openContext); + this.pendingDataFutures = ConcurrentHashMap.newKeySet(); + this.schemaBarrierFuture = CompletableFuture.completedFuture(null); + this.emittedCreateTableEventTables = ConcurrentHashMap.newKeySet(); + + this.projectionProcessorCaches = new ConcurrentLinkedQueue<>(); + this.filterProcessorCaches = new ConcurrentLinkedQueue<>(); + this.projectionProcessors = + ThreadLocal.withInitial( + () -> { + Table<TableId, PostTransformer, TransformProjectionProcessor> + processors = HashBasedTable.create(); + projectionProcessorCaches.add(processors); + return processors; + }); + this.filterProcessors = + ThreadLocal.withInitial( + () -> { + Table<TableId, PostTransformer, TransformFilterProcessor> processors = + HashBasedTable.create(); + filterProcessorCaches.add(processors); + return processors; + }); + + initializeAiModelClients(); + initializeUdf(); + + this.transformers = createTransformers(); + this.transformersCache = + CacheBuilder.newBuilder() + .maximumSize(1024) + .build( + new CacheLoader<>() { + @Override + public Optional<PostTransformer> load(TableId tableId) { + return getEffectiveTransformer(tableId); + } + }); + this.executorService = + Executors.newFixedThreadPool( + asyncWorkerThreads, + new ThreadFactoryBuilder() + .setNameFormat( + "post-transform-async-" + + getRuntimeContext() + .getTaskInfo() + .getIndexOfThisSubtask() + + "-%d") + .build()); + } + + @Override + public void close() throws Exception { + try { + boolean executorTerminated = true; + if (executorService != null) { + executorService.shutdownNow(); + executorTerminated = + executorService.awaitTermination( + EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + if (executorTerminated) { + TransformExpressionCompiler.cleanUp(); + destroyUdf(); + destroyAiModelClients(); + if (transformersCache != null) { + transformersCache.invalidateAll(); + } + } else { + LOG.warn( + "Async post-transform workers did not terminate within {} seconds; " + + "processor resources will remain open to avoid concurrent close.", + EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS); + } + } finally { + super.close(); + } + } + + @Override + public void snapshotState(FunctionSnapshotContext context) throws Exception { + tableState.clear(); + for (byte[] serializedTableState : serializeTableStates()) { + tableState.add(serializedTableState); + } + } + + @Override + public void initializeState(FunctionInitializationContext context) throws Exception { + tableState = + context.getOperatorStateStore() + .getListState(new ListStateDescriptor<>(TABLE_STATE_NAME, byte[].class)); + if (context.isRestored()) { + for (byte[] serializedTableState : tableState.get()) { + restoreTableState(serializedTableState); + } + } + } + + @Override + public void asyncInvoke(Event event, ResultFuture<Event> resultFuture) { + if (event instanceof DataChangeEvent) { + // Resolve a restored CreateTableEvent on the mailbox thread so ORDERED output + // deterministically prepends it to the first data event for the table. + TableId tableId = ((DataChangeEvent) event).tableId(); + List<Event> prependedEvents = prependCreateTableEventIfNeeded(tableId); + asyncInvokeDataChangeEvent(event, resultFuture, prependedEvents); + } else { + asyncInvokeBarrierEvent(event, resultFuture); Review Comment: **`CreateTableEvent` can be emitted after a schema change (with a stale schema) on restore** The restored `CreateTableEvent` is only prepended when the first post-restore event is a `DataChangeEvent` (`asyncInvoke`, line 279). The `SchemaChangeEvent` branch of `process()` (line 407) never marks the table as emitted. Reproduced locally with `restore → AddColumnEvent → DataChangeEvent`: ```text AddColumnEvent{A, addedColumns=[region AFTER blk]} CreateTableEvent{A, schema=(id, name, blk)} // stale — missing `region` DataChangeEvent{A, ...} // transformed against the new schema ``` So the `CreateTableEvent` arrives after the schema change, and carries the pre-change schema while the following record uses the post-change one. Downstream `registerNewSchema` bumps the version for any incoming `CreateTableEvent`, so the latest evolved schema rolls back to a version without `region` — while `MetadataApplier` has already applied the `ALTER TABLE` and records are laid out for the new schema. The sink then fails on a column mismatch or silently drops the new column. `ORDERED` mode can't help here, since the `CreateTableEvent` is packed inside the data event's own result. Could you add a test for `restore → SchemaChangeEvent → DataChangeEvent`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
