GH-2958: Also Add to Common Header in AMHEMS

Resolves #2958

The previous commit added the raw source message to a common header
`IntegrationMessageHeaderAccessor.SOURCE_DATA` so that the spring-cloud-stream
can construct an error message containing the data when using functions.

However, we also need to add it to the common header when the `ErrorMessage`
is created by Spring Integration for the normal message channel binder.

This allows the DLQ processing to be agnostic as to who created the
error message.

(cherry picked from commit 78a0ec7)

Author: garyrussell

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java

@@ -16,12 +16,12 @@
 
 package org.springframework.integration.amqp.support;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.core.AttributeAccessor;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageUtils;
 import org.springframework.lang.Nullable;
@@ -49,8 +49,11 @@ public class AmqpMessageHeaderErrorMessageStrategy implements ErrorMessageStrate
 	public ErrorMessage buildErrorMessage(Throwable throwable, @Nullable AttributeAccessor context) {
 		Object inputMessage = context == null ? null
 				: context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY);
-		Map<String, Object> headers = context == null ? new HashMap<String, Object>() :
-				Collections.singletonMap(AMQP_RAW_MESSAGE, context.getAttribute(AMQP_RAW_MESSAGE));
+		Map<String, Object> headers = new HashMap<String, Object>();
+		if (context != null) {
+			headers.put(AMQP_RAW_MESSAGE, context.getAttribute(AMQP_RAW_MESSAGE));
+			headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, context.getAttribute(AMQP_RAW_MESSAGE));
+		}
 		if (inputMessage instanceof Message) {
 			return new ErrorMessage(throwable, headers, (Message<?>) inputMessage);
 		}