This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 9cd5d0368a [Improve][Zeta] Use fixed serialVersionUID in task state
messages (#12466)
9cd5d0368a is described below
commit 9cd5d0368a4c6ccb72ea040edf315bcc1328e481
Author: Goutam Adwant <[email protected]>
AuthorDate: Tue Nov 3 06:23:53 2026 +0000
[Improve][Zeta] Use fixed serialVersionUID in task state messages (#12466)
Signed-off-by: Goutam Adwant <[email protected]>
---
.../engine/server/execution/TaskDeployState.java | 6 +
.../server/execution/TaskExecutionState.java | 6 +
.../serializable/TaskStateSerializationTest.java | 178 +++++++++++++++++++++
3 files changed, 190 insertions(+)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskDeployState.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskDeployState.java
index d683d6d86c..063c253df0 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskDeployState.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskDeployState.java
@@ -25,6 +25,12 @@ import java.io.Serializable;
@Data
public class TaskDeployState implements Serializable {
+ // Same as the previously computed value. Keep this UID for compatible
field additions; new
+ // reference fields deserialize as null when absent from old instances;
consumers must handle
+ // that null explicitly. Constructors do not supply defaults during
deserialization.
+ // Incompatible state changes need an explicit versioned serialization
contract.
+ private static final long serialVersionUID = 2646079648150626562L;
+
private final boolean success;
private final String throwableMsg;
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionState.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionState.java
index a2a5e747e5..04c98954e2 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionState.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionState.java
@@ -23,6 +23,12 @@ import java.io.Serializable;
public class TaskExecutionState implements Serializable {
+ // Same as the previously computed value. Keep this UID for compatible
field additions; new
+ // reference fields deserialize as null when absent from old instances;
consumers must handle
+ // that null explicitly. Constructors do not supply defaults during
deserialization.
+ // Incompatible state changes need an explicit versioned serialization
contract.
+ private static final long serialVersionUID = -108652017022658969L;
+
private final TaskGroupLocation taskGroupLocation;
private final ExecutionState executionState;
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/serializable/TaskStateSerializationTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/serializable/TaskStateSerializationTest.java
new file mode 100644
index 0000000000..1251553e62
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/serializable/TaskStateSerializationTest.java
@@ -0,0 +1,178 @@
+/*
+ * 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.serializable;
+
+import org.apache.seatunnel.engine.server.execution.ExecutionState;
+import org.apache.seatunnel.engine.server.execution.TaskDeployState;
+import org.apache.seatunnel.engine.server.execution.TaskExecutionState;
+import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.internal.serialization.Data;
+import com.hazelcast.internal.serialization.InternalSerializationService;
+import
com.hazelcast.internal.serialization.impl.DefaultSerializationServiceBuilder;
+import
com.hazelcast.internal.serialization.impl.defaultserializers.JavaDefaultSerializers.JavaSerializer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+import java.util.Base64;
+
+/**
+ * Verifies that task state messages exchanged between workers and the master
keep the Java
+ * serialization form they had before their serialVersionUID was declared.
+ *
+ * <p>The fixtures were serialized from the unmodified pre-UID classes using
JDK 8 and JDK 11; both
+ * produced identical bytes. The UID and ReadsPreviousWireForm tests guard
old-writer/new-reader
+ * compatibility; do not regenerate their fixtures when adding fields. This
Java serialization proxy
+ * applies while these classes remain plain Serializable without a custom
Hazelcast serializer. The
+ * Hazelcast round-trip test checks that fallback and current field values,
not exact archived
+ * bytes.
+ */
+class TaskStateSerializationTest {
+
+ private static final String TASK_EXECUTION_STATE_FAILED =
+
"rO0ABXNyAD9vcmcuYXBhY2hlLnNlYXR1bm5lbC5lbmdpbmUuc2VydmVyLmV4ZWN1dGlvbi5UYXNr"
+ +
"RXhlY3V0aW9uU3RhdGX+ff2Q8TwWZwIAA0wADmV4ZWN1dGlvblN0YXRldAA9TG9yZy9hcGFjaGUv"
+ +
"c2VhdHVubmVsL2VuZ2luZS9zZXJ2ZXIvZXhlY3V0aW9uL0V4ZWN1dGlvblN0YXRlO0wAEXRhc2tH"
+ +
"cm91cExvY2F0aW9udABATG9yZy9hcGFjaGUvc2VhdHVubmVsL2VuZ2luZS9zZXJ2ZXIvZXhlY3V0"
+ +
"aW9uL1Rhc2tHcm91cExvY2F0aW9uO0wADHRocm93YWJsZU1zZ3QAEkxqYXZhL2xhbmcvU3RyaW5n"
+ +
"O3hwfnIAO29yZy5hcGFjaGUuc2VhdHVubmVsLmVuZ2luZS5zZXJ2ZXIuZXhlY3V0aW9uLkV4ZWN1"
+ +
"dGlvblN0YXRlAAAAAAAAAAASAAB4cgAOamF2YS5sYW5nLkVudW0AAAAAAAAAABIAAHhwdAAGRkFJ"
+ +
"TEVEc3IAPm9yZy5hcGFjaGUuc2VhdHVubmVsLmVuZ2luZS5zZXJ2ZXIuZXhlY3V0aW9uLlRhc2tH"
+ +
"cm91cExvY2F0aW9ujIP/i/I+4/kCAANKAAVqb2JJZEkACnBpcGVsaW5lSWRKAAt0YXNrR3JvdXBJ"
+ +
"ZHhwAAAAAAAAA+kAAAABAAAAAAAATiF0AC5qYXZhLmxhbmcuSWxsZWdhbFN0YXRlRXhjZXB0aW9u"
+ + "OiBzb3VyY2UgZmFpbGVk";
+
+ private static final String TASK_EXECUTION_STATE_FINISHED =
+
"rO0ABXNyAD9vcmcuYXBhY2hlLnNlYXR1bm5lbC5lbmdpbmUuc2VydmVyLmV4ZWN1dGlvbi5UYXNr"
+ +
"RXhlY3V0aW9uU3RhdGX+ff2Q8TwWZwIAA0wADmV4ZWN1dGlvblN0YXRldAA9TG9yZy9hcGFjaGUv"
+ +
"c2VhdHVubmVsL2VuZ2luZS9zZXJ2ZXIvZXhlY3V0aW9uL0V4ZWN1dGlvblN0YXRlO0wAEXRhc2tH"
+ +
"cm91cExvY2F0aW9udABATG9yZy9hcGFjaGUvc2VhdHVubmVsL2VuZ2luZS9zZXJ2ZXIvZXhlY3V0"
+ +
"aW9uL1Rhc2tHcm91cExvY2F0aW9uO0wADHRocm93YWJsZU1zZ3QAEkxqYXZhL2xhbmcvU3RyaW5n"
+ +
"O3hwfnIAO29yZy5hcGFjaGUuc2VhdHVubmVsLmVuZ2luZS5zZXJ2ZXIuZXhlY3V0aW9uLkV4ZWN1"
+ +
"dGlvblN0YXRlAAAAAAAAAAASAAB4cgAOamF2YS5sYW5nLkVudW0AAAAAAAAAABIAAHhwdAAIRklO"
+ +
"SVNIRURzcgA+b3JnLmFwYWNoZS5zZWF0dW5uZWwuZW5naW5lLnNlcnZlci5leGVjdXRpb24uVGFz"
+ +
"a0dyb3VwTG9jYXRpb26Mg/+L8j7j+QIAA0oABWpvYklkSQAKcGlwZWxpbmVJZEoAC3Rhc2tHcm91"
+ + "cElkeHAAAAAAAAAD6QAAAAEAAAAAAABOIXA=";
+
+ private static final String TASK_DEPLOY_STATE_SUCCESS =
+
"rO0ABXNyADxvcmcuYXBhY2hlLnNlYXR1bm5lbC5lbmdpbmUuc2VydmVyLmV4ZWN1dGlvbi5UYXNr"
+ +
"RGVwbG95U3RhdGUkuMNro2ihAgIAAloAB3N1Y2Nlc3NMAAx0aHJvd2FibGVNc2d0ABJMamF2YS9s"
+ + "YW5nL1N0cmluZzt4cAFw";
+
+ private static final String TASK_DEPLOY_STATE_FAILED =
+
"rO0ABXNyADxvcmcuYXBhY2hlLnNlYXR1bm5lbC5lbmdpbmUuc2VydmVyLmV4ZWN1dGlvbi5UYXNr"
+ +
"RGVwbG95U3RhdGUkuMNro2ihAgIAAloAB3N1Y2Nlc3NMAAx0aHJvd2FibGVNc2d0ABJMamF2YS9s"
+ +
"YW5nL1N0cmluZzt4cAB0AC5qYXZhLmxhbmcuSWxsZWdhbFN0YXRlRXhjZXB0aW9uOiBkZXBsb3kg"
+ + "ZmFpbGVk";
+
+ private static final TaskGroupLocation LOCATION = new
TaskGroupLocation(1001L, 1, 20001L);
+
+ @Test
+ void testDeclaredSerialVersionUidsMatchPreviouslyGeneratedValues() {
+ Assertions.assertEquals(
+ -108652017022658969L,
+
ObjectStreamClass.lookup(TaskExecutionState.class).getSerialVersionUID());
+ Assertions.assertEquals(
+ 2646079648150626562L,
+
ObjectStreamClass.lookup(TaskDeployState.class).getSerialVersionUID());
+ }
+
+ @Test
+ void testTaskExecutionStateReadsPreviousWireForm() throws Exception {
+ TaskExecutionState failed = (TaskExecutionState)
deserialize(TASK_EXECUTION_STATE_FAILED);
+ Assertions.assertEquals(LOCATION, failed.getTaskGroupLocation());
+ Assertions.assertEquals(ExecutionState.FAILED,
failed.getExecutionState());
+ Assertions.assertEquals(
+ "java.lang.IllegalStateException: source failed",
failed.getThrowableMsg());
+
+ TaskExecutionState finished =
+ (TaskExecutionState)
deserialize(TASK_EXECUTION_STATE_FINISHED);
+ Assertions.assertEquals(LOCATION, finished.getTaskGroupLocation());
+ Assertions.assertEquals(ExecutionState.FINISHED,
finished.getExecutionState());
+ Assertions.assertNull(finished.getThrowableMsg());
+ }
+
+ @Test
+ void testTaskDeployStateReadsPreviousWireForm() throws Exception {
+ TaskDeployState success = (TaskDeployState)
deserialize(TASK_DEPLOY_STATE_SUCCESS);
+ Assertions.assertTrue(success.isSuccess());
+ Assertions.assertNull(success.getThrowableMsg());
+
+ TaskDeployState failed = (TaskDeployState)
deserialize(TASK_DEPLOY_STATE_FAILED);
+ Assertions.assertFalse(failed.isSuccess());
+ Assertions.assertEquals(
+ "java.lang.IllegalStateException: deploy failed",
failed.getThrowableMsg());
+ }
+
+ @Test
+ void testCurrentTaskStatesRoundTripThroughHazelcastJavaSerialization() {
+ InternalSerializationService service =
+ (InternalSerializationService) new
DefaultSerializationServiceBuilder().build();
+ try {
+ TaskExecutionState originalFailed =
+ new TaskExecutionState(
+ LOCATION,
+ ExecutionState.FAILED,
+ new IllegalStateException("source failed"));
+ TaskExecutionState failed = roundTrip(service, originalFailed);
+ Assertions.assertEquals(LOCATION, failed.getTaskGroupLocation());
+ Assertions.assertEquals(ExecutionState.FAILED,
failed.getExecutionState());
+ Assertions.assertEquals(originalFailed.getThrowableMsg(),
failed.getThrowableMsg());
+
+ TaskExecutionState finished =
+ roundTrip(service, new TaskExecutionState(LOCATION,
ExecutionState.FINISHED));
+ Assertions.assertEquals(LOCATION, finished.getTaskGroupLocation());
+ Assertions.assertEquals(ExecutionState.FINISHED,
finished.getExecutionState());
+ Assertions.assertNull(finished.getThrowableMsg());
+
+ TaskDeployState success = roundTrip(service,
TaskDeployState.success());
+ Assertions.assertTrue(success.isSuccess());
+ Assertions.assertNull(success.getThrowableMsg());
+
+ TaskDeployState originalDeployFailed =
+ TaskDeployState.failed(new IllegalStateException("deploy
failed"));
+ TaskDeployState deployFailed = roundTrip(service,
originalDeployFailed);
+ Assertions.assertFalse(deployFailed.isSuccess());
+ Assertions.assertEquals(
+ originalDeployFailed.getThrowableMsg(),
deployFailed.getThrowableMsg());
+ } finally {
+ service.dispose();
+ }
+ }
+
+ private static <T> T roundTrip(InternalSerializationService service, T
value) {
+ Data data = service.toData(value);
+ Assertions.assertEquals(
+ new JavaSerializer(false, false, null).getTypeId(),
+ data.getType(),
+ "Archived Java fixtures require the Hazelcast Java
serialization fallback");
+ return service.toObject(data);
+ }
+
+ private static Object deserialize(String base64) throws Exception {
+ try (ObjectInputStream input =
+ new ObjectInputStream(
+ new
ByteArrayInputStream(Base64.getDecoder().decode(base64)))) {
+ return input.readObject();
+ }
+ }
+}
