Fix AbstractAwsMessageHandler impls for explicit headerMapper setting

src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java

@@ -61,6 +61,8 @@ public abstract class AbstractAwsMessageHandler<H> extends AbstractMessageProduc
 
 	private HeaderMapper<H> headerMapper;
 
+	private boolean headerMapperSet;
+
 	public void setSendTimeout(long sendTimeout) {
 		setSendTimeoutExpression(new ValueExpression<>(sendTimeout));
 	}
@@ -84,6 +86,11 @@ public abstract class AbstractAwsMessageHandler<H> extends AbstractMessageProduc
 	 */
 	public void setHeaderMapper(HeaderMapper<H> headerMapper) {
 		this.headerMapper = headerMapper;
+		this.headerMapperSet = true;
+	}
+
+	protected boolean isHeaderMapperSet() {
+		return this.headerMapperSet;
 	}
 
 	/**

src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java

@@ -187,7 +187,9 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
 	@Override
 	protected void onInit() {
 		super.onInit();
-		setHeaderMapper(new SnsHeaderMapper());
+		if (!isHeaderMapperSet()) {
+			setHeaderMapper(new SnsHeaderMapper());
+		}
 		TypeLocator typeLocator = getEvaluationContext().getTypeLocator();
 		if (typeLocator instanceof StandardTypeLocator standardTypeLocator) {
 			/*

src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java

@@ -148,7 +148,9 @@ public class SqsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
 	@Override
 	protected void onInit() {
 		super.onInit();
-		setHeaderMapper(new SqsHeaderMapper());
+		if (!isHeaderMapperSet()) {
+			setHeaderMapper(new SqsHeaderMapper());
+		}
 		if (this.messageConverter == null) {
 			this.messageConverter = new GenericMessageConverter(getConversionService());
 		}