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 26d4ab9060 [Fix][Zeta] Ignore unknown table schema changes (#11328)
26d4ab9060 is described below
commit 26d4ab906039d5850d00189b508d53703d416fa0
Author: Jast <[email protected]>
AuthorDate: Tue Jul 7 20:46:57 2026 +0800
[Fix][Zeta] Ignore unknown table schema changes (#11328)
---
.../server/task/SeaTunnelSourceCollector.java | 11 ++-
.../SeaTunnelSourceCollectorSchemaChangeTest.java | 97 ++++++++++++++++++++++
2 files changed, 106 insertions(+), 2 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java
index 3112e993f6..ee566a5ad3 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java
@@ -265,9 +265,16 @@ public class SeaTunnelSourceCollector<T> implements
Collector<T> {
rowType = dataTypeChangeEventHandler.reset((SeaTunnelRowType)
rowType).apply(event);
} else if (rowType instanceof MultipleRowType) {
String tableId = event.tablePath().toString();
+ SeaTunnelRowType currentRowType = rowTypeMap.get(tableId);
+ if (currentRowType == null) {
+ log.warn(
+ "Ignore schema change event for unknown table {},
current table ids: {}",
+ tableId,
+ rowTypeMap.keySet());
+ return;
+ }
rowTypeMap.put(
- tableId,
-
dataTypeChangeEventHandler.reset(rowTypeMap.get(tableId)).apply(event));
+ tableId,
dataTypeChangeEventHandler.reset(currentRowType).apply(event));
} else {
throw new SeaTunnelEngineException(
"Unsupported row type: " +
rowType.getClass().getName());
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollectorSchemaChangeTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollectorSchemaChangeTest.java
new file mode 100644
index 0000000000..540e05a6bc
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollectorSchemaChangeTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.engine.server.task;
+
+import org.apache.seatunnel.api.common.metrics.Counter;
+import org.apache.seatunnel.api.common.metrics.Meter;
+import org.apache.seatunnel.api.common.metrics.MetricsContext;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.MultipleRowType;
+import org.apache.seatunnel.api.table.type.Record;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.core.starter.flowcontrol.FlowControlStrategy;
+import org.apache.seatunnel.engine.common.config.EngineConfig;
+import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
+import org.apache.seatunnel.engine.server.task.flow.OneInputFlowLifeCycle;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class SeaTunnelSourceCollectorSchemaChangeTest {
+
+ @Test
+ void shouldIgnoreSchemaChangeForUnknownTableInMultipleRowType() throws
IOException {
+ TablePath knownTable = TablePath.of("test_db", "known_table");
+ TablePath unknownTable = TablePath.of("test_db", "unknown_table");
+ SeaTunnelRowType knownRowType =
+ new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+ MultipleRowType multipleRowType =
+ new MultipleRowType(
+ new String[] {knownTable.toString()},
+ new SeaTunnelRowType[] {knownRowType});
+
+ OneInputFlowLifeCycle<Record<?>> output =
Mockito.mock(OneInputFlowLifeCycle.class);
+ SeaTunnelSourceCollector<Object> collector =
+ new SeaTunnelSourceCollector<>(
+ new Object(),
+ Collections.singletonList(output),
+ metricsContext(),
+ FlowControlStrategy.builder().build(),
+ multipleRowType,
+ Collections.singletonList(knownTable),
+ sourceTask(),
+ new EngineConfig());
+
+ collector.collect(
+ AlterTableChangeColumnEvent.change(
+ TableIdentifier.of("", unknownTable),
+ "old_name",
+ PhysicalColumn.of(
+ "new_name", BasicType.STRING_TYPE, (Long)
null, true, null, null)));
+
+ verify(output, never()).received(Mockito.any());
+ }
+
+ private static SeaTunnelTask sourceTask() {
+ SeaTunnelTask sourceTask = Mockito.mock(SeaTunnelTask.class);
+ when(sourceTask.getTaskLocation())
+ .thenReturn(new TaskLocation(new TaskGroupLocation(1L, 1, 1L),
1L, 0));
+ return sourceTask;
+ }
+
+ private static MetricsContext metricsContext() {
+ MetricsContext metricsContext = Mockito.mock(MetricsContext.class);
+
when(metricsContext.counter(Mockito.anyString())).thenReturn(Mockito.mock(Counter.class));
+
when(metricsContext.meter(Mockito.anyString())).thenReturn(Mockito.mock(Meter.class));
+ return metricsContext;
+ }
+}
