This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12466-26ad719a48bd8ec17a32106747b86ccbe1f2b04b in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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(); + } + } +}
