This is an automated email from the ASF dual-hosted git repository.
nzw921rx 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 fb8e7a7eea [Fix][Zeta] Restore legacy job status WAL entries (#11781)
fb8e7a7eea is described below
commit fb8e7a7eeaa8ece1a4fc169619a8cc7b2c3f69f2
Author: Jast <[email protected]>
AuthorDate: Wed Oct 21 22:13:21 2026 +0800
[Fix][Zeta] Restore legacy job status WAL entries (#11781)
Co-authored-by: zhangshenghang <[email protected]>
Co-authored-by: zhangshenghang <[email protected]>
Co-authored-by: zhangshenghang <[email protected]>
Co-authored-by: jast <[email protected]>
Co-authored-by: David Zollo <[email protected]>
Co-authored-by: Claude Sonnet 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>
Co-authored-by: David Zollo <[email protected]>
---
.../introduction/concepts/incompatible-changes.md | 5 +
.../introduction/concepts/incompatible-changes.md | 5 +
.../imap-storage-plugins/imap-storage-file/pom.xml | 9 ++
.../engine/imap/storage/file/common/WALReader.java | 60 +++++++-
.../file/common/WALReaderAndWriterTest.java | 163 +++++++++++++++++++++
5 files changed, 238 insertions(+), 4 deletions(-)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index eff49771be..15453b9ae6 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -573,6 +573,11 @@ or state format migration is required.
### Engine Behavior Changes
+- **File-based IMap storage: legacy job model class names in WAL files are
remapped on startup**
+ - **Affected component**:
`seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file`
+ - **Description**: SeaTunnel 3.0 relocated the job status model classes
(`JobStatus`, `JobResult`, `JobStatusData`) from
`org.apache.seatunnel.engine.core.job.*` to
`org.apache.seatunnel.engine.common.job.*` (#9689). WAL files written by
releases before that relocation still record the old fully-qualified class
names, so replaying them on a current distribution fails with
`ClassNotFoundException`. The file-based IMap storage now transparently
rewrites these three legacy names to their [...]
+ - **Impact**: Clusters using file-based IMap persistence can upgrade and
restart across the #9689 relocation without manual WAL surgery. Writing is
unaffected: new WAL entries always record the current class names. Note that
the rewrite is one-directional: WAL files written by a current distribution
contain class names that older releases cannot resolve, so downgrading a
cluster after it has written new WAL entries requires deleting or manually
rewriting the persisted WAL data. (#11781)
+
- **Behavior change: the REST log-content endpoints return at most 64 MB by
default**
- **Affected component**: `seatunnel-engine-server`, REST v2 endpoints `GET
/logs/:file` and
`GET /log/:file` and their REST v1 equivalents `GET
/hazelcast/rest/maps/logs/:file` and
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index bb0974bc39..ff9e054595 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -475,6 +475,11 @@ Zeta SQL ARRAY 函数现在将 TINYINT 元素声明为 `ARRAY<TINYINT>`，与实
### 引擎行为变更
+- **文件型 IMap 存储：启动时自动重映射 WAL 文件中的旧版作业模型类名**
+ -
**受影响组件**：`seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file`
+ - **描述**：SeaTunnel 3.0 将作业状态模型类（`JobStatus`、`JobResult`、`JobStatusData`）从
`org.apache.seatunnel.engine.core.job.*` 迁移到了
`org.apache.seatunnel.engine.common.job.*`（#9689）。迁移之前版本写入的 WAL
文件仍记录旧的完整类名，在当前发行版上回放会抛出 `ClassNotFoundException`。文件型 IMap 存储现在在读取 WAL
条目时会自动将这三个旧类名重写到当前位置，无需任何配置变更。
+ - **影响**：使用文件型 IMap 持久化的集群可以跨 #9689 迁移直接升级重启，无需手工修改 WAL 数据。写入不受影响：新的 WAL
条目始终记录当前类名。注意该重写是单向的：当前发行版写出的 WAL 文件中的类名无法被旧版本解析，因此在写出新 WAL
条目后回滚降级集群时，需要删除或手工改写已持久化的 WAL 数据。(#11781)
+
- **行为变更：REST 日志内容接口默认最多返回 64 MB**
- **受影响组件**：`seatunnel-engine-server`，REST v2 接口 `GET /logs/:file`、`GET
/log/:file`，
以及对应的 REST v1 接口 `GET /hazelcast/rest/maps/logs/:file`、`GET
/hazelcast/rest/maps/log/:file`。
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml
index 96069b75c0..cb7bf09e26 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml
@@ -43,6 +43,15 @@
<artifactId>serializer-protobuf</artifactId>
<version>${project.version}</version>
</dependency>
+ <!-- Test-scoped only: production code intentionally references the
pre-relocation
+ job model class names as strings; the tests bind those strings to
the real
+ engine-common classes so a future rename fails the build. -->
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-engine-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- hadoop jar -->
<dependency>
<groupId>org.apache.seatunnel</groupId>
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
index f026e2f940..f69d0cfb38 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
@@ -33,6 +33,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -42,6 +43,38 @@ import java.util.Set;
import static
org.apache.seatunnel.engine.imap.storage.file.common.LatestMutationAccumulator.SerializedKey;
public class WALReader {
+ /**
+ * Maps the fully-qualified class names recorded in WAL entries written by
SeaTunnel versions
+ * older than #9689 (which moved the job status model from {@code
seatunnel-engine-core} to
+ * {@code seatunnel-engine-common}) to their current class names.
+ *
+ * <p>Without this mapping, restarting a cluster that persists job state
through the file-based
+ * IMap storage fails with {@code ClassNotFoundException} while replaying
existing WAL entries,
+ * see #9928. Removing an entry from this table breaks upgrades from any
release that still
+ * wrote the old class name, so keep it in place and extend it whenever a
persisted job model
+ * class is moved again.
+ *
+ * <p>Resolution is single-level: each entry rewrites one legacy name to
its current location.
+ * If a target class is relocated again later, add an entry for every
intermediate name so WAL
+ * files written between relocations still resolve; do not rely on the map
being followed
+ * transitively.
+ */
+ private static final Map<String, String> LEGACY_CLASS_NAME_MAPPINGS;
+
+ static {
+ Map<String, String> legacyClassNameMappings = new HashMap<>();
+ legacyClassNameMappings.put(
+ "org.apache.seatunnel.engine.core.job.JobResult",
+ "org.apache.seatunnel.engine.common.job.JobResult");
+ legacyClassNameMappings.put(
+ "org.apache.seatunnel.engine.core.job.JobStatus",
+ "org.apache.seatunnel.engine.common.job.JobStatus");
+ legacyClassNameMappings.put(
+ "org.apache.seatunnel.engine.core.job.JobStatusData",
+ "org.apache.seatunnel.engine.common.job.JobStatusData");
+ LEGACY_CLASS_NAME_MAPPINGS =
Collections.unmodifiableMap(legacyClassNameMappings);
+ }
+
private final Serializer serializer;
private final IFileReader<IMapFileData> fileReader;
@@ -108,19 +141,38 @@ public class WALReader {
private Object deserializeData(byte[] data, String className) {
try {
- Class<?> clazz = ClassUtils.getClass(className);
+ Class<?> clazz = ClassUtils.getClass(resolveClassName(className));
try {
return serializer.deserialize(data, clazz);
} catch (IOException e) {
// log.error("deserialize data error, data is {}, className is
{}", data, className,
// e);
+ // IMapStorageException formats the message with
String.format, so use %s
+ // placeholders instead of {} to keep the data and class name
visible.
throw new IMapStorageException(
- e, "deserialize data error: data is s%, className is
s%", data, className);
+ e, "deserialize data error: data is %s, className is
%s", data, className);
}
} catch (ClassNotFoundException e) {
- // log.error("deserialize data error, class name is {}",
className, e);
+ // log.error("deserialize data error, class name is {}, resolved
class name is {}",
+ // className, resolveClassName(className), e);
+ // IMapStorageException formats the message with String.format, so
use %s
+ // placeholders instead of {} to keep both class names visible.
throw new IMapStorageException(
- e, "deserialize data error, class name is {}", className);
+ e,
+ "deserialize data error, class name is %s, resolved class
name is %s",
+ className,
+ resolveClassName(className));
}
}
+
+ /**
+ * Resolves class names written by SeaTunnel versions before the job model
moved to
+ * engine-common.
+ *
+ * @param className class name recorded in the WAL
+ * @return class name available in the current distribution
+ */
+ private static String resolveClassName(String className) {
+ return LEGACY_CLASS_NAME_MAPPINGS.getOrDefault(className, className);
+ }
}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
index 6b1fb4b7ff..2eeb243669 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
@@ -20,6 +20,10 @@
package org.apache.seatunnel.engine.imap.storage.file.common;
+import org.apache.seatunnel.engine.common.job.JobResult;
+import org.apache.seatunnel.engine.common.job.JobStatus;
+import org.apache.seatunnel.engine.common.job.JobStatusData;
+import
org.apache.seatunnel.engine.imap.storage.api.exception.IMapStorageException;
import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration;
import org.apache.seatunnel.engine.serializer.api.Serializer;
@@ -55,6 +59,15 @@ public class WALReaderAndWriterTest {
private static final Path PARENT_PATH = new Path("/tmp/9/");
private static final Path SAME_TIMESTAMP_TOMBSTONE_PATH =
new Path("/tmp/imap-wal-same-timestamp-tombstone/");
+ // A real pre-#9689 WAL file is not required to exercise the legacy path:
the
+ // relocated types (JobStatus, JobResult, JobStatusData) kept identical
field
+ // layout and enum ordinals across the #9689 move, and the static
serialVersionUID
+ // added later is skipped by ProtoStuff's RuntimeSchema, so the only wire
difference
+ // is the valueClassName string, which these fixtures set to the old names
explicitly.
+ // Replaying a captured pre-relocation WAL frame remains a welcome
follow-up.
+ private static final Path LEGACY_JOB_STATUS_PATH = new
Path("/tmp/imap-wal-legacy-job-status/");
+ private static final Path UNKNOWN_CLASS_NAME_PATH =
+ new Path("/tmp/imap-wal-unknown-class-name/");
private static final Serializer SERIALIZER = new ProtoStuffSerializer();
@BeforeAll
@@ -240,10 +253,160 @@ public class WALReaderAndWriterTest {
Assertions.assertFalse(reader.loadAllKeys(SAME_TIMESTAMP_TOMBSTONE_PATH).contains(key));
}
+ /**
+ * Binds every legacy mapping target to the live engine-common class at
compile time: if a
+ * target class is renamed again, this test fails to compile or fails the
assertion instead of
+ * leaving a stale shim behind.
+ */
+ @Test
+ public void testLegacyMappingsTargetCurrentClassNames() throws Exception {
+ java.lang.reflect.Field field =
+ WALReader.class.getDeclaredField("LEGACY_CLASS_NAME_MAPPINGS");
+ field.setAccessible(true);
+ java.util.Map<String, String> mappings = (java.util.Map<String,
String>) field.get(null);
+
+ Assertions.assertEquals(
+ JobStatus.class.getName(),
+
mappings.get("org.apache.seatunnel.engine.core.job.JobStatus"));
+ Assertions.assertEquals(
+ JobResult.class.getName(),
+
mappings.get("org.apache.seatunnel.engine.core.job.JobResult"));
+ Assertions.assertEquals(
+ JobStatusData.class.getName(),
+
mappings.get("org.apache.seatunnel.engine.core.job.JobStatusData"));
+ }
+
+ @Test
+ public void testReaderLoadsJobStatusWrittenWithLegacyClassName() throws
Exception {
+ String key = "job-status";
+ try (WALWriter writer =
+ new WALWriter(FS, FileConfiguration.HDFS,
LEGACY_JOB_STATUS_PATH, SERIALIZER)) {
+ writer.write(
+ IMapFileData.builder()
+ .key(SERIALIZER.serialize(key))
+ .keyClassName(String.class.getName())
+ .value(SERIALIZER.serialize(JobStatus.RUNNING))
+
.valueClassName("org.apache.seatunnel.engine.core.job.JobStatus")
+ .deleted(false)
+ .timestamp(System.nanoTime())
+ .build());
+ }
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+ Map<Object, Object> data = reader.loadAllData(LEGACY_JOB_STATUS_PATH,
new HashSet<>());
+
+ Assertions.assertEquals(JobStatus.class, data.get(key).getClass());
+ // ProtoStuff returns a non-canonical enum instance, so compare by
value, not identity.
+ Assertions.assertEquals(JobStatus.RUNNING.name(), ((JobStatus)
data.get(key)).name());
+ }
+
+ @Test
+ public void testReaderLoadsAllLegacyJobModelClassNames() throws Exception {
+ String jobStatusKey = "legacy-job-status";
+ String jobResultKey = "legacy-job-result";
+ String jobStatusDataKey = "legacy-job-status-data";
+ JobStatusData jobStatusData =
+ new JobStatusData(1L, "legacy-job", JobStatus.FINISHED, 1L,
2L, 3L);
+ JobResult jobResult = new JobResult(JobStatus.CANCELED, "legacy
error");
+ try (WALWriter writer =
+ new WALWriter(FS, FileConfiguration.HDFS,
LEGACY_JOB_STATUS_PATH, SERIALIZER)) {
+ writer.write(
+ IMapFileData.builder()
+ .key(SERIALIZER.serialize(jobStatusKey))
+ .keyClassName(String.class.getName())
+ .value(SERIALIZER.serialize(JobStatus.RUNNING))
+
.valueClassName("org.apache.seatunnel.engine.core.job.JobStatus")
+ .deleted(false)
+ .timestamp(System.nanoTime())
+ .build());
+ writer.write(
+ IMapFileData.builder()
+ .key(SERIALIZER.serialize(jobResultKey))
+ .keyClassName(String.class.getName())
+ .value(SERIALIZER.serialize(jobResult))
+
.valueClassName("org.apache.seatunnel.engine.core.job.JobResult")
+ .deleted(false)
+ .timestamp(System.nanoTime())
+ .build());
+ writer.write(
+ IMapFileData.builder()
+ .key(SERIALIZER.serialize(jobStatusDataKey))
+ .keyClassName(String.class.getName())
+ .value(SERIALIZER.serialize(jobStatusData))
+
.valueClassName("org.apache.seatunnel.engine.core.job.JobStatusData")
+ .deleted(false)
+ .timestamp(System.nanoTime())
+ .build());
+ }
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+ Map<Object, Object> data = reader.loadAllData(LEGACY_JOB_STATUS_PATH,
new HashSet<>());
+
+ Assertions.assertEquals(
+ JobStatus.RUNNING.name(), ((JobStatus)
data.get(jobStatusKey)).name());
+
+ JobResult replayedResult = (JobResult) data.get(jobResultKey);
+ Assertions.assertEquals(JobResult.class, replayedResult.getClass());
+ Assertions.assertEquals(JobStatus.CANCELED.name(),
replayedResult.getStatus().name());
+ Assertions.assertEquals("legacy error", replayedResult.getError());
+
+ JobStatusData replayedStatusData = (JobStatusData)
data.get(jobStatusDataKey);
+ Assertions.assertEquals(JobStatusData.class,
replayedStatusData.getClass());
+ Assertions.assertEquals(jobStatusData.getJobId(),
replayedStatusData.getJobId());
+ Assertions.assertEquals(jobStatusData.getJobName(),
replayedStatusData.getJobName());
+ Assertions.assertEquals(
+ jobStatusData.getJobStatus().name(),
replayedStatusData.getJobStatus().name());
+ Assertions.assertEquals(jobStatusData.getSubmitTime(),
replayedStatusData.getSubmitTime());
+ Assertions.assertEquals(jobStatusData.getStartTime(),
replayedStatusData.getStartTime());
+ Assertions.assertEquals(jobStatusData.getFinishTime(),
replayedStatusData.getFinishTime());
+ }
+
+ /**
+ * Feeds a record whose class name is neither a current class nor a legacy
mapping key: the
+ * failure must surface the recorded and the resolved class name so a
stale or wrong mapping is
+ * diagnosable from the restart error alone. A name absent from the legacy
table resolves to
+ * itself, so both rendered names equal the recorded one.
+ */
+ @Test
+ public void testUnknownClassNameSurfacesInExceptionMessage() throws
Exception {
+ String unknownClassName =
"org.apache.seatunnel.engine.core.job.NeverExistedJobStatus";
+ try (WALWriter writer =
+ new WALWriter(FS, FileConfiguration.HDFS,
UNKNOWN_CLASS_NAME_PATH, SERIALIZER)) {
+ writer.write(
+ IMapFileData.builder()
+ .key(SERIALIZER.serialize("unknown-class"))
+ .keyClassName(String.class.getName())
+ .value(SERIALIZER.serialize("value"))
+ .valueClassName(unknownClassName)
+ .deleted(false)
+ .timestamp(System.nanoTime())
+ .build());
+ }
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+ IMapStorageException exception =
+ Assertions.assertThrows(
+ IMapStorageException.class,
+ () -> reader.loadAllData(UNKNOWN_CLASS_NAME_PATH, new
HashSet<>()));
+
+ Assertions.assertTrue(
+ exception
+ .getMessage()
+ .contains(
+ "class name is "
+ + unknownClassName
+ + ", resolved class name is "
+ + unknownClassName),
+ "exception message should render the recorded and resolved
class names, got: "
+ + exception.getMessage());
+ }
+
@AfterAll
public static void close() throws IOException {
FS.delete(PARENT_PATH, true);
FS.delete(SAME_TIMESTAMP_TOMBSTONE_PATH, true);
+ FS.delete(LEGACY_JOB_STATUS_PATH, true);
+ FS.delete(UNKNOWN_CLASS_NAME_PATH, true);
FS.close();
}
}
