This is an automated email from the ASF dual-hosted git repository.
dybyte pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new a9ef95a276 [Fix] [Connector-V2] Fix Flink schema evolution hang caused
by XA transaction MDL deadlock with MySQL CDC (#10648)
a9ef95a276 is described below
commit a9ef95a276f049f5436fab0401c0ffe12c761bd9
Author: cloverdew <[email protected]>
AuthorDate: Sun May 24 01:52:59 2026 +0800
[Fix] [Connector-V2] Fix Flink schema evolution hang caused by XA
transaction MDL deadlock with MySQL CDC (#10648)
---
.../cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java | 95 ++-
...ysql_with_flink_schema_change_exactly_once.conf | 55 ++
.../translation/flink/sink/FlinkSinkWriter.java | 1 -
.../flink/schema/BroadcastSchemaSinkOperator.java | 42 +-
.../translation/flink/schema/SchemaOperator.java | 934 +++++++--------------
.../schema/coordinator/LocalSchemaCoordinator.java | 63 +-
.../translation/flink/sink/FlinkSinkWriter.java | 1 -
.../flink/schema/SchemaOperatorTest.java | 400 +++++++++
8 files changed, 950 insertions(+), 641 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
index 98bbd94063..37fea77fe2 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
@@ -148,6 +148,97 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
assertTableStructureAndData(MYSQL_DATABASE, SOURCE_TABLE, SINK_TABLE);
}
+ @Order(2)
+ @TestTemplate
+ public void testMysqlCdcWithSchemaEvolutionCaseExactlyOnce(TestContainer
container) {
+
+ shopDatabase.setTemplateName("shop").createAndInitialize();
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/mysqlcdc_to_mysql_with_flink_schema_change_exactly_once.conf");
+ } catch (Exception e) {
+ log.error("Commit task exception :" + e.getMessage());
+ throw new RuntimeException(e);
+ }
+ });
+
+ assertSchemaEvolution(MYSQL_DATABASE, SOURCE_TABLE, SINK_TABLE2);
+ }
+
+ private void assertSchemaEvolution(String database, String sourceTable,
String sinkTable) {
+ await().atMost(180000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+ query(String.format(QUERY, database,
sourceTable)),
+ query(String.format(QUERY, database,
sinkTable))));
+
+ // case1 add columns with cdc data at same time
+ shopDatabase.setTemplateName("add_columns").createAndInitialize();
+ await().atMost(180000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+ query(String.format(DESC, database,
sourceTable)),
+ query(String.format(DESC, database,
sinkTable))));
+ await().atMost(180000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertIterableEquals(
+ query(
+ String.format(QUERY, database,
sourceTable)
+ + " where id >= 128"),
+ query(
+ String.format(QUERY, database,
sinkTable)
+ + " where id >= 128"));
+
+ Assertions.assertIterableEquals(
+ query(String.format(PROJECTION_QUERY,
database, sourceTable)),
+ query(String.format(PROJECTION_QUERY,
database, sinkTable)));
+
+ // The default value of add_column4 is
current_timestamp()，so the
+ // history data of sink table with this column may
be different from the
+ // source table because delay of apply schema
change.
+ String query =
+ String.format(
+ "SELECT t1.id AS table1_id,
t1.add_column4 AS table1_timestamp, "
+ + "t2.id AS table2_id,
t2.add_column4 AS table2_timestamp, "
+ +
"ABS(TIMESTAMPDIFF(SECOND, t1.add_column4, t2.add_column4)) AS time_diff "
+ + "FROM %s.%s t1 "
+ + "INNER JOIN %s.%s t2 ON
t1.id = t2.id",
+ database, sourceTable, database,
sinkTable);
+ try (Connection jdbcConnection =
getJdbcConnection();
+ Statement statement =
jdbcConnection.createStatement();
+ ResultSet resultSet =
statement.executeQuery(query); ) {
+ while (resultSet.next()) {
+ int timeDiff =
resultSet.getInt("time_diff");
+ Assertions.assertTrue(
+ timeDiff <= 60,
+ "Time difference exceeds 60
seconds: "
+ + timeDiff
+ + " seconds");
+ }
+ }
+ });
+
+ // case2 drop columns with cdc data at same time
+ assertCaseByDdlName("drop_columns", database, sourceTable, sinkTable);
+
+ // case3 change column name with cdc data at same time
+ assertCaseByDdlName("change_columns", database, sourceTable,
sinkTable);
+
+ // case4 modify column data type with cdc data at same time
+ assertCaseByDdlName("modify_columns", database, sourceTable,
sinkTable);
+ }
+
+ private void assertCaseByDdlName(
+ String drop_columns, String database, String sourceTable, String
sinkTable) {
+ shopDatabase.setTemplateName(drop_columns).createAndInitialize();
+ assertTableStructureAndData(database, sourceTable, sinkTable);
+ }
+
private void assertSchemaEvolutionForAddColumns(
String database, String sourceTable, String sinkTable) {
await().atMost(180000, TimeUnit.MILLISECONDS)
@@ -197,8 +288,8 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
while (resultSet.next()) {
int timeDiff =
resultSet.getInt("time_diff");
Assertions.assertTrue(
- timeDiff <= 6,
- "Time difference exceeds 6
seconds: "
+ timeDiff <= 60,
+ "Time difference exceeds 60
seconds: "
+ timeDiff
+ " seconds");
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_flink_schema_change_exactly_once.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_flink_schema_change_exactly_once.conf
new file mode 100644
index 0000000000..09c88f3120
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_flink_schema_change_exactly_once.conf
@@ -0,0 +1,55 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in
seatunnel config
+######
+
+env {
+ # You can set engine configuration here
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+ read_limit.bytes_per_second=7000000
+ read_limit.rows_per_second=400
+}
+
+source {
+ MySQL-CDC {
+ server-id = 5652-5657
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["shop.products"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+
+ schema-changes.enabled = true
+ }
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ generate_sink_sql = true
+ database = shop
+ table = mysql_cdc_e2e_sink_table_with_schema_change_exactly_once
+ primary_keys = ["id"]
+ is_exactly_once = true
+ xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
+ }
+}
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
index 5241748f5e..e1e063bf19 100644
---
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
@@ -130,7 +130,6 @@ public class FlinkSinkWriter<CommT, WriterStateT>
"FlinkSinkWriter applying SchemaChangeEvent for table: {}",
schemaChangeEvent.tableIdentifier());
- sinkWriter.prepareCommit();
if (!(sinkWriter instanceof SupportSchemaEvolutionSinkWriter)) {
log.warn(
"Sink writer {} does not support schema evolution,
ignoring SchemaChangeEvent for table: {}",
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/BroadcastSchemaSinkOperator.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/BroadcastSchemaSinkOperator.java
index b4cc194211..7594737c96 100644
---
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/BroadcastSchemaSinkOperator.java
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/BroadcastSchemaSinkOperator.java
@@ -160,33 +160,35 @@ public class BroadcastSchemaSinkOperator extends
AbstractStreamOperator<SeaTunne
}
int subtaskId = getRuntimeContext().getIndexOfThisSubtask();
log.info(
- "Subtask {} applying schema change immediately for table
{} (epoch {}, change: {}). This prevents deadlock by allowing checkpoint
barriers to propagate.",
+ "Subtask {} forwarding schema change to sink for table {}
(epoch {}, change: {}).",
subtaskId,
tableId,
epoch,
event.getClass().getSimpleName());
- try {
- emitApplySchemaEventToSink(event, epoch);
- lastProcessedEpoch.put(tableId, epoch);
+ // Forward to FlinkSinkWriter which performs the actual ALTER
TABLE and sends
+ // ACK to coordinator after completion. We do NOT send ACK here
because
+ // output.collect() may be asynchronous if operators are not
chained, and
+ // sending ACK before ALTER TABLE finishes would cause
SchemaOperator to
+ // release new-schema data before the sink table is actually
altered.
+ emitApplySchemaEventToSink(event, epoch);
+ lastProcessedEpoch.put(tableId, epoch);
- // send ACK to coordinator indicating this subtask has
processed the schema change
- coordinator.notifySchemaChangeApplied(tableId, epoch,
subtaskId, true);
-
- log.info(
- "Subtask {} processed schema change for table {}
(epoch {}) and sent ACK to coordinator.",
- subtaskId,
- tableId,
- epoch);
- } catch (Exception e) {
- coordinator.notifySchemaChangeApplied(tableId, epoch,
subtaskId, false);
- throw e;
- }
+ log.info(
+ "Subtask {} forwarded schema change for table {} (epoch
{}) to sink writer. "
+ + "ACK will be sent by FlinkSinkWriter after ALTER
TABLE completes.",
+ subtaskId,
+ tableId,
+ epoch);
} catch (SchemaValidationException | SchemaCoordinationException e) {
log.error("Schema broadcast or coordination error", e);
+ coordinator.notifySchemaChangeApplied(
+ tableId, epoch,
getRuntimeContext().getIndexOfThisSubtask(), false);
throw e;
} catch (Exception e) {
log.error("Schema change dispatch failed", e);
+ coordinator.notifySchemaChangeApplied(
+ tableId, epoch,
getRuntimeContext().getIndexOfThisSubtask(), false);
throw new SchemaEvolutionException(
SchemaEvolutionErrorCode.SCHEMA_EVENT_PROCESSING_FAILED,
e.getMessage(),
@@ -214,9 +216,11 @@ public class BroadcastSchemaSinkOperator extends
AbstractStreamOperator<SeaTunne
@Override
public void close() throws Exception {
+ int subtaskId = getRuntimeContext().getIndexOfThisSubtask();
+ if (coordinator != null) {
+ coordinator.unregisterSinkSubtask(subtaskId);
+ }
super.close();
- log.info(
- "BroadcastSchemaSinkOperator closed on subtask {}",
- getRuntimeContext().getIndexOfThisSubtask());
+ log.info("BroadcastSchemaSinkOperator closed on subtask {}",
subtaskId);
}
}
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
index 3948e1c95a..0f993dcda0 100644
---
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/SchemaOperator.java
@@ -25,6 +25,8 @@ 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.SchemaEvolutionErrorCode;
+import
org.apache.seatunnel.api.table.schema.exception.SchemaEvolutionException;
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;
@@ -42,34 +44,50 @@ import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import java.io.Serializable;
-import java.util.ArrayList;
import java.util.HashMap;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.CompletableFuture;
+import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
-/** operators added to the source and transformer pipelines to handle schema
evolution */
+/**
+ * Operator placed after the source to handle schema evolution.
+ *
+ * <p>schema change events are NOT processed synchronously in {@link
#processElement}. Instead, they
+ * are buffered and deferred until an additional checkpoint cycle has
completed after the first
+ * checkpoint that observed the pending DDL. This wait ensures that when the
sink executes ALTER
+ * TABLE, all XA transactions from prior checkpoint cycles have been fully
committed by the {@code
+ * FlinkGlobalCommitter} (which runs asynchronously after {@code
notifyCheckpointComplete}), so
+ * their metadata locks are released and the ALTER TABLE can acquire an
exclusive MDL lock without
+ * deadlock.
+ *
+ * <p>Per checkpoint cycle, at most ONE schema change is applied. If multiple
DDLs arrive between
+ * two checkpoints, they are processed across successive checkpoint cycles.
+ */
@Slf4j
public class SchemaOperator extends AbstractStreamOperator<SeaTunnelRow>
implements OneInputStreamOperator<SeaTunnelRow, SeaTunnelRow> {
- private static final int MAX_BUFFERED_ROWS_PER_KEY = 100000;
+ private static final int MAX_BUFFERED_RECORDS = 100000;
+ private static final long SCHEMA_CHANGE_TIMEOUT_MS = 300_000L;
+ private static final int CHECKPOINT_WAIT_ROUNDS = 1;
+
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 transient Queue<BufferedRecord> pendingQueue;
private volatile boolean schemaChangePending = false;
- private volatile CompletableFuture<Boolean> pendingSchemaFuture = null;
- private volatile boolean stateDirty = false;
+ private long firstSeenCheckpointId = -1L;
private transient ListState<SchemaStateEntry> localSchemaStateStore;
private transient ListState<Long> lastProcessedEventTimeState;
private transient ListState<Boolean> schemaChangePendingState;
- private transient ListState<BufferedDataEntry> bufferedDataRowsState;
+ private transient ListState<BufferedRecordEntry> bufferedRecordsState;
+ private transient ListState<Long> firstSeenCheckpointIdState;
public SchemaOperator(String jobId, SupportSchemaEvolution source, Config
pluginConfig) {
this.jobId = jobId;
@@ -85,20 +103,16 @@ public class SchemaOperator extends
AbstractStreamOperator<SeaTunnelRow>
if (!flinkJobId.equals(this.jobId)) {
this.jobId = flinkJobId;
}
- this.bufferedDataRows = new ConcurrentHashMap<>();
- this.coordinator = LocalSchemaCoordinator.getInstance(this.jobId);
-
- // if schema change was pending and we have buffered data, handle
recovery scenario
- if (schemaChangePending && pendingSchemaFuture == null) {
- handleSchemaChangeRecovery();
+ if (this.pendingQueue == null) {
+ this.pendingQueue = new LinkedList<>();
}
+ this.coordinator = LocalSchemaCoordinator.getInstance(this.jobId);
log.info(
- "SchemaOperator opened for job: {}, recovered state -
lastProcessedEventTime: {}, schemaChangePending: {}, bufferedDataRows size: {}",
+ "SchemaOperator opened for job: {}, schemaChangePending: {},
pendingQueue size: {}",
this.jobId,
- this.lastProcessedEventTime,
this.schemaChangePending,
- bufferedDataRows.size());
+ this.pendingQueue.size());
}
@Override
@@ -110,686 +124,384 @@ public class SchemaOperator extends
AbstractStreamOperator<SeaTunnelRow>
return;
}
+ // detect schema change events
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);
+ handleSchemaChangeDetected((SchemaChangeEvent) object,
streamRecord.getTimestamp());
return;
}
}
+ // while a schema change is pending, buffer ALL subsequent records
if (schemaChangePending) {
- String tableId = element.getTableId();
- if (tableId != null && lastProcessedEventTime != null) {
- String key = createKey(tableId, lastProcessedEventTime);
- bufferedDataRows(key, element, streamRecord.getTimestamp());
- return;
- }
+ enqueueDataRecord(element, streamRecord.getTimestamp());
+ return;
}
output.collect(streamRecord);
}
- private boolean isSchemaEvolutionEnabled(Config pluginConfig) {
- if (pluginConfig.hasPath("schema-changes.enabled")) {
- return pluginConfig.getBoolean("schema-changes.enabled");
- }
-
- return false;
- }
-
- private String createKey(String tableId, Long eventTime) {
- return tableId + "#" + eventTime;
- }
-
- private void bufferedDataRows(String key, SeaTunnelRow element, long
timestamp) {
- try {
- BufferedDataRow bufferedRow = new BufferedDataRow(element,
timestamp);
-
- synchronized (this) {
- List<BufferedDataRow> bufferedList =
- bufferedDataRows.computeIfAbsent(key, k -> new
ArrayList<>());
-
- if (bufferedList.size() >= MAX_BUFFERED_ROWS_PER_KEY) {
- log.warn(
- "Buffer for key {} exceeded max size {}, dropping
oldest row",
- key,
- MAX_BUFFERED_ROWS_PER_KEY);
- bufferedList.remove(0);
- }
-
- bufferedList.add(bufferedRow);
- stateDirty = true;
-
- log.debug(
- "buffered data row for key: {}, total buffered: {}",
- key,
- bufferedList.size());
- }
- } catch (Exception e) {
- log.error("Failed to buffer data for key: {}, dropping this data
row", key, e);
- }
- }
-
- private void handleSchemaChangeEvent(SchemaChangeEvent schemaChangeEvent) {
+ private void handleSchemaChangeDetected(SchemaChangeEvent event, long
timestamp) {
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);
+ log.info("Source does not support any schema change types,
skipping");
return;
}
-
- if (!isSchemaChangeSupported(schemaChangeEvent, supportedTypes)) {
- log.warn(
- "Schema change type {} not supported by source {},
skipping",
- schemaChangeEvent.getEventType(),
- source);
+ if (!isSchemaChangeSupported(event, supportedTypes)) {
+ log.warn("Schema change type {} not supported, skipping",
event.getEventType());
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);
+ if (event instanceof TableEvent) {
+ event.setJobId(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 schema change processing for table: {}, job: {},
event time: {}",
- tableId,
- jobId,
- eventTime);
-
- String key = createKey(tableId.toString(), eventTime);
-
- // initialize buffer for this schema change
- synchronized (this) {
- List<BufferedDataRow> newBufferList = new ArrayList<>();
- bufferedDataRows.put(key, newBufferList);
- stateDirty = true;
- }
-
- schemaChangePending = true;
-
- sendSchemaChangeEventToDownstream(schemaChangeEvent);
- CatalogTable newSchema = schemaChangeEvent.getChangeAfter();
- if (newSchema != null) {
- localSchemaState.put(tableId, newSchema);
- log.debug("Updated local schema state for table: {}", tableId);
- }
- lastProcessedEventTime = eventTime;
-
- try {
- log.info(
- "Synchronously processing schema change for table {}
(epoch {}). Business data buffered.",
- tableId,
- eventTime);
- long timeoutMs = 300_000L;
- boolean success = coordinator.requestSchemaChange(tableId,
eventTime, timeoutMs);
-
- if (success) {
- if (schemaChangeEvent.getChangeAfter() != null) {
- localSchemaState.put(tableId,
schemaChangeEvent.getChangeAfter());
- }
- lastProcessedEventTime = eventTime;
- log.info(
- "Schema change for table {} (epoch {}) confirmed
successfully by all sink subtasks.",
- tableId,
- eventTime);
- } else {
- log.error(
- "Schema change for table {} (epoch {}) failed or
timed out.",
- tableId,
- eventTime);
- }
- } catch (Exception e) {
- log.error(
- "Error during synchronous schema change processing for
table {} (epoch {})",
- tableId,
- eventTime,
- e);
- } finally {
- schemaChangePending = false;
- pendingSchemaFuture = null;
- releaseBufferedData(key, tableId);
+ log.info(
+ "Schema change detected for table {} (epoch {}). "
+ + "Deferring until next checkpoint completes to avoid
XA/MDL deadlock.",
+ event.tableIdentifier(),
+ event.getCreatedTime());
- log.info(
- "Synchronous schema change processing completed for
table {}, data flow resumed",
- tableId);
- }
+ pendingQueue.add(BufferedRecord.schemaChange(event));
+ schemaChangePending = true;
+ }
- log.info(
- "Synchronous schema change processing completed for table
{}. Checkpoint barriers can propagate normally.",
- tableId);
- } catch (Exception e) {
- log.error("Error starting schema change processing", e);
- schemaChangePending = false;
- try {
- schemaChangePendingState.clear();
- schemaChangePendingState.add(false);
- } catch (Exception stateException) {
- log.error(
- "Error updating schemaChangePending state during error
handling",
- stateException);
- }
- pendingSchemaFuture = null;
+ private void enqueueDataRecord(SeaTunnelRow row, long timestamp) {
+ if (pendingQueue.size() >= MAX_BUFFERED_RECORDS) {
+ TableIdentifier tableIdentifier =
getPendingSchemaTableIdentifier();
+ throw new SchemaEvolutionException(
+ SchemaEvolutionErrorCode.SCHEMA_EVENT_PROCESSING_FAILED,
+ String.format(
+ "Pending schema buffer overflow (max=%d). "
+ + "Failing fast to avoid dropping schema
change control events.",
+ MAX_BUFFERED_RECORDS),
+ tableIdentifier,
+ jobId);
}
+ pendingQueue.add(BufferedRecord.data(row, timestamp));
}
- private void releaseBufferedData(String key, TableIdentifier tableId) {
- try {
- List<BufferedDataRow> bufferedRows;
- synchronized (this) {
- bufferedRows = bufferedDataRows.remove(key);
- stateDirty = true;
- }
-
- 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(
- "Successfully released {} buffered rows for table {}",
- bufferedRows.size(),
- tableId);
+ private TableIdentifier getPendingSchemaTableIdentifier() {
+ for (BufferedRecord record : pendingQueue) {
+ if (record.isSchemaChange && record.schemaEvent != null) {
+ return record.schemaEvent.tableIdentifier();
}
-
- } catch (Exception e) {
- log.error(
- "CRITICAL: Failed to release buffered data for key: {}. "
- + "Data may be lost if this continues to fail!",
- key,
- e);
-
- try {
- Iterable<BufferedDataEntry> stateEntries =
bufferedDataRowsState.get();
- for (BufferedDataEntry entry : stateEntries) {
- if (entry.key.equals(key)) {
- List<BufferedDataRow> stateData = entry.bufferedRows;
- if (stateData != null && !stateData.isEmpty()) {
- synchronized (this) {
- bufferedDataRows.put(key, new
ArrayList<>(stateData));
- stateDirty = true;
- }
- log.info(
- "Restored {} rows to memory buffer for
retry",
- stateData.size());
- }
- break;
- }
- }
- } catch (Exception restoreException) {
- log.error("Failed to restore buffered data to memory",
restoreException);
- }
-
- throw e;
}
+ return null;
}
- private void handleSchemaChangeRecovery() {
- log.info(
- "Detected schema change pending after recovery with {}
buffered entries. "
- + "Querying sink state to determine correct recovery
action.",
- bufferedDataRows.size());
-
- try {
- // wait for sink operators to register their state providers with
retry mechanism
- waitForSinkStateProviders(10, 500);
-
- boolean allDataReleased = true;
- int totalReleased = 0;
-
- for (Map.Entry<String, List<BufferedDataRow>> entry :
bufferedDataRows.entrySet()) {
- String key = entry.getKey();
- List<BufferedDataRow> bufferedRows = entry.getValue();
-
- if (bufferedRows == null || bufferedRows.isEmpty()) {
- continue;
- }
-
- String[] keyParts = key.split("#");
- if (keyParts.length != 2) {
- log.warn("Invalid buffer key format: {}, releasing data",
key);
- releaseBufferedDataForKey(key, bufferedRows);
- totalReleased += bufferedRows.size();
- continue;
- }
+ /**
+ * Called by Flink after a checkpoint succeeds. Uses an extra completed
checkpoint round to
+ * ensure safety:
+ *
+ * <ul>
+ * <li><b>first time seeing the DDL: record {@link
#firstSeenCheckpointId} but do NOT
+ * broadcast the DDL yet. At this point the {@code
FlinkGlobalCommitter} may still be
+ * running {@code XA COMMIT} for this checkpoint's prepared
transactions, holding MDL
+ * locks on the sink table.
+ * <li><b>{@code checkpointId >= firstSeenCheckpointId +
CHECKPOINT_WAIT_ROUNDS} : the XA
+ * COMMIT from the earlier checkpoint cycle is guaranteed to have
finished (at least one
+ * additional checkpoint cycle has completed, which implies the
committer ran). The sink's
+ * ALTER TABLE will not encounter MDL lock, it is now safe to
broadcast the DDL.
+ * </ul>
+ */
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ super.notifyCheckpointComplete(checkpointId);
- String tableIdStr = keyParts[0];
- long epoch;
- try {
- epoch = Long.parseLong(keyParts[1]);
- } catch (NumberFormatException e) {
- log.warn("Invalid epoch in buffer key: {}, releasing
data", key);
- releaseBufferedDataForKey(key, bufferedRows);
- totalReleased += bufferedRows.size();
- continue;
- }
- TableIdentifier tableId;
- String[] parts = tableIdStr.split("\\.");
- if (parts.length < 3) {
- throw new IllegalArgumentException("Invalid table id
format: " + tableIdStr);
- }
- tableId = TableIdentifier.of(parts[0], parts[1], parts[2]);
-
- // query sink processing status using string representation
directly
- LocalSchemaCoordinator.SchemaProcessingStatus status =
- coordinator.querySchemaProcessingStatus(tableId,
epoch);
-
- switch (status) {
- case FULLY_PROCESSED:
- log.info(
- "Schema change for table {} epoch {} fully
processed, releasing {} buffered rows",
- tableIdStr,
- epoch,
- bufferedRows.size());
- releaseBufferedDataForKey(key, bufferedRows);
- totalReleased += bufferedRows.size();
- break;
-
- case NOT_PROCESSED:
- log.info(
- "Schema change for table {} epoch {} not
processed, need to restart coordination for {} buffered rows",
- tableIdStr,
- epoch,
- bufferedRows.size());
- restartSchemaChangeCoordination(tableId, epoch, key);
- allDataReleased = false;
- break;
-
- case PARTIALLY_PROCESSED:
- log.warn(
- "Schema change for table {} epoch {} partially
processed, need to restart coordination for {} buffered rows",
- tableIdStr,
- epoch,
- bufferedRows.size());
- restartSchemaChangeCoordination(tableId, epoch, key);
- allDataReleased = false;
- break;
-
- default:
- log.error(
- "Unknown schema processing status: {},
releasing data to avoid deadlock",
- status);
- releaseBufferedDataForKey(key, bufferedRows);
- totalReleased += bufferedRows.size();
- }
- }
+ if (!schemaChangePending || pendingQueue.isEmpty()) {
+ return;
+ }
- // only reset schemaChangePending if all data was released
- if (allDataReleased) {
- schemaChangePending = false;
- schemaChangePendingState.clear();
- schemaChangePendingState.add(false);
- log.info(
- "Recovery completed: Released {} buffered data rows
and resumed normal data flow.",
- totalReleased);
- } else {
- log.info(
- "Recovery in progress: Released {} buffered data rows,
{} entries still need coordination.",
- totalReleased,
- bufferedDataRows.size());
- }
+ BufferedRecord head = pendingQueue.peek();
+ while (head != null && !head.isSchemaChange) {
+ output.collect(new StreamRecord<>(head.row, head.timestamp));
+ pendingQueue.poll();
+ head = pendingQueue.peek();
+ }
+ if (head == null) {
+ schemaChangePending = false;
+ firstSeenCheckpointId = -1L;
+ return;
+ }
- } catch (Exception e) {
- log.error(
- "Error during schema change recovery, releasing all
buffered data to avoid deadlock",
- e);
- releaseAllBufferedData();
+ // first time seeing this DDL at head of queue — just record the
checkpoint id
+ if (firstSeenCheckpointId < 0) {
+ firstSeenCheckpointId = checkpointId;
+ log.info(
+ "Checkpoint {} completed. DDL for table {} (epoch {})
first seen. "
+ + "Waiting {} more checkpoint round(s) for XA
COMMIT to finish.",
+ checkpointId,
+ head.schemaEvent.tableIdentifier(),
+ head.schemaEvent.getCreatedTime(),
+ CHECKPOINT_WAIT_ROUNDS);
+ return;
}
- }
- private void waitForSinkStateProviders(int maxRetries, long
retryIntervalMs)
- throws InterruptedException {
- for (int i = 0; i < maxRetries; i++) {
- if (coordinator.querySchemaProcessingStatus(
- TableIdentifier.of("test", "test", "test"), 0L)
- != null) {
- log.info("Sink state providers registered after {} retries",
i);
- return;
- }
- Thread.sleep(retryIntervalMs);
+ if (checkpointId < firstSeenCheckpointId + CHECKPOINT_WAIT_ROUNDS) {
+ log.info(
+ "Checkpoint {} completed. Still waiting for DDL on table
{} (epoch {}). "
+ + "Need checkpoint >= {} (first seen at {}, wait
rounds = {}).",
+ checkpointId,
+ head.schemaEvent.tableIdentifier(),
+ head.schemaEvent.getCreatedTime(),
+ firstSeenCheckpointId + CHECKPOINT_WAIT_ROUNDS,
+ firstSeenCheckpointId,
+ CHECKPOINT_WAIT_ROUNDS);
+ return;
}
- log.warn(
- "Sink state providers not fully registered after {} retries,
proceeding anyway",
- maxRetries);
- }
- private void releaseBufferedDataForKey(String key, List<BufferedDataRow>
bufferedRows) {
- try {
- for (BufferedDataRow buffered : bufferedRows) {
- output.collect(new StreamRecord<>(buffered.row,
buffered.timestamp));
- }
+ long waitedSince = firstSeenCheckpointId;
+ SchemaChangeEvent event = head.schemaEvent;
+ TableIdentifier tableId = event.tableIdentifier();
+ long eventTime = event.getCreatedTime();
- synchronized (this) {
- bufferedDataRows.remove(key);
- stateDirty = true;
- }
- } catch (Exception e) {
- log.error("Failed to release buffered data for key: {}", key, e);
+ log.info(
+ "Checkpoint {} completed (waited since checkpoint {}). "
+ + "Applying deferred schema change for table {} (epoch
{}).",
+ checkpointId,
+ waitedSince,
+ tableId,
+ eventTime);
+
+ if (lastProcessedEventTime != null && eventTime <=
lastProcessedEventTime) {
+ log.warn(
+ "Skipping outdated schema change event (epoch {} <= last
processed {})",
+ eventTime,
+ lastProcessedEventTime);
+ pendingQueue.poll();
+ firstSeenCheckpointId = -1L;
+ drainDataUntilNextSchemaChange();
+ return;
}
- }
- private void restartSchemaChangeCoordination(TableIdentifier tableId, long
epoch, String key) {
- try {
- log.info("Restarting schema change coordination for table {} epoch
{}", tableId, epoch);
-
- // create a new future for this coordination
- CompletableFuture<Boolean> newFuture =
- CompletableFuture.supplyAsync(
- () -> {
- try {
- long timeoutMs = 300_000L;
- boolean success =
- coordinator.requestSchemaChange(
- tableId, epoch, timeoutMs);
-
- if (success) {
- log.info(
- "Restarted schema change
coordination successful for table {} epoch {}",
- tableId,
- epoch);
- } else {
- log.error(
- "Restarted schema change
coordination failed for table {} epoch {}",
- tableId,
- epoch);
- }
-
- return success;
- } catch (Exception e) {
- log.error(
- "Error in restarted schema change
coordination for table {} epoch {}",
- tableId,
- epoch,
- e);
- return false;
- }
- });
-
- newFuture.whenComplete(
- (success, throwable) -> {
- try {
- if (throwable != null) {
- log.error(
- "Restarted schema change future
completed with exception",
- throwable);
- }
-
- // release the buffered data
- List<BufferedDataRow> bufferedRows =
bufferedDataRows.get(key);
- if (bufferedRows != null) {
- releaseBufferedDataForKey(key, bufferedRows);
- log.info(
- "Released {} buffered rows after
restarted coordination for key {}",
- bufferedRows.size(),
- key);
- }
-
- // check if this was the last pending coordination
- if (bufferedDataRows.isEmpty()) {
- schemaChangePending = false;
- schemaChangePendingState.clear();
- schemaChangePendingState.add(false);
- log.info(
- "All schema change coordination
completed, resumed normal data flow");
- }
-
- } catch (Exception e) {
- log.error("Error in restarted coordination
completion handling", e);
- }
- });
-
- if (pendingSchemaFuture == null) {
- pendingSchemaFuture = newFuture;
- }
+ sendSchemaChangeEventToDownstream(event);
- } catch (Exception e) {
- log.error(
- "Failed to restart schema change coordination for table {}
epoch {}, releasing data",
+ boolean success =
+ coordinator.requestSchemaChange(tableId, eventTime,
SCHEMA_CHANGE_TIMEOUT_MS);
+ if (!success) {
+ throw new SchemaEvolutionException(
+ SchemaEvolutionErrorCode.SCHEMA_EVENT_PROCESSING_FAILED,
+ String.format(
+ "Schema change for table %s (epoch %d) failed
during sink coordination.",
+ tableId, eventTime),
tableId,
- epoch,
- e);
- List<BufferedDataRow> bufferedRows = bufferedDataRows.get(key);
- if (bufferedRows != null) {
- releaseBufferedDataForKey(key, bufferedRows);
- }
+ jobId);
}
- }
-
- private void releaseAllBufferedData() {
- try {
- int totalReleased = 0;
- synchronized (this) {
- for (Map.Entry<String, List<BufferedDataRow>> entry :
bufferedDataRows.entrySet()) {
- List<BufferedDataRow> bufferedRows = entry.getValue();
- if (bufferedRows != null && !bufferedRows.isEmpty()) {
- for (BufferedDataRow buffered : bufferedRows) {
- output.collect(new StreamRecord<>(buffered.row,
buffered.timestamp));
- }
- totalReleased += bufferedRows.size();
- }
- }
-
- bufferedDataRows.clear();
- stateDirty = true;
- }
+ log.info(
+ "Schema change for table {} (epoch {}) confirmed by all sink
subtasks.",
+ tableId,
+ eventTime);
- schemaChangePending = false;
- schemaChangePendingState.clear();
- schemaChangePendingState.add(false);
+ pendingQueue.poll();
+ firstSeenCheckpointId = -1L;
- log.info("Emergency recovery: Released {} buffered data rows",
totalReleased);
- } catch (Exception e) {
- log.error("Failed to release all buffered data during emergency
recovery", e);
+ CatalogTable newSchema = event.getChangeAfter();
+ if (newSchema != null) {
+ localSchemaState.put(tableId, newSchema);
}
- }
+ lastProcessedEventTime = eventTime;
- private void sendSchemaChangeEventToDownstream(SchemaChangeEvent
schemaChangeEvent) {
- log.info(
- "Broadcasting SchemaChangeEvent to all downstream sink
subtasks for table: {}",
- schemaChangeEvent.tableIdentifier());
- SeaTunnelRow broadcastRow = new SeaTunnelRow(0);
- Map<String, Object> options = new HashMap<>();
- options.put("schema_change_broadcast", schemaChangeEvent);
- broadcastRow.setOptions(options);
+ drainDataUntilNextSchemaChange();
- output.collect(new StreamRecord<>(broadcastRow));
log.info(
- "SchemaChangeEvent broadcast sent for table: {}",
- schemaChangeEvent.tableIdentifier());
+ "Schema change for table {} (epoch {}) processing complete.
pendingQueue remaining: {}",
+ tableId,
+ eventTime,
+ pendingQueue.size());
}
- @Override
- public void close() throws Exception {
- try {
- if (pendingSchemaFuture != null && !pendingSchemaFuture.isDone()) {
- log.info("Cancelling ongoing schema change request during
close");
- pendingSchemaFuture.cancel(true);
+ private void drainDataUntilNextSchemaChange() {
+ int released = 0;
+ while (!pendingQueue.isEmpty()) {
+ BufferedRecord record = pendingQueue.peek();
+ if (record.isSchemaChange) {
+ // another DDL will stop here, wait for next checkpoint cycle
+ log.info(
+ "Released {} buffered data records. Another schema
change pending, "
+ + "waiting for next checkpoint.",
+ released);
+ return;
}
- } catch (Exception e) {
- log.warn("Error during SchemaOperator cleanup", e);
- } finally {
- super.close();
+ pendingQueue.poll();
+ output.collect(new StreamRecord<>(record.row, record.timestamp));
+ released++;
}
+
+ // queue is empty
+ schemaChangePending = false;
+ log.info("Released {} buffered data records. Normal data flow
resumed.", released);
}
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
super.snapshotState(context);
- try {
- // clear and update lastProcessedEventTime
- lastProcessedEventTimeState.clear();
- if (lastProcessedEventTime != null) {
- lastProcessedEventTimeState.add(lastProcessedEventTime);
- }
+ lastProcessedEventTimeState.clear();
+ if (lastProcessedEventTime != null) {
+ lastProcessedEventTimeState.add(lastProcessedEventTime);
+ }
- // clear and update schemaChangePending
- schemaChangePendingState.clear();
- schemaChangePendingState.add(schemaChangePending);
+ schemaChangePendingState.clear();
+ schemaChangePendingState.add(schemaChangePending);
- // clear and update local schema state
- localSchemaStateStore.clear();
- for (Map.Entry<TableIdentifier, CatalogTable> entry :
localSchemaState.entrySet()) {
- localSchemaStateStore.add(new SchemaStateEntry(entry.getKey(),
entry.getValue()));
- }
+ firstSeenCheckpointIdState.clear();
+ firstSeenCheckpointIdState.add(firstSeenCheckpointId);
- // batch sync buffered data to state only when dirty
- if (stateDirty) {
- bufferedDataRowsState.clear();
- synchronized (this) {
- for (Map.Entry<String, List<BufferedDataRow>> entry :
- bufferedDataRows.entrySet()) {
- bufferedDataRowsState.add(
- new BufferedDataEntry(entry.getKey(),
entry.getValue()));
- }
- stateDirty = false;
- }
- }
+ localSchemaStateStore.clear();
+ for (Map.Entry<TableIdentifier, CatalogTable> entry :
localSchemaState.entrySet()) {
+ localSchemaStateStore.add(new SchemaStateEntry(entry.getKey(),
entry.getValue()));
+ }
- log.debug(
- "SchemaOperator state snapshot completed using operator
state for checkpoint: {}, lastProcessedEventTime: {}, schemaChangePending: {},
localSchemaState size: {}, bufferedDataRows size: {}",
- context.getCheckpointId(),
- lastProcessedEventTime,
- schemaChangePending,
- localSchemaState.size(),
- bufferedDataRows.size());
- } catch (Exception e) {
- log.error("Error during state snapshot", e);
- throw e;
+ bufferedRecordsState.clear();
+ for (BufferedRecord record : pendingQueue) {
+ bufferedRecordsState.add(
+ new BufferedRecordEntry(
+ record.isSchemaChange,
+ record.row,
+ record.timestamp,
+ record.schemaEvent));
}
+
+ log.debug(
+ "State snapshot for checkpoint {}: lastEventTime={},
pending={}, "
+ + "firstSeenCkpt={}, queueSize={}",
+ context.getCheckpointId(),
+ lastProcessedEventTime,
+ schemaChangePending,
+ firstSeenCheckpointId,
+ pendingQueue.size());
}
@Override
public void initializeState(StateInitializationContext context) throws
Exception {
super.initializeState(context);
- if (this.bufferedDataRows == null) {
- this.bufferedDataRows = new ConcurrentHashMap<>();
+ if (this.pendingQueue == null) {
+ this.pendingQueue = new LinkedList<>();
}
- ListStateDescriptor<SchemaStateEntry> localSchemaStateDescriptor =
+ ListStateDescriptor<SchemaStateEntry> schemaDescriptor =
new ListStateDescriptor<>("localSchemaState",
SchemaStateEntry.class);
-
- ListStateDescriptor<Long> lastProcessedEventTimeDescriptor =
+ ListStateDescriptor<Long> eventTimeDescriptor =
new ListStateDescriptor<>("lastProcessedEventTime",
Long.class);
-
- ListStateDescriptor<Boolean> schemaChangePendingDescriptor =
+ ListStateDescriptor<Boolean> pendingDescriptor =
new ListStateDescriptor<>("schemaChangePending",
Boolean.class);
+ ListStateDescriptor<BufferedRecordEntry> bufferedDescriptor =
+ new ListStateDescriptor<>("bufferedRecords",
BufferedRecordEntry.class);
+ ListStateDescriptor<Long> firstSeenCkptDescriptor =
+ new ListStateDescriptor<>("firstSeenCheckpointId", Long.class);
- ListStateDescriptor<BufferedDataEntry> bufferedDataRowsDescriptor =
- new ListStateDescriptor<>("bufferedDataRows",
BufferedDataEntry.class);
-
- this.localSchemaStateStore =
-
context.getOperatorStateStore().getListState(localSchemaStateDescriptor);
+ this.localSchemaStateStore =
context.getOperatorStateStore().getListState(schemaDescriptor);
this.lastProcessedEventTimeState =
-
context.getOperatorStateStore().getListState(lastProcessedEventTimeDescriptor);
+
context.getOperatorStateStore().getListState(eventTimeDescriptor);
this.schemaChangePendingState =
-
context.getOperatorStateStore().getListState(schemaChangePendingDescriptor);
- this.bufferedDataRowsState =
-
context.getOperatorStateStore().getListState(bufferedDataRowsDescriptor);
+
context.getOperatorStateStore().getListState(pendingDescriptor);
+ this.bufferedRecordsState =
+
context.getOperatorStateStore().getListState(bufferedDescriptor);
+ this.firstSeenCheckpointIdState =
+
context.getOperatorStateStore().getListState(firstSeenCkptDescriptor);
if (context.isRestored()) {
- // restore from operator state
- Iterable<Long> eventTimes = lastProcessedEventTimeState.get();
- for (Long eventTime : eventTimes) {
- this.lastProcessedEventTime = eventTime;
+ for (Long t : lastProcessedEventTimeState.get()) {
+ this.lastProcessedEventTime = t;
break;
}
-
- Iterable<Boolean> pendingFlags = schemaChangePendingState.get();
- for (Boolean pending : pendingFlags) {
- this.schemaChangePending = pending;
+ for (Boolean p : schemaChangePendingState.get()) {
+ this.schemaChangePending = p;
break;
}
-
- // restore schema state
- Iterable<SchemaStateEntry> schemaEntries =
localSchemaStateStore.get();
- for (SchemaStateEntry entry : schemaEntries) {
+ for (Long ckpt : firstSeenCheckpointIdState.get()) {
+ this.firstSeenCheckpointId = ckpt;
+ break;
+ }
+ for (SchemaStateEntry entry : localSchemaStateStore.get()) {
localSchemaState.put(entry.tableId, entry.catalogTable);
- log.info("Restored schema state for table: {}", entry.tableId);
}
-
- // restore buffered data rows
- Iterable<BufferedDataEntry> bufferedEntries =
bufferedDataRowsState.get();
- if (bufferedEntries != null) {
- synchronized (this) {
- for (BufferedDataEntry entry : bufferedEntries) {
- if (entry != null && entry.key != null &&
entry.bufferedRows != null) {
- bufferedDataRows.put(entry.key, new
ArrayList<>(entry.bufferedRows));
- log.info(
- "Restored {} buffered data rows for key:
{}",
- entry.bufferedRows.size(),
- entry.key);
- }
- }
+ for (BufferedRecordEntry entry : bufferedRecordsState.get()) {
+ if (entry.isSchemaChange) {
+
pendingQueue.add(BufferedRecord.schemaChange(entry.schemaEvent));
+ } else {
+ pendingQueue.add(BufferedRecord.data(entry.row,
entry.timestamp));
}
}
+ log.info(
+ "State restored: lastEventTime={}, pending={},
firstSeenCkpt={}, queueSize={}",
+ lastProcessedEventTime,
+ schemaChangePending,
+ firstSeenCheckpointId,
+ pendingQueue.size());
}
+ }
- log.info(
- "SchemaOperator state initialized using operator state -
lastProcessedEventTime: {}, schemaChangePending: {}, localSchemaState size: {},
bufferedDataRows size: {}",
- this.lastProcessedEventTime,
- this.schemaChangePending,
- localSchemaState.size(),
- bufferedDataRows.size());
+ private boolean isSchemaEvolutionEnabled(Config config) {
+ return config.hasPath("schema-changes.enabled")
+ && config.getBoolean("schema-changes.enabled");
}
- @Setter
- @Getter
- public static class BufferedDataRow implements Serializable {
- private static final long serialVersionUID = 1L;
+ 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 SeaTunnelRow row;
- private long timestamp;
+ private void sendSchemaChangeEventToDownstream(SchemaChangeEvent
schemaChangeEvent) {
+ log.info(
+ "Broadcasting SchemaChangeEvent to downstream for table: {}",
+ schemaChangeEvent.tableIdentifier());
+ SeaTunnelRow broadcastRow = new SeaTunnelRow(0);
+ Map<String, Object> options = new HashMap<>();
+ options.put("schema_change_broadcast", schemaChangeEvent);
+ broadcastRow.setOptions(options);
+ output.collect(new StreamRecord<>(broadcastRow));
+ }
- public BufferedDataRow() {}
+ @Override
+ public void close() throws Exception {
+ super.close();
+ }
- public BufferedDataRow(SeaTunnelRow row, long timestamp) {
+ static class BufferedRecord {
+ final boolean isSchemaChange;
+ final SeaTunnelRow row;
+ final long timestamp;
+ final SchemaChangeEvent schemaEvent;
+
+ private BufferedRecord(
+ boolean isSchemaChange,
+ SeaTunnelRow row,
+ long timestamp,
+ SchemaChangeEvent schemaEvent) {
+ this.isSchemaChange = isSchemaChange;
this.row = row;
this.timestamp = timestamp;
+ this.schemaEvent = schemaEvent;
+ }
+
+ static BufferedRecord data(SeaTunnelRow row, long timestamp) {
+ return new BufferedRecord(false, row, timestamp, null);
+ }
+
+ static BufferedRecord schemaChange(SchemaChangeEvent event) {
+ return new BufferedRecord(true, null, 0L, event);
}
}
@@ -797,7 +509,6 @@ public class SchemaOperator extends
AbstractStreamOperator<SeaTunnelRow>
@Getter
public static class SchemaStateEntry implements Serializable {
private static final long serialVersionUID = 1L;
-
private TableIdentifier tableId;
private CatalogTable catalogTable;
@@ -811,17 +522,24 @@ public class SchemaOperator extends
AbstractStreamOperator<SeaTunnelRow>
@Setter
@Getter
- public static class BufferedDataEntry implements Serializable {
+ public static class BufferedRecordEntry implements Serializable {
private static final long serialVersionUID = 1L;
+ private boolean isSchemaChange;
+ private SeaTunnelRow row;
+ private long timestamp;
+ private SchemaChangeEvent schemaEvent;
- private String key;
- private List<BufferedDataRow> bufferedRows;
-
- public BufferedDataEntry() {}
+ public BufferedRecordEntry() {}
- public BufferedDataEntry(String key, List<BufferedDataRow>
bufferedRows) {
- this.key = key;
- this.bufferedRows = bufferedRows;
+ public BufferedRecordEntry(
+ boolean isSchemaChange,
+ SeaTunnelRow row,
+ long timestamp,
+ SchemaChangeEvent schemaEvent) {
+ this.isSchemaChange = isSchemaChange;
+ this.row = row;
+ this.timestamp = timestamp;
+ this.schemaEvent = schemaEvent;
}
}
}
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/coordinator/LocalSchemaCoordinator.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/coordinator/LocalSchemaCoordinator.java
index 41310992a1..8d99f433a3 100644
---
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/coordinator/LocalSchemaCoordinator.java
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/schema/coordinator/LocalSchemaCoordinator.java
@@ -59,7 +59,7 @@ public class LocalSchemaCoordinator {
private static final long CLEANUP_INTERVAL_MS = 60_000L;
private final String jobId;
private final long requestTtlMs;
- private volatile int sinkParallelism = 0;
+ private final Set<Integer> activeSinkSubtasks =
ConcurrentHashMap.newKeySet();
private final Map<String, TimestampedPendingRequest> pendingRequests =
new ConcurrentHashMap<>();
private final Map<String, Set<Integer>> receivedAcks = new
ConcurrentHashMap<>();
@@ -106,19 +106,61 @@ public class LocalSchemaCoordinator {
.get();
}
+ /**
+ * @deprecated Sink parallelism is now tracked dynamically via {@link
+ * #registerSinkStateProvider} and {@link #unregisterSinkSubtask}.
This method is kept for
+ * backward compatibility but is effectively a no-op.
+ */
+ @Deprecated
public void registerSinkParallelism(int parallelism) {
- this.sinkParallelism = parallelism;
log.info(
- "Registered sink parallelism: {} for schema change
coordination in jobId: {}",
+ "Registered sink parallelism hint: {} for jobId: {} (active
tracking used instead)",
parallelism,
jobId);
}
public void registerSinkStateProvider(int subtaskId, SinkStateProvider
provider) {
sinkStateProviders.put(subtaskId, provider);
+ activeSinkSubtasks.add(subtaskId);
log.info("Registered sink state provider for subtask {} in jobId: {}",
subtaskId, jobId);
}
+ public void unregisterSinkSubtask(int subtaskId) {
+ boolean removed = activeSinkSubtasks.remove(subtaskId);
+ sinkStateProviders.remove(subtaskId);
+ if (!removed) {
+ return;
+ }
+ int remaining = activeSinkSubtasks.size();
+ log.info(
+ "Sink subtask {} unregistered (closed). Active sink subtasks
remaining: {} in jobId: {}",
+ subtaskId,
+ remaining,
+ jobId);
+
+ for (Map.Entry<String, TimestampedPendingRequest> entry :
pendingRequests.entrySet()) {
+ String key = entry.getKey();
+ TimestampedPendingRequest request = entry.getValue();
+ Set<Integer> applied = receivedAcks.get(key);
+
+ int expectedActive = Math.max(remaining, 1);
+ if (applied != null && applied.size() >= expectedActive) {
+ if (request.appliedPhaseCompleteAtomic.compareAndSet(false,
true)) {
+ boolean allSuccess = request.allSuccess.get();
+ request.future.complete(allSuccess);
+ log.info(
+ "After subtask {} unregistered, all {} active
subtasks have applied "
+ + "schema change for table {} (epoch {}).
Completing request with result: {}",
+ subtaskId,
+ expectedActive,
+ request.tableId,
+ request.epoch,
+ allSuccess);
+ }
+ }
+ }
+ }
+
public SchemaProcessingStatus querySchemaProcessingStatus(TableIdentifier
tableId, long epoch) {
if (sinkStateProviders.isEmpty()) {
log.warn(
@@ -176,10 +218,10 @@ public class LocalSchemaCoordinator {
public boolean requestSchemaChange(TableIdentifier tableId, long epoch,
long timeoutMs)
throws InterruptedException, SchemaCoordinationException {
String key = tableId.toString() + "#" + epoch;
- int expectedAcks = sinkParallelism;
+ int expectedAcks = activeSinkSubtasks.size();
if (expectedAcks == 0) {
log.warn(
- "Sink parallelism not registered yet. Cannot coordinate
schema change for table {} (epoch {}). "
+ "No active sink subtasks. Cannot coordinate schema change
for table {} (epoch {}). "
+ "Assuming success to avoid deadlock.",
tableId,
epoch);
@@ -270,6 +312,8 @@ public class LocalSchemaCoordinator {
}
appliedSubtasks.add(subtaskId);
+ int currentExpected = Math.min(request.expectedAcks,
activeSinkSubtasks.size());
+ currentExpected = Math.max(currentExpected, 1);
log.info(
"Subtask {} applied schema change for table {} (epoch {}),
success: {}. {}/{} subtasks applied.",
subtaskId,
@@ -277,20 +321,19 @@ public class LocalSchemaCoordinator {
epoch,
success,
appliedSubtasks.size(),
- request.expectedAcks);
+ currentExpected);
if (!success) {
request.allSuccess.set(false);
}
- // if all subtasks have applied, complete the future
- if (appliedSubtasks.size() >= request.expectedAcks) {
+ if (appliedSubtasks.size() >= currentExpected) {
if (request.appliedPhaseCompleteAtomic.compareAndSet(false, 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,
+ "All {} active subtasks have applied schema change for
table {} (epoch {}). Completing request with result: {}",
+ currentExpected,
tableId,
epoch,
allSuccess);
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
index cd60a52614..f776520acd 100644
---
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriter.java
@@ -141,7 +141,6 @@ public class FlinkSinkWriter<InputT, CommT, WriterStateT>
"FlinkSinkWriter applying SchemaChangeEvent for table: {}",
schemaChangeEvent.tableIdentifier());
- sinkWriter.prepareCommit();
if (!(sinkWriter instanceof SupportSchemaEvolutionSinkWriter)) {
log.warn(
"Sink writer {} does not support schema evolution,
ignoring SchemaChangeEvent for table: {}",
diff --git
a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/schema/SchemaOperatorTest.java
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/schema/SchemaOperatorTest.java
new file mode 100644
index 0000000000..fd3b7d2126
--- /dev/null
+++
b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/schema/SchemaOperatorTest.java
@@ -0,0 +1,400 @@
+/*
+ * 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.ConfigFactory;
+
+import org.apache.seatunnel.api.source.SupportSchemaEvolution;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.schema.SchemaChangeType;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import
org.apache.seatunnel.api.table.schema.exception.SchemaCoordinationException;
+import
org.apache.seatunnel.api.table.schema.exception.SchemaEvolutionException;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.translation.flink.schema.coordinator.LocalSchemaCoordinator;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.BroadcastState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.OperatorStateStore;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.OutputTag;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class SchemaOperatorTest {
+
+ @Test
+ void testWaitRoundBeforeReleasingBufferedRecords() throws Exception {
+ LocalSchemaCoordinator coordinator =
Mockito.mock(LocalSchemaCoordinator.class);
+ Mockito.when(
+ coordinator.requestSchemaChange(
+ Mockito.any(), Mockito.anyLong(),
Mockito.anyLong()))
+ .thenReturn(true);
+
+ OperatorTestContext context = createOperator(false);
+ setField(context.operator, "coordinator", coordinator);
+
+ AlterTableAddColumnEvent event = createSchemaChangeEvent();
+ SeaTunnelRow row = createDataRow("row-after-schema");
+
+ context.operator.processElement(new
StreamRecord<>(createSchemaRow(event), 100L));
+ context.operator.processElement(new StreamRecord<>(row, 101L));
+
+ context.operator.notifyCheckpointComplete(10L);
+
+ assertTrue(context.output.records.isEmpty());
+ assertEquals(10L, getLongField(context.operator,
"firstSeenCheckpointId"));
+ assertEquals(2, getPendingQueue(context.operator).size());
+ Mockito.verifyNoInteractions(coordinator);
+
+ context.operator.notifyCheckpointComplete(11L);
+
+ assertEquals(2, context.output.records.size());
+ assertSchemaBroadcast(context.output.records.get(0), event);
+ assertEquals(row, context.output.records.get(1).getValue());
+ assertFalse(getBooleanField(context.operator, "schemaChangePending"));
+ assertEquals(-1L, getLongField(context.operator,
"firstSeenCheckpointId"));
+ assertTrue(getPendingQueue(context.operator).isEmpty());
+ Mockito.verify(coordinator)
+ .requestSchemaChange(event.tableIdentifier(),
event.getCreatedTime(), 300_000L);
+ }
+
+ @Test
+ void testSnapshotRestoreWithPendingSchemaChange() throws Exception {
+ OperatorStateStoreStub stateStore = new OperatorStateStoreStub();
+ OperatorTestContext originalContext = createOperator(stateStore,
false);
+
+ AlterTableAddColumnEvent event = createSchemaChangeEvent();
+ SeaTunnelRow row = createDataRow("buffered-across-restore");
+
+ originalContext.operator.processElement(new
StreamRecord<>(createSchemaRow(event), 200L));
+ originalContext.operator.processElement(new StreamRecord<>(row, 201L));
+ originalContext.operator.notifyCheckpointComplete(20L);
+ originalContext.operator.snapshotState(snapshotContext(20L));
+
+ LocalSchemaCoordinator restoredCoordinator =
Mockito.mock(LocalSchemaCoordinator.class);
+ Mockito.when(
+ restoredCoordinator.requestSchemaChange(
+ Mockito.any(), Mockito.anyLong(),
Mockito.anyLong()))
+ .thenReturn(true);
+
+ OperatorTestContext restoredContext = createOperator(stateStore, true);
+ setField(restoredContext.operator, "coordinator", restoredCoordinator);
+
+ assertEquals(20L, getLongField(restoredContext.operator,
"firstSeenCheckpointId"));
+ assertEquals(2, getPendingQueue(restoredContext.operator).size());
+ assertTrue(getBooleanField(restoredContext.operator,
"schemaChangePending"));
+
+ restoredContext.operator.notifyCheckpointComplete(21L);
+
+ assertEquals(2, restoredContext.output.records.size());
+ assertSchemaBroadcast(restoredContext.output.records.get(0), event);
+ assertEquals(row, restoredContext.output.records.get(1).getValue());
+ assertTrue(getPendingQueue(restoredContext.operator).isEmpty());
+ assertFalse(getBooleanField(restoredContext.operator,
"schemaChangePending"));
+ Mockito.verify(restoredCoordinator)
+ .requestSchemaChange(event.tableIdentifier(),
event.getCreatedTime(), 300_000L);
+ }
+
+ @Test
+ void testCoordinationFailureKeepsBufferedRecordsBlocked() throws Exception
{
+ LocalSchemaCoordinator coordinator =
Mockito.mock(LocalSchemaCoordinator.class);
+ AlterTableAddColumnEvent event = createSchemaChangeEvent();
+ Mockito.when(
+ coordinator.requestSchemaChange(
+ Mockito.any(), Mockito.anyLong(),
Mockito.anyLong()))
+ .thenThrow(
+ SchemaCoordinationException.timeout(
+ event.tableIdentifier(),
+ "job-under-test",
+ 1L,
+ new RuntimeException("timeout")));
+
+ OperatorTestContext context = createOperator(false);
+ setField(context.operator, "coordinator", coordinator);
+
+ SeaTunnelRow row = createDataRow("must-stay-buffered");
+ context.operator.processElement(new
StreamRecord<>(createSchemaRow(event), 300L));
+ context.operator.processElement(new StreamRecord<>(row, 301L));
+ context.operator.notifyCheckpointComplete(30L);
+
+ assertThrows(
+ SchemaEvolutionException.class,
+ () -> context.operator.notifyCheckpointComplete(31L));
+
+ assertEquals(1, context.output.records.size());
+ assertSchemaBroadcast(context.output.records.get(0), event);
+ assertTrue(getBooleanField(context.operator, "schemaChangePending"));
+ assertEquals(30L, getLongField(context.operator,
"firstSeenCheckpointId"));
+ Queue<SchemaOperator.BufferedRecord> pendingQueue =
getPendingQueue(context.operator);
+ assertEquals(2, pendingQueue.size());
+ assertTrue(pendingQueue.peek().isSchemaChange);
+ }
+
+ private static OperatorTestContext createOperator(boolean restored) throws
Exception {
+ return createOperator(new OperatorStateStoreStub(), restored);
+ }
+
+ private static OperatorTestContext createOperator(
+ OperatorStateStoreStub stateStore, boolean restored) throws
Exception {
+ SupportSchemaEvolution source =
Mockito.mock(SupportSchemaEvolution.class);
+ Mockito.when(source.supports())
+
.thenReturn(Collections.singletonList(SchemaChangeType.ADD_COLUMN));
+
+ SchemaOperator operator =
+ new SchemaOperator(
+ "bootstrap-job-id",
+ source,
+ ConfigFactory.parseString("schema-changes.enabled =
true"));
+
+ CollectingOutput output = new CollectingOutput();
+ setField(operator, AbstractStreamOperator.class, "output", output);
+ setField(operator, AbstractStreamOperator.class, "runtimeContext",
runtimeContext());
+ setField(
+ operator,
+ AbstractStreamOperator.class,
+ "stateHandler",
+ Mockito.mock(StreamOperatorStateHandler.class));
+
+ StateInitializationContext initializationContext =
+ Mockito.mock(StateInitializationContext.class);
+
Mockito.when(initializationContext.getOperatorStateStore()).thenReturn(stateStore);
+ Mockito.when(initializationContext.isRestored()).thenReturn(restored);
+
+ operator.initializeState(initializationContext);
+ operator.open();
+ return new OperatorTestContext(operator, output);
+ }
+
+ private static StreamingRuntimeContext runtimeContext() {
+ StreamingRuntimeContext runtimeContext =
Mockito.mock(StreamingRuntimeContext.class);
+ Mockito.when(runtimeContext.getJobId()).thenReturn(new JobID());
+ return runtimeContext;
+ }
+
+ private static StateSnapshotContext snapshotContext(long checkpointId)
throws Exception {
+ StateSnapshotContext snapshotContext =
Mockito.mock(StateSnapshotContext.class);
+
Mockito.when(snapshotContext.getCheckpointId()).thenReturn(checkpointId);
+ return snapshotContext;
+ }
+
+ private static AlterTableAddColumnEvent createSchemaChangeEvent() {
+ return AlterTableAddColumnEvent.add(
+ TableIdentifier.of("catalog", "database", "table"),
+ PhysicalColumn.of("added_col", BasicType.STRING_TYPE, 64L,
true, null, null));
+ }
+
+ private static SeaTunnelRow createSchemaRow(AlterTableAddColumnEvent
event) {
+ SeaTunnelRow row = new SeaTunnelRow(0);
+ row.setTableId("__SCHEMA_CHANGE_EVENT__");
+ Map<String, Object> options = new LinkedHashMap<>();
+ options.put("schema_change_event", event);
+ row.setOptions(options);
+ return row;
+ }
+
+ private static SeaTunnelRow createDataRow(String value) {
+ SeaTunnelRow row = new SeaTunnelRow(1);
+ row.setTableId("catalog.database.table");
+ row.setField(0, value);
+ return row;
+ }
+
+ private static void assertSchemaBroadcast(
+ StreamRecord<SeaTunnelRow> record, AlterTableAddColumnEvent event)
{
+ Object broadcastEvent =
record.getValue().getOptions().get("schema_change_broadcast");
+ assertInstanceOf(AlterTableAddColumnEvent.class, broadcastEvent);
+ assertEquals(event, broadcastEvent);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Queue<SchemaOperator.BufferedRecord>
getPendingQueue(SchemaOperator operator)
+ throws Exception {
+ return (Queue<SchemaOperator.BufferedRecord>) getField(operator,
"pendingQueue");
+ }
+
+ private static boolean getBooleanField(Object target, String fieldName)
throws Exception {
+ return (boolean) getField(target, fieldName);
+ }
+
+ private static long getLongField(Object target, String fieldName) throws
Exception {
+ return (long) getField(target, fieldName);
+ }
+
+ private static Object getField(Object target, String fieldName) throws
Exception {
+ Field field = findField(target.getClass(), fieldName);
+ field.setAccessible(true);
+ return field.get(target);
+ }
+
+ private static void setField(Object target, String fieldName, Object
value) throws Exception {
+ Field field = findField(target.getClass(), fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ private static void setField(Object target, Class<?> owner, String
fieldName, Object value)
+ throws Exception {
+ Field field = owner.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ private static Field findField(Class<?> type, String fieldName) throws
NoSuchFieldException {
+ Class<?> current = type;
+ while (current != null) {
+ try {
+ return current.getDeclaredField(fieldName);
+ } catch (NoSuchFieldException ignored) {
+ current = current.getSuperclass();
+ }
+ }
+ throw new NoSuchFieldException(fieldName);
+ }
+
+ private static final class OperatorTestContext {
+ private final SchemaOperator operator;
+ private final CollectingOutput output;
+
+ private OperatorTestContext(SchemaOperator operator, CollectingOutput
output) {
+ this.operator = operator;
+ this.output = output;
+ }
+ }
+
+ private static final class CollectingOutput implements
Output<StreamRecord<SeaTunnelRow>> {
+ private final List<StreamRecord<SeaTunnelRow>> records = new
ArrayList<>();
+
+ @Override
+ public void collect(StreamRecord<SeaTunnelRow> record) {
+ records.add(record);
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void emitWatermark(Watermark mark) {}
+
+ @Override
+ public void emitWatermarkStatus(WatermarkStatus watermarkStatus) {}
+
+ @Override
+ public <X> void collect(OutputTag<X> outputTag, StreamRecord<X>
record) {}
+
+ @Override
+ public void emitLatencyMarker(LatencyMarker latencyMarker) {}
+ }
+
+ private static final class OperatorStateStoreStub implements
OperatorStateStore {
+ private final Map<String, TestingListState<?>> listStates = new
LinkedHashMap<>();
+
+ @Override
+ public <K, V> BroadcastState<K, V> getBroadcastState(
+ MapStateDescriptor<K, V> stateDescriptor) {
+ throw new UnsupportedOperationException("Broadcast state is not
needed in this test");
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <S> ListState<S> getListState(ListStateDescriptor<S>
stateDescriptor) {
+ return (ListState<S>)
+ listStates.computeIfAbsent(
+ stateDescriptor.getName(), ignored -> new
TestingListState<>());
+ }
+
+ @Override
+ public <S> ListState<S> getUnionListState(ListStateDescriptor<S>
stateDescriptor) {
+ return getListState(stateDescriptor);
+ }
+
+ @Override
+ public Set<String> getRegisteredStateNames() {
+ return listStates.keySet();
+ }
+
+ @Override
+ public Set<String> getRegisteredBroadcastStateNames() {
+ return Collections.emptySet();
+ }
+ }
+
+ private static final class TestingListState<T> implements ListState<T> {
+ private final List<T> values = new ArrayList<>();
+
+ @Override
+ public Iterable<T> get() {
+ return new ArrayList<>(values);
+ }
+
+ @Override
+ public void add(T value) {
+ values.add(value);
+ }
+
+ @Override
+ public void update(List<T> values) {
+ this.values.clear();
+ if (values != null) {
+ this.values.addAll(values);
+ }
+ }
+
+ @Override
+ public void addAll(List<T> values) {
+ if (values != null) {
+ this.values.addAll(values);
+ }
+ }
+
+ @Override
+ public void clear() {
+ values.clear();
+ }
+ }
+}
