Make MessageHistory JSON-serializable

* Add `org.springframework.integration.history` to trusted default packaged of the `JacksonJsonUtils`
* Add `@JsonCreator` to `MessageHistory` `private` ctor to let it to be created automatically by Jackson
* Add `equals()` and `hashCode()` into `MessageHistory` for the proper `Message` comparison
* Add  `MessageHistory` into headers for testing with Redis JSON (de)serialization

**Cherry-pick to `5.4.x` & `5.3.x`**

spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.ListIterator;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.stream.Collectors;
 
@@ -42,6 +43,8 @@ import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+
 /**
  * @author Mark Fisher
  * @author Artem Bilan
@@ -133,7 +136,7 @@ public final class MessageHistory implements List<Properties>, Serializable {
 		return message;
 	}
 
-
+	@JsonCreator
 	private MessageHistory(List<Properties> components) {
 		Assert.notEmpty(components, "component list must not be empty");
 		this.components = components;
@@ -205,6 +208,21 @@ public final class MessageHistory implements List<Properties>, Serializable {
 		return this.components.lastIndexOf(o);
 	}
 
+	@Override public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (!(o instanceof MessageHistory)) {
+			return false;
+		}
+		MessageHistory that = (MessageHistory) o;
+		return this.components.equals(that.components);
+	}
+
+	@Override public int hashCode() {
+		return Objects.hash(this.components);
+	}
+
 	@Override
 	public String toString() {
 		return this.components

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -157,7 +157,8 @@ public final class JacksonJsonUtils {
 						"org.springframework.messaging.support",
 						"org.springframework.integration.support",
 						"org.springframework.integration.message",
-						"org.springframework.integration.store"
+						"org.springframework.integration.store",
+						"org.springframework.integration.history"
 				);
 
 		private final TypeIdResolver delegate;

spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2019 the original author or authors.
+ * Copyright 2007-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -40,6 +40,7 @@ import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.message.AdviceMessage;
@@ -143,7 +144,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
 		store.removeMessageGroup(this.groupId);
 		MessageGroup messageGroupA = store.getMessageGroup(this.groupId);
 		assertThat(messageGroupA).isNotSameAs(messageGroup);
-//		assertEquals(0, messageGroupA.getMarked().size());
+		//		assertEquals(0, messageGroupA.getMarked().size());
 		assertThat(messageGroupA.getMessages().size()).isEqualTo(0);
 		assertThat(messageGroupA.size()).isEqualTo(0);
 
@@ -438,6 +439,9 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
 		store.setValueSerializer(serializer);
 
 		Message<?> genericMessage = new GenericMessage<>(new Date());
+		NullChannel testComponent = new NullChannel();
+		testComponent.setBeanName("testChannel");
+		genericMessage = MessageHistory.write(genericMessage, testComponent);
 		Message<?> mutableMessage = new MutableMessage<>(UUID.randomUUID());
 		Message<?> adviceMessage = new AdviceMessage<>("foo", genericMessage);
 		ErrorMessage errorMessage = new ErrorMessage(new RuntimeException("test exception"), mutableMessage);
@@ -448,6 +452,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
 		assertThat(messageGroup.size()).isEqualTo(4);
 		List<Message<?>> messages = new ArrayList<>(messageGroup.getMessages());
 		assertThat(messages.get(0)).isEqualTo(genericMessage);
+		assertThat(messages.get(0).getHeaders()).containsKeys(MessageHistory.HEADER_NAME);
 		assertThat(messages.get(1)).isEqualTo(mutableMessage);
 		assertThat(messages.get(2)).isEqualTo(adviceMessage);
 		Message<?> errorMessageResult = messages.get(3);