GH-10083: Apply Nullability to the transformer package

Related to: #10083

The `transformer` pattern cannot produce `null` at all.
Therefore, making respective `Transformer.transform()` contract as non-null
leads to changes in many places where it was possible to return null
with subsequent `ReplyRequiredException` thrown from the `handler`.

The current fix is a bit radical and transformer-specific:
whenever we encounter `null` in the `transform()` call, we thrown a
`MessageTransformationException` instead.
Which is indeed making more sense from the `Transformer` perspective.
We may treat it as a breaking change, but the docs still state clearly that there is an exception in case of null.
Although it is different now.

The Nullability work here has also led to the review of the `getComponentType()` implementation
for the `Transformer` hierarchy.
Plus, the `AbstractMessageProcessingTransformer` is now an `IntegrationObjectSupport`.
The `MessageTransformingHandler` performs respective propagation in its `doInit()` to to the `transformer`
if that one is `IntegrationObjectSupport`.

The `HeaderEnricher` doesn't do a hard reference to the provided `Map` anymore to avoid
potential mutation of the source externally.
This has led to the change in the `HeaderEnricherSpec` to create a `HeaderEnricher` later in the `doGet()`
with all respective properties propagation.

Some of the test have suffered changes as well according to a new logic of the `transformer` package

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java

@@ -59,12 +59,15 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
 
 	private static final String HEADERS_MUST_NOT_BE_NULL = "'headers' must not be null";
 
-	protected final Map<String, HeaderValueMessageProcessor<?>> headerToAdd = new HashMap<>(); // NOSONAR - final
+	private final Map<String, HeaderValueMessageProcessor<?>> headerToAdd = new HashMap<>();
 
-	protected final HeaderEnricher headerEnricher = new HeaderEnricher(this.headerToAdd); // NOSONAR - final
+	private boolean defaultOverwrite;
+
+	private boolean shouldSkipNulls;
+
+	private @Nullable MessageProcessor<?> messageProcessor;
 
 	protected HeaderEnricherSpec() {
-		this.handler = new MessageTransformingHandler(this.headerEnricher);
 	}
 
 	/**
@@ -75,7 +78,7 @@ protected HeaderEnricherSpec() {
 	 * @see HeaderEnricher#setDefaultOverwrite(boolean)
 	 */
 	public HeaderEnricherSpec defaultOverwrite(boolean defaultOverwrite) {
-		this.headerEnricher.setDefaultOverwrite(defaultOverwrite);
+		this.defaultOverwrite = defaultOverwrite;
 		return _this();
 	}
 
