This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 d11453ef95 [Fix][Flink] Process same-timestamp schema restore events
(#12306)
d11453ef95 is described below
commit d11453ef95b106cf80de361ee3a343bda9937f7e
Author: zhiweiniu <[email protected]>
AuthorDate: Tue Sep 22 23:02:24 2026 +0800
[Fix][Flink] Process same-timestamp schema restore events (#12306)
---
.../translation/flink/schema/SchemaOperator.java | 7 ++-
.../flink/schema/SchemaOperatorTest.java | 65 ++++++++++++++++++----
2 files changed, 60 insertions(+), 12 deletions(-)
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 1e8b46c37d..8cccdad7fb 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
@@ -374,7 +374,12 @@ public class SchemaOperator extends
AbstractStreamOperator<SeaTunnelRow>
TableIdentifier tableId = event.tableIdentifier();
long eventTime = event.getCreatedTime();
- if (lastProcessedEventTime != null && eventTime <=
lastProcessedEventTime) {
+ // Restore events only refresh runtime schema state and are safe to
apply repeatedly. They
+ // can also share the same millisecond timestamp when multiple tables
are restored in one
+ // batch, so applying the normal timestamp deduplication would
silently skip later tables.
+ if (!(event instanceof RestoreTableSchemaEvent)
+ && lastProcessedEventTime != null
+ && eventTime <= lastProcessedEventTime) {
log.warn(
"Skipping outdated schema change event (epoch {} <= last
processed {})",
eventTime,
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
index 56e2698a24..a19b9cd9a5 100644
---
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
@@ -75,17 +75,7 @@ public class SchemaOperatorTest {
@Test
void testRestoreSchemaEventIsAlwaysSupported() throws Exception {
- CatalogTable restoredTable =
- CatalogTable.of(
- TableIdentifier.of("catalog", "database", "table"),
- TableSchema.builder()
- .column(
- PhysicalColumn.of(
- "id", BasicType.LONG_TYPE,
20L, false, null, null))
- .build(),
- Collections.emptyMap(),
- Collections.emptyList(),
- null);
+ CatalogTable restoredTable = createRestoredTable("table");
OperatorTestContext context =
createOperator(new OperatorStateStoreStub(), false,
Collections.emptyList());
@@ -97,6 +87,46 @@ public class SchemaOperatorTest {
assertEquals(1, getPendingQueue(context.operator).size());
}
+ @Test
+ void testRestoreSchemaEventsWithSameTimestampAreBothProcessed() throws
Exception {
+ LocalSchemaCoordinator coordinator =
Mockito.mock(LocalSchemaCoordinator.class);
+ Mockito.when(
+ coordinator.requestSchemaChange(
+ Mockito.any(), Mockito.anyLong(),
Mockito.anyLong()))
+ .thenReturn(true);
+
+ OperatorTestContext context =
+ createOperator(new OperatorStateStoreStub(), false,
Collections.emptyList());
+ setField(context.operator, "coordinator", coordinator);
+
+ RestoreTableSchemaEvent firstEvent =
+ new
RestoreTableSchemaEvent(createRestoredTable("first_table"));
+ RestoreTableSchemaEvent secondEvent =
+ new
RestoreTableSchemaEvent(createRestoredTable("second_table"));
+ long sharedTimestamp = firstEvent.getCreatedTime();
+ setField(secondEvent, "createdTime", sharedTimestamp);
+
+ context.operator.processElement(
+ new StreamRecord<>(createSchemaRow(firstEvent),
sharedTimestamp));
+ context.operator.processElement(
+ new StreamRecord<>(createSchemaRow(secondEvent),
sharedTimestamp));
+
+ context.operator.notifyCheckpointComplete(10L);
+ context.operator.notifyCheckpointComplete(11L);
+ context.operator.notifyCheckpointComplete(12L);
+ context.operator.notifyCheckpointComplete(13L);
+
+ assertEquals(2, context.output.records.size());
+ assertSchemaBroadcast(context.output.records.get(0), firstEvent);
+ assertSchemaBroadcast(context.output.records.get(1), secondEvent);
+ assertTrue(getPendingQueue(context.operator).isEmpty());
+ assertFalse(getBooleanField(context.operator, "schemaChangePending"));
+ Mockito.verify(coordinator)
+ .requestSchemaChange(firstEvent.tableIdentifier(),
sharedTimestamp, 300_000L);
+ Mockito.verify(coordinator)
+ .requestSchemaChange(secondEvent.tableIdentifier(),
sharedTimestamp, 300_000L);
+ }
+
@Test
void testWaitRoundBeforeReleasingBufferedRecords() throws Exception {
LocalSchemaCoordinator coordinator =
Mockito.mock(LocalSchemaCoordinator.class);
@@ -363,6 +393,19 @@ public class SchemaOperatorTest {
PhysicalColumn.of("added_col", BasicType.STRING_TYPE, 64L,
true, null, null));
}
+ private static CatalogTable createRestoredTable(String tableName) {
+ return CatalogTable.of(
+ TableIdentifier.of("catalog", "database", tableName),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.LONG_TYPE, 20L, false,
null, null))
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null);
+ }
+
private static SeaTunnelRow createSchemaRow(SchemaChangeEvent event) {
SeaTunnelRow row = new SeaTunnelRow(0);
row.setTableId("__SCHEMA_CHANGE_EVENT__");
