GH-9369: Fix MutableMessageHeaders for serialization

Fixes: #9369 

* Implement `MutableMessageHeaders.readResolve()` to reinstate the instance

**Auto-cherry-pick to `6.3.x` & `6.2.x`**

Author: mitchmcd18

spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageHeaders.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.support;
 
+import java.io.ObjectStreamException;
+import java.io.Serial;
 import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.UUID;
@@ -31,6 +33,7 @@
  * @author David Turanski
  * @author Artem Bilan
  * @author Nathan Kurtyka
+ * @author Mitchell McDonald
  *
  * @since 4.2
  */
@@ -73,6 +76,11 @@ public Object remove(Object key) {
 		return super.getRawHeaders().remove(key);
 	}
 
+	@Serial
+	private Object readResolve() throws ObjectStreamException {
+		return new MutableMessageHeaders(this);
+	}
+
 	@Nullable
 	private static UUID extractId(@Nullable Map<String, Object> headers) {
 		if (headers != null && headers.containsKey(MessageHeaders.ID)) {

spring-integration-core/src/test/java/org/springframework/integration/support/MutableMessageTests.java

@@ -16,6 +16,11 @@
 
 package org.springframework.integration.support;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
@@ -26,10 +31,12 @@
 import org.springframework.messaging.MessageHeaders;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 
 /**
  * @author Stuart Williams
  * @author Nathan Kurtyka
+ * @author Mitchell McDonald
  *
  * @since 4.2
  */
@@ -104,4 +111,34 @@ public void testMessageHeaderIsSerializable() {
 		assertThat(mutableMessageBytes.getHeaders().getTimestamp()).isEqualTo(timestamp);
 	}
 
+	@Test
+	public void testMessageHeaderIsSerializableAndDeserializableWithNonSerializableValues()
+			throws IOException, ClassNotFoundException {
+
+		String payload = "payload";
+
+		Map<String, Object> headerMap = new HashMap<>();
+		headerMap.put("header1", "serializableValue");
+		headerMap.put("header2", new Object()); // Non-Serializable value
+
+		MutableMessage<String> mutableMessage = new MutableMessage<>(payload, headerMap);
+
+		ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+		ObjectOutputStream outputStream = new ObjectOutputStream(byteArrayOutputStream);
+		outputStream.writeObject(mutableMessage);
+		outputStream.flush();
+
+		ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray());
+		ObjectInputStream inputStream = new ObjectInputStream(byteArrayInputStream);
+		Object deserializedObject = inputStream.readObject();
+
+		assertThat(deserializedObject).isInstanceOf(MutableMessage.class);
+		MutableMessage<?> deserializedMessage =
+				(MutableMessage<?>) deserializedObject;
+
+		assertThat(deserializedMessage.getHeaders().get("header2")).isNull(); // Non-serializable value removed
+		assertThat(deserializedMessage.getHeaders().get("header1")).isEqualTo("serializableValue");
+		assertThatNoException().isThrownBy(() -> deserializedMessage.getRawHeaders().put("header3", "newValue"));
+	}
+
 }