@@ -85,7 +88,7 @@ public HeaderEnricherSpec defaultOverwrite(boolean defaultOverwrite) {
 	 * @see HeaderEnricher#setShouldSkipNulls(boolean)
 	 */
 	public HeaderEnricherSpec shouldSkipNulls(boolean shouldSkipNulls) {
-		this.headerEnricher.setShouldSkipNulls(shouldSkipNulls);
+		this.shouldSkipNulls = shouldSkipNulls;
 		return _this();
 	}
 
@@ -99,7 +102,7 @@ public HeaderEnricherSpec shouldSkipNulls(boolean shouldSkipNulls) {
 	 * @see HeaderEnricher#setMessageProcessor(MessageProcessor)
 	 */
 	public HeaderEnricherSpec messageProcessor(MessageProcessor<?> messageProcessor) {
-		this.headerEnricher.setMessageProcessor(messageProcessor);
+		this.messageProcessor = messageProcessor;
 		return _this();
 	}
 
@@ -877,7 +880,14 @@ public HeaderEnricherSpec headerChannelsToString(@Nullable String timeToLiveExpr
 
 	@Override
 	protected Tuple2<ConsumerEndpointFactoryBean, MessageTransformingHandler> doGet() {
-		this.componentsToRegister.put(this.headerEnricher, null);
+		HeaderEnricher headerEnricher = new HeaderEnricher(this.headerToAdd);
+		headerEnricher.setDefaultOverwrite(this.defaultOverwrite);
+		headerEnricher.setShouldSkipNulls(this.shouldSkipNulls);
+		if (this.messageProcessor != null) {
+			headerEnricher.setMessageProcessor(this.messageProcessor);
+		}
+		this.handler = new MessageTransformingHandler(headerEnricher);
+		this.componentsToRegister.put(headerEnricher, null);
 		return super.doGet();
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java

@@ -298,7 +298,7 @@ public static <T> DecodingTransformer<T> decoding(Codec codec, Expression typeEx
 	 * @return the {@link StreamTransformer} instance.
 	 */
 	public static StreamTransformer fromStream() {
-		return fromStream(null);
+		return new StreamTransformer();
 	}
 
 	/**
@@ -307,7 +307,7 @@ public static StreamTransformer fromStream() {
 	 * @param charset the charset.
 	 * @return the {@link StreamTransformer} instance.
 	 */
-	public static StreamTransformer fromStream(@Nullable String charset) {
+	public static StreamTransformer fromStream(String charset) {
 		return new StreamTransformer(charset);
 	}
 

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

@@ -154,7 +154,7 @@ public AbstractIntegrationMessageBuilder<T> setPriority(Integer priority) {
 	 * @see #copyHeadersIfAbsent(Map)
 	 */
 	public AbstractIntegrationMessageBuilder<T> filterAndCopyHeadersIfAbsent(Map<String, ?> headersToCopy,
-			@Nullable String... headerPatternsToFilter) {
+			String @Nullable ... headerPatternsToFilter) {
 
 		Map<String, ?> headers = headersToCopy;
 

spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java

@@ -18,15 +18,14 @@
 
 import java.util.Arrays;
 
-import org.springframework.beans.factory.BeanFactory;
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.context.Lifecycle;
+import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
-import org.springframework.integration.support.DefaultMessageBuilderFactory;
-import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.management.ManageableLifecycle;
-import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
@@ -40,17 +39,12 @@
  * @author Ngoc Nhan
  */
 public abstract class AbstractMessageProcessingTransformer
-		implements Transformer, BeanFactoryAware, ManageableLifecycle {
+		extends IntegrationObjectSupport
+		implements Transformer, ManageableLifecycle {
 
 	private final MessageProcessor<?> messageProcessor;
 
-	private BeanFactory beanFactory;
-
-	private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
-
-	private boolean messageBuilderFactorySet;
-
-	private String[] notPropagatedHeaders;
+	private String @Nullable [] notPropagatedHeaders;
 
 	private boolean selectiveHeaderPropagation;
 
@@ -60,21 +54,11 @@ protected AbstractMessageProcessingTransformer(MessageProcessor<?> messageProces
 	}
 
 	@Override
-	public void setBeanFactory(BeanFactory beanFactory) {
-		this.beanFactory = beanFactory;
+	protected void onInit() {
+		super.onInit();
 		if (this.messageProcessor instanceof BeanFactoryAware beanFactoryAware) {
-			beanFactoryAware.setBeanFactory(beanFactory);
-		}
-	}
-
-	protected MessageBuilderFactory getMessageBuilderFactory() {
-		if (!this.messageBuilderFactorySet) {
-			if (this.beanFactory != null) {
-				this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
-			}
-			this.messageBuilderFactorySet = true;
+			beanFactoryAware.setBeanFactory(getBeanFactory());
 		}
-		return this.messageBuilderFactory;
 	}
 
 	@Override
@@ -99,7 +83,7 @@ public boolean isRunning() {
 	/**
 	 * Set headers that will NOT be copied from the inbound message if
 	 * the handler is configured to copy headers.
-	 * @param headers the headers to not propagate from the inbound message.
+	 * @param headers the headers do not propagate from the inbound message.
 	 * @since 5.1
 	 */
 	public void setNotPropagatedHeaders(String... headers) {
@@ -115,16 +99,17 @@ public void setNotPropagatedHeaders(String... headers) {
 	public final Message<?> transform(Message<?> message) {
 		Object result = this.messageProcessor.processMessage(message);
 		if (result == null) {
-			return null;
+			throw new MessageTransformationException(message,
+					"MessageProcessor returned null in: " + getComponentName());
 		}
-		if (result instanceof Message<?>) {
-			return (Message<?>) result;
+		if (result instanceof Message<?> messageToReply) {
+			return messageToReply;
 		}
 
 		AbstractIntegrationMessageBuilder<?> messageBuilder;
 
-		if (result instanceof AbstractIntegrationMessageBuilder<?>) {
-			messageBuilder = (AbstractIntegrationMessageBuilder<?>) result;
+		if (result instanceof AbstractIntegrationMessageBuilder<?> integrationMessageBuilder) {
+			messageBuilder = integrationMessageBuilder;
 		}
 		else {
 			messageBuilder = getMessageBuilderFactory().withPayload(result);

spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractPayloadTransformer.java

@@ -34,7 +34,7 @@ public abstract class AbstractPayloadTransformer<T, U> extends AbstractTransform
 	@Override
 	@SuppressWarnings("unchecked")
 	public final U doTransform(Message<?> message) {
-		return this.transformPayload((T) message.getPayload());
+		return transformPayload((T) message.getPayload());
 	}
 
 	protected abstract U transformPayload(T payload);

spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractTransformer.java

@@ -24,38 +24,31 @@
  *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Artem Bilan
  */
 public abstract class AbstractTransformer extends IntegrationObjectSupport implements Transformer {
 
 	@Override
 	public final Message<?> transform(Message<?> message) {
 		try {
-			Object result = this.doTransform(message);
-			if (result == null) {
-				return null;
-			}
-			return (result instanceof Message) ? (Message<?>) result
+			Object result = doTransform(message);
+			return result instanceof Message<?> resultMessage
+					? resultMessage
 					: getMessageBuilderFactory().withPayload(result).copyHeaders(message.getHeaders()).build();
 		}
-		catch (MessageTransformationException e) { // NOSONAR - catch and throw
-			throw e;
-		}
-		catch (Exception e) {
-			throw new MessageTransformationException(message, "failed to transform message", e);
+		catch (Exception ex) {
+			if (ex instanceof MessageTransformationException messageTransformationException) {
+				throw messageTransformationException;
+			}
+			throw new MessageTransformationException(message, "failed to transform message", ex);
 		}
 	}
 
-	@Override
-	public String getComponentType() {
-		return "transformer";
-	}
-
 	/**
 	 * Subclasses must implement this method to provide the transformation
 	 * logic. If the return value is itself a Message, it will be used as the
 	 * result. Otherwise, any non-null return value will be used as the payload
 	 * of the result Message.
-	 *
 	 * @param message The message.
 	 * @return The result of the transformation.
 	 */

spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckInTransformer.java

@@ -39,7 +39,6 @@ public class ClaimCheckInTransformer extends AbstractTransformer implements Inte
 
 	/**
 	 * Create a claim check-in transformer that will delegate to the provided MessageStore.
-	 *
 	 * @param messageStore The message store.
 	 */
 	public ClaimCheckInTransformer(MessageStore messageStore) {
@@ -59,7 +58,6 @@ public IntegrationPatternType getIntegrationPatternType() {
 
 	@Override
 	protected Object doTransform(Message<?> message) {
-		Assert.notNull(message, "message must not be null");
 		UUID id = message.getHeaders().getId();
 		Assert.notNull(id, "ID header must not be null");
 		this.messageStore.addMessage(message);

spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckOutTransformer.java

@@ -47,7 +47,6 @@ public class ClaimCheckOutTransformer extends AbstractTransformer implements Int
 
 	/**
 	 * Create a claim check-out transformer that will delegate to the provided MessageStore.
-	 *
 	 * @param messageStore The message store.
 	 */
 	public ClaimCheckOutTransformer(MessageStore messageStore) {
@@ -71,7 +70,6 @@ public IntegrationPatternType getIntegrationPatternType() {
 
 	@Override
 	protected Object doTransform(Message<?> message) {
-		Assert.notNull(message, "message must not be null");
 		Assert.isTrue(message.getPayload() instanceof UUID, "payload must be a UUID");
 		UUID id = (UUID) message.getPayload();
 		Message<?> retrievedMessage;