Fix AbstractAwsMessageHandler impls for explicit headerMapper setting

Author: artembilan

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 @@ protected Expression getSendTimeoutExpression() {
 	 */
 	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 void setBodyExpression(Expression bodyExpression) {
 	@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 void setMessageConverter(MessageConverter messageConverter) {
 	@Override
 	protected void onInit() {
 		super.onInit();
-		setHeaderMapper(new SqsHeaderMapper());
+		if (!isHeaderMapperSet()) {
+			setHeaderMapper(new SqsHeaderMapper());
+		}
 		if (this.messageConverter == null) {
 			this.messageConverter = new GenericMessageConverter(getConversionService());
 		}