Carl-Zhou-CN commented on code in PR #10107: URL: https://github.com/apache/seatunnel/pull/10107#discussion_r2618359827
########## seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java: ########## @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.translation.flink.schema; + +import org.apache.seatunnel.shade.com.typesafe.config.Config; + +import org.apache.seatunnel.api.source.SupportSchemaEvolution; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.schema.SchemaChangeType; +import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent; +import org.apache.seatunnel.api.table.schema.event.TableEvent; +import org.apache.seatunnel.api.table.schema.exception.SchemaValidationException; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.translation.flink.schema.coordinator.LocalSchemaCoordinator; + +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; + +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; + +/** operators added to the source and transformer pipelines to handle schema evolution */ +@Slf4j +public class SchemaOperator extends AbstractStreamOperator<SeaTunnelRow> + implements OneInputStreamOperator<SeaTunnelRow, SeaTunnelRow> { + private final Map<TableIdentifier, CatalogTable> localSchemaState; + private String jobId; + private final SupportSchemaEvolution source; + private final Config pluginConfig; + private volatile Long lastProcessedEventTime; + private transient LocalSchemaCoordinator coordinator; + private transient Map<String, List<BufferedDataRow>> bufferedDataRows; + private volatile boolean schemaChangePending = false; + private volatile CompletableFuture<Boolean> pendingSchemaFuture = null; + + public SchemaOperator(String jobId, SupportSchemaEvolution source, Config pluginConfig) { + this.jobId = jobId; + this.source = source; + this.pluginConfig = pluginConfig; + this.localSchemaState = new ConcurrentHashMap<>(); + } + + @Override + public void open() throws Exception { + super.open(); + String flinkJobId = getRuntimeContext().getJobId().toString(); + if (!flinkJobId.equals(this.jobId)) { + this.jobId = flinkJobId; + } + this.bufferedDataRows = new ConcurrentHashMap<>(); + this.coordinator = LocalSchemaCoordinator.getInstance(this.jobId); + log.info("SchemaOperator opened for job: {}", this.jobId); + } + + @Override + public void processElement(StreamRecord<SeaTunnelRow> streamRecord) { + SeaTunnelRow element = streamRecord.getValue(); + + if (pluginConfig.hasPath("schema-changes.enabled")) { + output.collect(new StreamRecord<>(element, streamRecord.getTimestamp())); + return; + } + + if ("__SCHEMA_CHANGE_EVENT__".equals(element.getTableId()) + && element.getOptions() != null) { + Object object = element.getOptions().get("schema_change_event"); + if (object instanceof SchemaChangeEvent) { + handleSchemaChangeEvent((SchemaChangeEvent) object); + return; + } + } + + if (schemaChangePending && pendingSchemaFuture != null) { + String tableId = element.getTableId(); + if (tableId != null) { + String key = tableId + "#" + System.currentTimeMillis(); + bufferedDataRows + .computeIfAbsent(key, k -> new ArrayList<>()) + .add(new BufferedDataRow(element, streamRecord.getTimestamp())); + return; + } + } + + output.collect(new StreamRecord<>(element, streamRecord.getTimestamp())); + } + + private void handleSchemaChangeEvent(SchemaChangeEvent schemaChangeEvent) { + List<SchemaChangeType> supportedTypes = source.supports(); + if (supportedTypes == null || supportedTypes.isEmpty()) { + log.info( + "Source: {} does not support any schema change types, skipping schema change event", + source); + return; + } + + if (!isSchemaChangeSupported(schemaChangeEvent, supportedTypes)) { + log.warn( + "Schema change type {} not supported by source {}, skipping", + schemaChangeEvent.getEventType(), + source); + return; + } + + processSchemaChangeEvent(schemaChangeEvent); + } + + private boolean isSchemaChangeSupported( + SchemaChangeEvent event, List<SchemaChangeType> supportedTypes) { + switch (event.getEventType()) { + case SCHEMA_CHANGE_ADD_COLUMN: + return supportedTypes.contains(SchemaChangeType.ADD_COLUMN); + case SCHEMA_CHANGE_DROP_COLUMN: + return supportedTypes.contains(SchemaChangeType.DROP_COLUMN); + case SCHEMA_CHANGE_MODIFY_COLUMN: + return supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN); + case SCHEMA_CHANGE_CHANGE_COLUMN: + return supportedTypes.contains(SchemaChangeType.RENAME_COLUMN); + case SCHEMA_CHANGE_UPDATE_COLUMNS: + return supportedTypes.contains(SchemaChangeType.ADD_COLUMN) + || supportedTypes.contains(SchemaChangeType.DROP_COLUMN) + || supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN) + || supportedTypes.contains(SchemaChangeType.RENAME_COLUMN); + default: + log.error("Unknown schema change event type: {}", event.getEventType()); + throw SchemaValidationException.unsupportedChangeType( + event.tableIdentifier(), jobId); + } + } + + private void processSchemaChangeEvent(SchemaChangeEvent schemaChangeEvent) { + TableIdentifier tableId = schemaChangeEvent.tableIdentifier(); + long eventTime = schemaChangeEvent.getCreatedTime(); + + try { + if (lastProcessedEventTime != null && eventTime <= lastProcessedEventTime) { Review Comment: Could you give an example of the scene that occurred? ########## seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java: ########## @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.translation.flink.schema; + +import org.apache.seatunnel.shade.com.typesafe.config.Config; + +import org.apache.seatunnel.api.source.SupportSchemaEvolution; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.schema.SchemaChangeType; +import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent; +import org.apache.seatunnel.api.table.schema.event.TableEvent; +import org.apache.seatunnel.api.table.schema.exception.SchemaValidationException; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.translation.flink.schema.coordinator.LocalSchemaCoordinator; + +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; + +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; + +/** operators added to the source and transformer pipelines to handle schema evolution */ +@Slf4j +public class SchemaOperator extends AbstractStreamOperator<SeaTunnelRow> + implements OneInputStreamOperator<SeaTunnelRow, SeaTunnelRow> { + private final Map<TableIdentifier, CatalogTable> localSchemaState; + private String jobId; + private final SupportSchemaEvolution source; + private final Config pluginConfig; + private volatile Long lastProcessedEventTime; + private transient LocalSchemaCoordinator coordinator; + private transient Map<String, List<BufferedDataRow>> bufferedDataRows; + private volatile boolean schemaChangePending = false; + private volatile CompletableFuture<Boolean> pendingSchemaFuture = null; + + public SchemaOperator(String jobId, SupportSchemaEvolution source, Config pluginConfig) { + this.jobId = jobId; + this.source = source; + this.pluginConfig = pluginConfig; + this.localSchemaState = new ConcurrentHashMap<>(); + } + + @Override + public void open() throws Exception { + super.open(); + String flinkJobId = getRuntimeContext().getJobId().toString(); + if (!flinkJobId.equals(this.jobId)) { + this.jobId = flinkJobId; + } + this.bufferedDataRows = new ConcurrentHashMap<>(); + this.coordinator = LocalSchemaCoordinator.getInstance(this.jobId); + log.info("SchemaOperator opened for job: {}", this.jobId); + } + + @Override + public void processElement(StreamRecord<SeaTunnelRow> streamRecord) { + SeaTunnelRow element = streamRecord.getValue(); + + if (pluginConfig.hasPath("schema-changes.enabled")) { + output.collect(new StreamRecord<>(element, streamRecord.getTimestamp())); + return; + } + + if ("__SCHEMA_CHANGE_EVENT__".equals(element.getTableId()) + && element.getOptions() != null) { + Object object = element.getOptions().get("schema_change_event"); + if (object instanceof SchemaChangeEvent) { + handleSchemaChangeEvent((SchemaChangeEvent) object); + return; + } + } + + if (schemaChangePending && pendingSchemaFuture != null) { + String tableId = element.getTableId(); + if (tableId != null) { + String key = tableId + "#" + System.currentTimeMillis(); + bufferedDataRows + .computeIfAbsent(key, k -> new ArrayList<>()) + .add(new BufferedDataRow(element, streamRecord.getTimestamp())); + return; + } + } + + output.collect(new StreamRecord<>(element, streamRecord.getTimestamp())); + } + + private void handleSchemaChangeEvent(SchemaChangeEvent schemaChangeEvent) { + List<SchemaChangeType> supportedTypes = source.supports(); + if (supportedTypes == null || supportedTypes.isEmpty()) { + log.info( + "Source: {} does not support any schema change types, skipping schema change event", + source); + return; + } + + if (!isSchemaChangeSupported(schemaChangeEvent, supportedTypes)) { + log.warn( + "Schema change type {} not supported by source {}, skipping", + schemaChangeEvent.getEventType(), + source); + return; + } + + processSchemaChangeEvent(schemaChangeEvent); + } + + private boolean isSchemaChangeSupported( + SchemaChangeEvent event, List<SchemaChangeType> supportedTypes) { + switch (event.getEventType()) { + case SCHEMA_CHANGE_ADD_COLUMN: + return supportedTypes.contains(SchemaChangeType.ADD_COLUMN); + case SCHEMA_CHANGE_DROP_COLUMN: + return supportedTypes.contains(SchemaChangeType.DROP_COLUMN); + case SCHEMA_CHANGE_MODIFY_COLUMN: + return supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN); + case SCHEMA_CHANGE_CHANGE_COLUMN: + return supportedTypes.contains(SchemaChangeType.RENAME_COLUMN); + case SCHEMA_CHANGE_UPDATE_COLUMNS: + return supportedTypes.contains(SchemaChangeType.ADD_COLUMN) + || supportedTypes.contains(SchemaChangeType.DROP_COLUMN) + || supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN) + || supportedTypes.contains(SchemaChangeType.RENAME_COLUMN); + default: + log.error("Unknown schema change event type: {}", event.getEventType()); + throw SchemaValidationException.unsupportedChangeType( + event.tableIdentifier(), jobId); + } + } + + private void processSchemaChangeEvent(SchemaChangeEvent schemaChangeEvent) { + TableIdentifier tableId = schemaChangeEvent.tableIdentifier(); + long eventTime = schemaChangeEvent.getCreatedTime(); + + try { + if (lastProcessedEventTime != null && eventTime <= lastProcessedEventTime) { + throw SchemaValidationException.outdatedEvent( + tableId, jobId, eventTime, lastProcessedEventTime); + } + + if (schemaChangeEvent instanceof TableEvent) { + schemaChangeEvent.setJobId(jobId); + } + + log.info( + "Starting async schema change processing for table: {}, job: {}, event time: {}", + tableId, + jobId, + eventTime); + + String key = tableId.toString() + "#" + eventTime; Review Comment: One method can be used. I see that it has been applied in several places ########## seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java: ########## @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.translation.flink.schema; + +import org.apache.seatunnel.shade.com.typesafe.config.Config; + +import org.apache.seatunnel.api.source.SupportSchemaEvolution; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.schema.SchemaChangeType; +import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent; +import org.apache.seatunnel.api.table.schema.event.TableEvent; +import org.apache.seatunnel.api.table.schema.exception.SchemaValidationException; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.translation.flink.schema.coordinator.LocalSchemaCoordinator; + +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; + +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; + +/** operators added to the source and transformer pipelines to handle schema evolution */ +@Slf4j +public class SchemaOperator extends AbstractStreamOperator<SeaTunnelRow> + implements OneInputStreamOperator<SeaTunnelRow, SeaTunnelRow> { + private final Map<TableIdentifier, CatalogTable> localSchemaState; + private String jobId; + private final SupportSchemaEvolution source; + private final Config pluginConfig; + private volatile Long lastProcessedEventTime; + private transient LocalSchemaCoordinator coordinator; + private transient Map<String, List<BufferedDataRow>> bufferedDataRows; Review Comment: Doesn't it need to be stored in the checkpoint? ########## seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/coordinator/LocalSchemaCoordinator.java: ########## @@ -0,0 +1,336 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.translation.flink.schema.coordinator; + +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.schema.exception.SchemaCoordinationException; +import org.apache.seatunnel.api.table.schema.exception.SchemaEvolutionErrorCode; +import org.apache.seatunnel.api.table.schema.exception.SchemaEvolutionException; + +import lombok.extern.slf4j.Slf4j; + +import java.lang.ref.WeakReference; +import java.util.Iterator; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Local coordinator for schema change synchronization. This coordinator only manages temporary + * communication between SchemaOperator and sink subtasks. All persistent state is managed by + * BroadcastSchemaSinkOperator in Flink State. + */ +@Slf4j +public class LocalSchemaCoordinator { + + private static final Map<String, WeakReference<LocalSchemaCoordinator>> instances = + new ConcurrentHashMap<>(); + private static final ScheduledExecutorService cleanupExecutor = + new ScheduledThreadPoolExecutor( + 1, + r -> { + Thread t = new Thread(r, "LocalSchemaCoordinator-Cleanup"); + t.setDaemon(true); + return t; + }); + private static final long DEFAULT_REQUEST_TTL_MS = 300_000L; + private static final long CLEANUP_INTERVAL_MS = 60_000L; + private final String jobId; + private final long requestTtlMs; + private volatile int sinkParallelism = 0; + private final Map<String, TimestampedPendingRequest> pendingRequests = + new ConcurrentHashMap<>(); + private final Map<String, Set<Integer>> receivedAcks = new ConcurrentHashMap<>(); + + private final AtomicLong totalRequests = new AtomicLong(0); + private final AtomicLong completedRequests = new AtomicLong(0); + private final AtomicLong timeoutRequests = new AtomicLong(0); + + private LocalSchemaCoordinator(String jobId, long requestTtlMs) { + this.jobId = jobId; + this.requestTtlMs = requestTtlMs; + + cleanupExecutor.scheduleWithFixedDelay( + this::performPeriodicCleanup, + CLEANUP_INTERVAL_MS, + CLEANUP_INTERVAL_MS, + TimeUnit.MILLISECONDS); + + log.info( + "Created LocalSchemaCoordinator for jobId: {} with TTL: {}ms", jobId, requestTtlMs); + } + + public static LocalSchemaCoordinator getInstance(String jobId) { + if (jobId == null || jobId.trim().isEmpty()) { + throw new IllegalArgumentException("JobId cannot be null or empty"); + } + + return instances + .compute( + jobId, + (key, weakRef) -> { + LocalSchemaCoordinator coordinator = null; + if (weakRef != null) { + coordinator = weakRef.get(); + } + + if (coordinator == null) { + coordinator = + new LocalSchemaCoordinator(jobId, DEFAULT_REQUEST_TTL_MS); + log.info( + "Created new LocalSchemaCoordinator instance for jobId: {}", + jobId); + } + + return new WeakReference<>(coordinator); + }) + .get(); + } + + public void registerSinkParallelism(int parallelism) { + this.sinkParallelism = parallelism; + log.info( + "Registered sink parallelism: {} for schema change coordination in jobId: {}", + parallelism, + jobId); + } + + public boolean requestSchemaChange(TableIdentifier tableId, long epoch, long timeoutMs) + throws InterruptedException, SchemaCoordinationException { + String key = tableId.toString() + "#" + epoch; + int expectedAcks = sinkParallelism; + if (expectedAcks == 0) { + log.warn( + "Sink parallelism not registered yet. Cannot coordinate schema change for table {} (epoch {}). " + + "Assuming success to avoid deadlock.", + tableId, + epoch); + return true; + } + log.info( + "Requesting schema change for table {} (epoch {}). Waiting for all {} sink subtasks to apply after checkpoint completion.", + tableId, + epoch, + expectedAcks); + + long now = System.currentTimeMillis(); + TimestampedPendingRequest request = + new TimestampedPendingRequest( + tableId, epoch, expectedAcks, now, Math.min(timeoutMs, requestTtlMs)); + + pendingRequests.put(key, request); + receivedAcks.put(key, ConcurrentHashMap.newKeySet()); + + try { + Boolean result = request.future.get(timeoutMs, TimeUnit.MILLISECONDS); + if (result == null) { + throw SchemaCoordinationException.conflict(tableId, jobId, jobId); + } + if (!result) { + throw SchemaCoordinationException.conflict(tableId, jobId, jobId); + } + return result; + } catch (TimeoutException e) { + log.error( + "Schema change request for table {} (epoch {}) timed out after {}ms. " + + "Checkpoint may not have completed in time.", + tableId, + epoch, + timeoutMs); + request.future.cancel(true); + throw SchemaCoordinationException.timeout(tableId, jobId, timeoutMs / 1000, e); + } catch (ExecutionException e) { + log.error( + "Schema change request for table {} (epoch {}) failed with execution exception.", + tableId, + epoch, + e); + throw new SchemaEvolutionException( + SchemaEvolutionErrorCode.SCHEMA_EVENT_PROCESSING_FAILED, + e.getMessage(), + tableId, + jobId, + e); + } finally { + pendingRequests.remove(key); + receivedAcks.remove(key); + } + } + + public void notifySchemaChangeApplied( + TableIdentifier tableId, long epoch, int subtaskId, boolean success) { + String key = tableId.toString() + "#" + epoch; + TimestampedPendingRequest request = pendingRequests.get(key); + + if (request == null) { + log.warn( + "Received application notification for unknown schema change request: table {} (epoch {}), subtask {}", + tableId, + epoch, + subtaskId); + return; + } + + // check if this subtask already applied + Set<Integer> appliedSubtasks = receivedAcks.get(key); + if (appliedSubtasks == null) { + log.warn( + "Received application notification but no ack set found for table {} (epoch {}), subtask {}", + tableId, + epoch, + subtaskId); + return; + } + + if (appliedSubtasks.contains(subtaskId)) { + log.warn( + "Subtask {} already applied schema change for table {} (epoch {}). Ignoring duplicate notification.", + subtaskId, + tableId, + epoch); + return; + } + + appliedSubtasks.add(subtaskId); + log.info( + "Subtask {} applied schema change for table {} (epoch {}), success: {}. {}/{} subtasks applied.", + subtaskId, + tableId, + epoch, + success, + appliedSubtasks.size(), + request.expectedAcks); + + if (!success) { + request.allSuccess.set(false); + } + + // if all subtasks have applied, complete the future + if (appliedSubtasks.size() >= request.expectedAcks && !request.appliedPhaseComplete) { + request.appliedPhaseComplete = true; + boolean allSuccess = request.allSuccess.get(); + request.future.complete(allSuccess); + log.info( + "All {} subtasks have applied schema change for table {} (epoch {}). Completing request with result: {}", + request.expectedAcks, + tableId, + epoch, + allSuccess); + } + } + + private void performPeriodicCleanup() { Review Comment: Under what circumstances will an expired task occur? ########## seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java: ########## @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.translation.flink.schema; + +import org.apache.seatunnel.shade.com.typesafe.config.Config; + +import org.apache.seatunnel.api.source.SupportSchemaEvolution; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.schema.SchemaChangeType; +import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent; +import org.apache.seatunnel.api.table.schema.event.TableEvent; +import org.apache.seatunnel.api.table.schema.exception.SchemaValidationException; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.translation.flink.schema.coordinator.LocalSchemaCoordinator; + +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; + +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; + +/** operators added to the source and transformer pipelines to handle schema evolution */ +@Slf4j +public class SchemaOperator extends AbstractStreamOperator<SeaTunnelRow> + implements OneInputStreamOperator<SeaTunnelRow, SeaTunnelRow> { + private final Map<TableIdentifier, CatalogTable> localSchemaState; + private String jobId; + private final SupportSchemaEvolution source; + private final Config pluginConfig; + private volatile Long lastProcessedEventTime; + private transient LocalSchemaCoordinator coordinator; + private transient Map<String, List<BufferedDataRow>> bufferedDataRows; + private volatile boolean schemaChangePending = false; + private volatile CompletableFuture<Boolean> pendingSchemaFuture = null; + + public SchemaOperator(String jobId, SupportSchemaEvolution source, Config pluginConfig) { + this.jobId = jobId; + this.source = source; + this.pluginConfig = pluginConfig; + this.localSchemaState = new ConcurrentHashMap<>(); + } + + @Override + public void open() throws Exception { + super.open(); + String flinkJobId = getRuntimeContext().getJobId().toString(); + if (!flinkJobId.equals(this.jobId)) { + this.jobId = flinkJobId; + } + this.bufferedDataRows = new ConcurrentHashMap<>(); + this.coordinator = LocalSchemaCoordinator.getInstance(this.jobId); + log.info("SchemaOperator opened for job: {}", this.jobId); + } + + @Override + public void processElement(StreamRecord<SeaTunnelRow> streamRecord) { + SeaTunnelRow element = streamRecord.getValue(); + + if (pluginConfig.hasPath("schema-changes.enabled")) { + output.collect(new StreamRecord<>(element, streamRecord.getTimestamp())); + return; + } + + if ("__SCHEMA_CHANGE_EVENT__".equals(element.getTableId()) + && element.getOptions() != null) { + Object object = element.getOptions().get("schema_change_event"); + if (object instanceof SchemaChangeEvent) { + handleSchemaChangeEvent((SchemaChangeEvent) object); + return; + } + } + + if (schemaChangePending && pendingSchemaFuture != null) { + String tableId = element.getTableId(); + if (tableId != null) { + String key = tableId + "#" + System.currentTimeMillis(); + bufferedDataRows + .computeIfAbsent(key, k -> new ArrayList<>()) + .add(new BufferedDataRow(element, streamRecord.getTimestamp())); + return; + } + } + + output.collect(new StreamRecord<>(element, streamRecord.getTimestamp())); + } + + private void handleSchemaChangeEvent(SchemaChangeEvent schemaChangeEvent) { + List<SchemaChangeType> supportedTypes = source.supports(); + if (supportedTypes == null || supportedTypes.isEmpty()) { + log.info( + "Source: {} does not support any schema change types, skipping schema change event", + source); + return; + } + + if (!isSchemaChangeSupported(schemaChangeEvent, supportedTypes)) { + log.warn( + "Schema change type {} not supported by source {}, skipping", + schemaChangeEvent.getEventType(), + source); + return; + } + + processSchemaChangeEvent(schemaChangeEvent); + } + + private boolean isSchemaChangeSupported( + SchemaChangeEvent event, List<SchemaChangeType> supportedTypes) { + switch (event.getEventType()) { + case SCHEMA_CHANGE_ADD_COLUMN: + return supportedTypes.contains(SchemaChangeType.ADD_COLUMN); + case SCHEMA_CHANGE_DROP_COLUMN: + return supportedTypes.contains(SchemaChangeType.DROP_COLUMN); + case SCHEMA_CHANGE_MODIFY_COLUMN: + return supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN); + case SCHEMA_CHANGE_CHANGE_COLUMN: + return supportedTypes.contains(SchemaChangeType.RENAME_COLUMN); + case SCHEMA_CHANGE_UPDATE_COLUMNS: + return supportedTypes.contains(SchemaChangeType.ADD_COLUMN) + || supportedTypes.contains(SchemaChangeType.DROP_COLUMN) + || supportedTypes.contains(SchemaChangeType.UPDATE_COLUMN) + || supportedTypes.contains(SchemaChangeType.RENAME_COLUMN); + default: + log.error("Unknown schema change event type: {}", event.getEventType()); + throw SchemaValidationException.unsupportedChangeType( + event.tableIdentifier(), jobId); + } + } + + private void processSchemaChangeEvent(SchemaChangeEvent schemaChangeEvent) { + TableIdentifier tableId = schemaChangeEvent.tableIdentifier(); + long eventTime = schemaChangeEvent.getCreatedTime(); + + try { + if (lastProcessedEventTime != null && eventTime <= lastProcessedEventTime) { + throw SchemaValidationException.outdatedEvent( + tableId, jobId, eventTime, lastProcessedEventTime); + } + + if (schemaChangeEvent instanceof TableEvent) { + schemaChangeEvent.setJobId(jobId); + } + + log.info( + "Starting async schema change processing for table: {}, job: {}, event time: {}", + tableId, + jobId, + eventTime); + + String key = tableId.toString() + "#" + eventTime; + schemaChangePending = true; + bufferedDataRows.put(key, new ArrayList<>()); + + sendSchemaChangeEventToDownstream(schemaChangeEvent); + CatalogTable newSchema = schemaChangeEvent.getChangeAfter(); + if (newSchema != null) { + localSchemaState.put(tableId, newSchema); + log.debug("Updated local schema state for table: {}", tableId); + } + lastProcessedEventTime = eventTime; + + pendingSchemaFuture = + CompletableFuture.supplyAsync( + () -> { + try { + log.info( + "Waiting for schema change confirmation for table {} (epoch {}). Business data buffered, checkpoint barriers can pass.", + tableId, + eventTime); + long timeoutMs = 300_000L; + boolean success = + coordinator.requestSchemaChange( + tableId, eventTime, timeoutMs); + + if (success) { + log.info( + "Schema change for table {} (epoch {}) confirmed successfully by all sink subtasks via checkpoint completion.", + tableId, + eventTime); + } else { + log.error( + "Schema change for table {} (epoch {}) failed or timed out.", + tableId, + eventTime); + } + + return success; + } catch (Exception e) { + log.error( + "Error during async schema change processing for table {} (epoch {})", + tableId, + eventTime, + e); + return false; + } + }); + pendingSchemaFuture.whenComplete( + (success, throwable) -> { + try { + schemaChangePending = false; + pendingSchemaFuture = null; + + if (throwable != null) { + log.error( + "Schema change future completed with exception", throwable); + } + + List<BufferedDataRow> bufferedRows = bufferedDataRows.remove(key); + if (bufferedRows != null && !bufferedRows.isEmpty()) { + log.info( + "Releasing {} buffered data rows after schema change processing for table {}", + bufferedRows.size(), + tableId); + for (BufferedDataRow buffered : bufferedRows) { + output.collect( + new StreamRecord<>(buffered.row, buffered.timestamp)); + } + } + + log.info( + "Async schema change processing completed for table {}, data flow resumed", + tableId); + + } catch (Exception e) { + log.error("Error during schema change completion handling", e); + } + }); + + log.info( + "Async schema change processing initiated for table {}. Checkpoint barriers can propagate normally.", + tableId); + + } catch (Exception e) { + log.error("Error starting async schema change processing", e); + schemaChangePending = false; + pendingSchemaFuture = null; + } + } + + private void sendSchemaChangeEventToDownstream(SchemaChangeEvent schemaChangeEvent) { Review Comment: ```suggestion private void sendSchemaChangeEventToDownStream(SchemaChangeEvent schemaChangeEvent) { ``` -- 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]
