GH-9442: Register dynamic flows as singletons

Fixes: #9442
Issue link: https://github.com/spring-projects/spring-integration/issues/9442

When we register a `BeanDefinition`, its metadata is cached in the `DefaultListableBeanFactory.mergedBeanDefinitionHolders`
and it is not removed when we destroy bean.
This causes a memory leak with dynamic integration flows where even we destroy beans, their metadata is still cached.
This has more serious impact when random ids are used for dynamic integration flows.

* Rework `IntegrationFlowContext` logic to register singletons instead of `BeanDefinition`
* Adjust `IntegrationFlowBeanPostProcessor` logic to register singletons or `BeanDefinition`
according to the presence of `BeanDefinition` of the `IntegrationFlow` we are processing
* Introduce `ComponentSourceAware` to mimic a `BeanDefinition` metadata logic for bean source and its description.
This info helps a lot with exceptions where class in the `IntegrationFlow` might be fully from a Spring Integration package.
So, to give a clue what end-user code is related to the exception,
such a `ComponentSourceAware` is there to preserver and transfer "bean source"
* Implement `ComponentSourceAware` in the `IntegrationObjectSupport` since this is a central
place where we generate info for the Spring Integration exceptions
* Implement `ComponentSourceAware` in the `StandardIntegrationFlow` to propagate bean source info
down to its components for the mentioned `IntegrationObjectSupport` logic
* Introduce inner `StandardIntegrationFlowContext.IntegrationFlowComponentSourceAwareAdapter`
to be able to transfer "bean source" info from builder down to the target `IntegrationFlow` bean

spring-integration-core/src/main/java/org/springframework/integration/context/ComponentSourceAware.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.context;
+
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.lang.Nullable;
+
+/**
+ * The contract to supply and provide useful information about
+ * a bean definition (or singleton) source - the place where this bean is declared.
+ * Usually populated from a respective {@link org.springframework.beans.factory.config.BeanDefinition}
+ * or via Spring Integration infrastructure.
+ * <p>
+ * The information from this contract is typically used from exceptions to easy determine
+ * the place in the application resources where this bean is declared.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.4
+ *
+ * @see org.springframework.beans.factory.config.BeanDefinition
+ */
+public interface ComponentSourceAware extends BeanNameAware {
+
+	/**
+	 * Set a configuration source {@code Object} for this bean definition.
+	 * For normal {@link org.springframework.beans.factory.config.BeanDefinition} this is supplied
+	 * by application context automatically.
+	 * Could be useful when bean is registered at runtime via
+	 * {@link org.springframework.beans.factory.config.SingletonBeanRegistry#registerSingleton(String, Object)}
+	 * @param source the configuration source
+	 */
+	void setComponentSource(Object source);
+
+	/**
+	 * Return the configuration source {@code Object} for this bean (maybe {@code null}).
+	 * Usually (if not set explicitly) a {@link org.springframework.beans.factory.config.BeanDefinition#getSource()}.
+	 * @return the configuration source for the bean (if any).
+	 */
+	@Nullable
+	Object getComponentSource();
+
+	/**
+	 * Set a human-readable description of this bean.
+	 * For normal bean definition a {@link org.springframework.beans.factory.config.BeanDefinition#getDescription()}
+	 * is used.
+	 * @param description the bean description
+	 */
+	void setComponentDescription(String description);
+
+	/**
+	 * Return a human-readable description of this bean.
+	 * Usually (if not set explicitly) a {@link org.springframework.beans.factory.config.BeanDefinition#getDescription()}.
+	 * @return the bean description (if any).
+	 */
+	@Nullable
+	String getComponentDescription();
+
+	/**
+	 * Return the bean name populated by the {@link BeanNameAware#setBeanName(String)}.
+	 * @return the bean name.
+	 */
+	@Nullable
+	String getBeanName();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -22,7 +22,6 @@ import org.springframework.aop.framework.AopProxyUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
-import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
@@ -64,7 +63,7 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @author Artem Bilan
  */
-public abstract class IntegrationObjectSupport implements BeanNameAware, NamedComponent,
+public abstract class IntegrationObjectSupport implements ComponentSourceAware, NamedComponent,
 		ApplicationContextAware, BeanFactoryAware, InitializingBean, ExpressionCapable {
 
 	protected static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
@@ -95,6 +94,10 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 
 	private Expression expression;
 
+	private Object beanSource;
+
+	private String beanDescription;
+
 	private boolean initialized;
 
 	@Override
@@ -132,28 +135,41 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 		return null;
 	}
 
+	@Override
+	public void setComponentSource(Object source) {
+		this.beanSource = source;
+	}
+
+	@Nullable
+	@Override
+	public Object getComponentSource() {
+		return this.beanSource;
+	}
+
+	@Override
+	public void setComponentDescription(String description) {
+		this.beanDescription = description;
+	}
+
+	@Nullable
+	@Override
+	public String getComponentDescription() {
+		return this.beanDescription;
+	}
+
 	public String getBeanDescription() {
-		String description = null;
-		Object source = null;
-
-		if (this.beanFactory instanceof ConfigurableListableBeanFactory &&
-				((ConfigurableListableBeanFactory) this.beanFactory).containsBeanDefinition(this.beanName)) {
-			BeanDefinition beanDefinition =
-					((ConfigurableListableBeanFactory) this.beanFactory).getBeanDefinition(this.beanName);
-			description = beanDefinition.getResourceDescription();
-			source = beanDefinition.getSource();
+		StringBuilder sb =
+				new StringBuilder("bean '")
+						.append(this.beanName).append("'");
+		String beanComponentName = getComponentName();
+		if (!this.beanName.equals(beanComponentName)) {
+			sb.append(" for component '").append(beanComponentName).append("'");
 		}
-
-		StringBuilder sb = new StringBuilder("bean '")
-				.append(this.beanName).append("'");
-		if (!this.beanName.equals(getComponentName())) {
-			sb.append(" for component '").append(getComponentName()).append("'");
+		if (this.beanDescription != null) {
+			sb.append("; defined in: '").append(this.beanDescription).append("'");
 		}
-		if (description != null) {
-			sb.append("; defined in: '").append(description).append("'");
-		}
-		if (source != null) {
-			sb.append("; from source: '").append(source).append("'");
+		if (this.beanSource != null) {
+			sb.append("; from source: '").append(this.beanSource).append("'");
 		}
 		return sb.toString();
 	}
@@ -205,6 +221,15 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 				this.messageBuilderFactory = new DefaultMessageBuilderFactory();
 			}
 		}
+		if (this.beanSource == null && this.beanName != null
+				&& this.beanFactory instanceof ConfigurableListableBeanFactory configurableListableBeanFactory
+				&& configurableListableBeanFactory.containsBeanDefinition(this.beanName)) {
+			BeanDefinition beanDefinition = configurableListableBeanFactory.getBeanDefinition(this.beanName);
+			this.beanSource = beanDefinition.getSource();
+			if (this.beanDescription == null) {
+				this.beanDescription = beanDefinition.getResourceDescription();
+			}
+		}
 		onInit();
 		this.initialized = true;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2024 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.function.Supplier;
 
 import org.reactivestreams.Publisher;
 
+import org.springframework.integration.context.ComponentSourceAware;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
@@ -59,18 +60,64 @@ import org.springframework.util.Assert;
  *
  * @since 5.0
  */
-public abstract class IntegrationFlowAdapter implements IntegrationFlow, ManageableSmartLifecycle {
+public abstract class IntegrationFlowAdapter
+		implements IntegrationFlow, ManageableSmartLifecycle, ComponentSourceAware {
 
 	private final AtomicBoolean running = new AtomicBoolean();
 
 	private StandardIntegrationFlow targetIntegrationFlow;
 
+	private String beanName;
+
+	private Object beanSource;
+
+	private String beanDescription;
+
+	@Override
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
+	@Nullable
+	@Override
+	public String getBeanName() {
+		return this.beanName;
+	}
+
+	@Override
+	public void setComponentSource(Object source) {
+		this.beanSource = source;
+	}
+
+	@Nullable
+	@Override
+	public Object getComponentSource() {
+		return this.beanSource;
+	}
+
+	@Override
+	public void setComponentDescription(String description) {
+		this.beanDescription = description;
+	}
+
+	@Nullable
+	@Override
+	public String getComponentDescription() {
+		return this.beanDescription;
+	}
+
 	@Override
 	public final void configure(IntegrationFlowDefinition<?> flow) {
 		IntegrationFlowDefinition<?> targetFlow = buildFlow();
 		flow.integrationComponents.clear();
 		flow.integrationComponents.putAll(targetFlow.integrationComponents);
 		this.targetIntegrationFlow = flow.get();
+		if (this.beanSource != null) {
+			this.targetIntegrationFlow.setComponentSource(this.beanSource);
+		}
+		if (this.beanDescription != null) {
+			this.targetIntegrationFlow.setComponentDescription(this.beanDescription);
+		}
 	}
 
 	@Nullable

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2024 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.
@@ -24,8 +24,8 @@ import java.util.ListIterator;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.context.SmartLifecycle;
+import org.springframework.integration.context.ComponentSourceAware;
 import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
@@ -69,7 +69,7 @@ import org.springframework.messaging.MessageChannel;
  * @see SmartLifecycle
  */
 public class StandardIntegrationFlow
-		implements IntegrationFlow, SmartLifecycle, BeanNameAware, NamedComponent {
+		implements IntegrationFlow, SmartLifecycle, ComponentSourceAware, NamedComponent {
 
 	private final Map<Object, String> integrationComponents;
 
@@ -79,6 +79,10 @@ public class StandardIntegrationFlow
 
 	private String beanName;
 
+	private Object beanSource;
+
+	private String beanDescription;
+
 	StandardIntegrationFlow(Map<Object, String> integrationComponents) {
 		this.integrationComponents = new LinkedHashMap<>(integrationComponents);
 	}
@@ -98,6 +102,34 @@ public class StandardIntegrationFlow
 		return "integration-flow";
 	}
 
+	@Override
+	public void setComponentSource(Object source) {
+		this.beanSource = source;
+	}
+
+	@Nullable
+	@Override
+	public Object getComponentSource() {
+		return this.beanSource;
+	}
+
+	@Override
+	public void setComponentDescription(String description) {
+		this.beanDescription = description;
+	}
+
+	@Nullable
+	@Override
+	public String getComponentDescription() {
+		return this.beanDescription;
+	}
+
+	@Nullable
+	@Override
+	public String getBeanName() {
+		return this.beanName;
+	}
+
 	@Override
 	public void configure(IntegrationFlowDefinition<?> flow) {
 		throw new UnsupportedOperationException();

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

@@ -29,17 +29,18 @@ import org.springframework.beans.factory.BeanCreationNotAllowedException;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.BeanFactoryUtils;
 import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.BeanDefinitionCustomizer;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
-import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.config.EmbeddedValueResolver;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionOverrideException;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -52,7 +53,6 @@ import org.springframework.context.ResourceLoaderAware;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.DescriptiveResource;
-import org.springframework.core.type.MethodMetadata;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
@@ -60,6 +60,7 @@ import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
+import org.springframework.integration.context.ComponentSourceAware;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.ComponentsRegistration;
@@ -97,7 +98,7 @@ public class IntegrationFlowBeanPostProcessor
 
 	private StringValueResolver embeddedValueResolver;
 
-	private ConfigurableListableBeanFactory beanFactory;
+	private DefaultListableBeanFactory beanFactory;
 
 	private volatile IntegrationFlowContext flowContext;
 
@@ -109,7 +110,7 @@ public class IntegrationFlowBeanPostProcessor
 		);
 
 		this.applicationContext = (ConfigurableApplicationContext) applicationContext;
-		this.beanFactory = this.applicationContext.getBeanFactory();
+		this.beanFactory = (DefaultListableBeanFactory) this.applicationContext.getBeanFactory();
 		this.embeddedValueResolver = new EmbeddedValueResolver(this.beanFactory);
 	}
 
@@ -142,8 +143,16 @@ public class IntegrationFlowBeanPostProcessor
 		}
 	}
 
-	private Object processStandardIntegrationFlow(StandardIntegrationFlow flow, String flowBeanName) { // NOSONAR
-		// complexity
+	private Object processStandardIntegrationFlow(StandardIntegrationFlow flow, String flowBeanName) { // NOSONAR complexity
+		boolean registerBeanDefinitions = this.beanFactory.containsBeanDefinition(flowBeanName);
+		if (registerBeanDefinitions) {
+			BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(flowBeanName);
+			flow.setComponentSource(beanDefinition.getSource());
+			flow.setComponentDescription(beanDefinition.getDescription());
+		}
+		Object beanSource = flow.getComponentSource();
+		String beanDescription = flow.getComponentDescription();
+
 		String flowNamePrefix = flowBeanName + ".";
 		if (this.flowContext == null) {
 			this.flowContext = this.beanFactory.getBean(IntegrationFlowContext.class);
@@ -172,18 +181,21 @@ public class IntegrationFlowBeanPostProcessor
 				if (noBeanPresentForComponent(messageHandler, flowBeanName)) {
 					String handlerBeanName = generateBeanName(messageHandler, flowNamePrefix);
 
-					registerComponent(messageHandler, handlerBeanName, flowBeanName);
+					registerComponent(registerBeanDefinitions, messageHandler, handlerBeanName,
+							beanSource, beanDescription, flowBeanName);
 					this.beanFactory.registerAlias(handlerBeanName, id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX);
 				}
 
-				registerComponent(endpoint, id, flowBeanName);
+				registerComponent(registerBeanDefinitions, endpoint, id, beanSource,
+						beanDescription, flowBeanName);
 				targetIntegrationComponents.put(endpoint, id);
 			}
 			else if (component instanceof MessageChannelReference messageChannelReference) {
 				String channelBeanName = messageChannelReference.name();
 				if (!this.beanFactory.containsBean(channelBeanName)) {
 					DirectChannel directChannel = new DirectChannel();
-					registerComponent(directChannel, channelBeanName, flowBeanName);
+					registerComponent(registerBeanDefinitions, directChannel, channelBeanName,
+							beanSource, beanDescription, flowBeanName);
 					targetIntegrationComponents.put(directChannel, channelBeanName);
 				}
 			}
@@ -194,9 +206,9 @@ public class IntegrationFlowBeanPostProcessor
 							.stream()
 							.filter(o -> noBeanPresentForComponent(o.getKey(), flowBeanName))
 							.forEach(o ->
-									registerComponent(o.getKey(),
+									registerComponent(registerBeanDefinitions, o.getKey(),
 											generateBeanName(o.getKey(), flowNamePrefix, o.getValue(),
-													useFlowIdAsPrefix)));
+													useFlowIdAsPrefix), beanSource, beanDescription, spec.getId()));
 				}
 				SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.getObject().getT1();
 				String id = spec.getId();
@@ -208,7 +220,8 @@ public class IntegrationFlowBeanPostProcessor
 					id = flowNamePrefix + id;
 				}
 
-				registerComponent(pollingChannelAdapterFactoryBean, id, flowBeanName);
+				registerComponent(registerBeanDefinitions, pollingChannelAdapterFactoryBean, id,
+						beanSource, beanDescription, flowBeanName);
 				targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id);
 
 				MessageSource<?> messageSource = spec.getObject().getT2();
@@ -219,7 +232,8 @@ public class IntegrationFlowBeanPostProcessor
 
 						messageSourceId = namedComponent.getComponentName();
 					}
-					registerComponent(messageSource, messageSourceId, flowBeanName);
+					registerComponent(registerBeanDefinitions, messageSource, messageSourceId,
+							beanSource, beanDescription, flowBeanName);
 				}
 			}
 			else {
@@ -233,7 +247,8 @@ public class IntegrationFlowBeanPostProcessor
 										BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
 							}
 						}
-						registerComponent(component, channelBeanName, flowBeanName);
+						registerComponent(registerBeanDefinitions, component, channelBeanName,
+								beanSource, beanDescription, flowBeanName);
 						targetIntegrationComponents.put(component, channelBeanName);
 					}
 					else if (component instanceof FixedSubscriberChannel fixedSubscriberChannel) {
@@ -242,7 +257,8 @@ public class IntegrationFlowBeanPostProcessor
 							channelBeanName = flowNamePrefix + "channel" +
 									BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
 						}
-						registerComponent(component, channelBeanName, flowBeanName);
+						registerComponent(registerBeanDefinitions, component, channelBeanName,
+								beanSource, beanDescription, flowBeanName);
 						targetIntegrationComponents.put(component, channelBeanName);
 					}
 					else if (component instanceof StandardIntegrationFlow) {
@@ -251,7 +267,8 @@ public class IntegrationFlowBeanPostProcessor
 										? entry.getValue()
 										: flowNamePrefix + "subFlow" +
 										BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + subFlowNameIndex++;
-						registerComponent(component, subFlowBeanName, flowBeanName);
+						registerComponent(registerBeanDefinitions, component, subFlowBeanName,
+								beanSource, beanDescription, flowBeanName);
 						targetIntegrationComponents.put(component, subFlowBeanName);
 					}
 					else if (component instanceof AnnotationGatewayProxyFactoryBean<?> gateway) {
@@ -264,15 +281,21 @@ public class IntegrationFlowBeanPostProcessor
 							gatewayId = flowNamePrefix + "gateway";
 						}
 
-						registerComponent(gateway, gatewayId, flowBeanName,
-								beanDefinition -> {
-									RootBeanDefinition definition = (RootBeanDefinition) beanDefinition;
-									Class<?> serviceInterface = gateway.getObjectType();
-									definition.setSource(new DescriptiveResource("" + serviceInterface));
-									definition.setTargetType(
-											ResolvableType.forClassWithGenerics(AnnotationGatewayProxyFactoryBean.class,
-													serviceInterface));
-								});
+						if (registerBeanDefinitions) {
+							registerBeanDefinition(gateway, gatewayId, beanSource,
+									beanDescription, flowBeanName,
+									beanDefinition -> {
+										RootBeanDefinition definition = (RootBeanDefinition) beanDefinition;
+										Class<?> serviceInterface = gateway.getObjectType();
+										definition.setSource(new DescriptiveResource("" + serviceInterface));
+										definition.setTargetType(
+												ResolvableType.forClassWithGenerics(AnnotationGatewayProxyFactoryBean.class,
+														serviceInterface));
+									});
+						}
+						else {
+							registerSingleton(gateway, gatewayId, beanSource, beanDescription, flowBeanName);
+						}
 
 						targetIntegrationComponents.put(component, gatewayId);
 					}
@@ -280,7 +303,8 @@ public class IntegrationFlowBeanPostProcessor
 						String generatedBeanName =
 								generateBeanName(component, flowNamePrefix, entry.getValue(), useFlowIdAsPrefix);
 
-						registerComponent(component, generatedBeanName, flowBeanName);
+						registerComponent(registerBeanDefinitions, component, generatedBeanName,
+								beanSource, beanDescription, flowBeanName);
 						targetIntegrationComponents.put(component, generatedBeanName);
 					}
 				}
@@ -352,13 +376,20 @@ public class IntegrationFlowBeanPostProcessor
 		if (bean instanceof ComponentsRegistration componentsRegistration) {
 			Map<Object, String> componentsToRegister = componentsRegistration.getComponentsToRegister();
 			if (!CollectionUtils.isEmpty(componentsToRegister)) {
+				boolean registerBeanDefinitions = this.beanFactory.containsBeanDefinition(beanName);
+				BeanDefinition beanDefinition = null;
+				if (registerBeanDefinitions) {
+					beanDefinition = this.beanFactory.getBeanDefinition(beanName);
+				}
+				Object beanDefinitionSource = registerBeanDefinitions ? beanDefinition.getSource() : null;
+				String beanDefinitionDescription = registerBeanDefinitions ? beanDefinition.getDescription() : null;
 				componentsToRegister.entrySet()
 						.stream()
 						.filter(component -> noBeanPresentForComponent(component.getKey(), beanName))
 						.forEach(component ->
-								registerComponent(component.getKey(),
-										generateBeanName(component.getKey(), beanName, component.getValue(), false)));
-
+								registerComponent(registerBeanDefinitions, component.getKey(),
+										generateBeanName(component.getKey(), beanName, component.getValue(), false),
+										beanDefinitionSource, beanDefinitionDescription, beanName));
 			}
 		}
 	}
@@ -405,21 +436,34 @@ public class IntegrationFlowBeanPostProcessor
 				return true;
 			}
 			else {
-				BeanDefinition existingBeanDefinition =
-						IntegrationContextUtils.getBeanDefinition(beanName, this.beanFactory);
-				if (!ConfigurableBeanFactory.SCOPE_PROTOTYPE.equals(existingBeanDefinition.getScope())
-						&& !instance.equals(this.beanFactory.getBean(beanName))) {
+				BeanDefinition existingBeanDefinition = null;
+				try {
+					existingBeanDefinition = IntegrationContextUtils.getBeanDefinition(beanName, this.beanFactory);
+				}
+				catch (NoSuchBeanDefinitionException ex) {
+					// Ignore and move on as no bean definition (possibly just singleton?)
+				}
 
-					AbstractBeanDefinition beanDefinition =
-							BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) instance.getClass(),
-											() -> instance)
-									.getBeanDefinition();
-					beanDefinition.setResourceDescription("the '" + parentBeanName + "' bean definition");
-					throw new BeanDefinitionOverrideException(beanName, beanDefinition, existingBeanDefinition);
-				}
-				else {
-					return false;
+				if (existingBeanDefinition == null
+						|| !ConfigurableBeanFactory.SCOPE_PROTOTYPE.equals(existingBeanDefinition.getScope())) {
+
+					Object existingBean = this.beanFactory.getBean(beanName);
+					if (!instance.equals(existingBean)) {
+						AbstractBeanDefinition beanDefinition =
+								BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) instance.getClass(),
+												() -> instance)
+										.getBeanDefinition();
+						beanDefinition.setResourceDescription("the '" + parentBeanName + "' bean definition");
+						if (existingBeanDefinition == null) {
+							existingBeanDefinition =
+									BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) existingBean.getClass(),
+													() -> existingBean)
+											.getBeanDefinition();
+						}
+						throw new BeanDefinitionOverrideException(beanName, beanDefinition, existingBeanDefinition);
+					}
 				}
+				return false;
 			}
 		}
 
@@ -427,12 +471,21 @@ public class IntegrationFlowBeanPostProcessor
 				.containsValue(instance);
 	}
 
-	private void registerComponent(Object component, String beanName) {
-		registerComponent(component, beanName, null);
+	private void registerComponent(boolean registerBeanDefinition,
+			Object component, String beanName, @Nullable Object source,
+			@Nullable String description, @Nullable String parentName) {
+
+		if (registerBeanDefinition) {
+			registerBeanDefinition(component, beanName, source, description, parentName);
+		}
+		else {
+			registerSingleton(component, beanName, source, description, parentName);
+		}
 	}
 
 	@SuppressWarnings("unchecked")
-	private void registerComponent(Object component, String beanName, @Nullable String parentName,
+	private void registerBeanDefinition(Object component, String beanName, @Nullable Object source,
+			@Nullable String description, @Nullable String parentName,
 			BeanDefinitionCustomizer... customizers) {
 
 		AbstractBeanDefinition beanDefinition =
@@ -440,20 +493,31 @@ public class IntegrationFlowBeanPostProcessor
 						.applyCustomizers(customizers)
 						.getRawBeanDefinition();
 
-		if (parentName != null && this.beanFactory.containsBeanDefinition(parentName)) {
-			AbstractBeanDefinition parentBeanDefinition =
-					(AbstractBeanDefinition) this.beanFactory.getBeanDefinition(parentName);
-			beanDefinition.setResource(parentBeanDefinition.getResource());
-			Object source = parentBeanDefinition.getSource();
-			if (source instanceof MethodMetadata) {
-				source = "bean method " + ((MethodMetadata) source).getMethodName();
-			}
-			beanDefinition.setSource(source);
+		beanDefinition.setSource(source);
+		beanDefinition.setDescription(description);
+
+		if (parentName != null) {
 			this.beanFactory.registerDependentBean(parentName, beanName);
 		}
 
 		((BeanDefinitionRegistry) this.beanFactory).registerBeanDefinition(beanName, beanDefinition);
+		this.beanFactory.getBean(beanName);
+	}
 
+	private void registerSingleton(Object component, String beanName, @Nullable Object source,
+			@Nullable String description, @Nullable String parentName) {
+
+		if (component instanceof ComponentSourceAware componentSourceAware) {
+			componentSourceAware.setComponentSource(source);
+			componentSourceAware.setComponentDescription(description);
+		}
+
+		if (parentName != null) {
+			this.beanFactory.registerDependentBean(parentName, beanName);
+		}
+
+		this.beanFactory.registerSingleton(beanName, component);
+		this.beanFactory.initializeBean(component, beanName);
 		this.beanFactory.getBean(beanName);
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2023 the original author or authors.
+ * Copyright 2018-2024 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.
@@ -214,6 +214,14 @@ public interface IntegrationFlowContext {
 		 */
 		IntegrationFlowRegistrationBuilder setSource(Object source);
 
+		/**
+		 * Set a human-readable description of this integration flow.
+		 * @param description the description for integration flow instance.
+		 * @return the current builder instance
+		 * @since 6.4
+		 */
+		IntegrationFlowRegistrationBuilder setDescription(String description);
+
 		/**
 		 * Invoke this method to prefix bean names in the flow with the (required) flow id
 		 * and a period. This is useful if you wish to register the same flow multiple times

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

@@ -27,14 +27,15 @@ import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.BeanFactoryUtils;
-import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
-import org.springframework.beans.factory.support.AbstractBeanDefinition;
-import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.context.SmartLifecycle;
-import org.springframework.core.type.MethodMetadata;
+import org.springframework.integration.context.ComponentSourceAware;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlowAdapter;
+import org.springframework.integration.dsl.IntegrationFlowBuilder;
+import org.springframework.integration.dsl.IntegrationFlowDefinition;
+import org.springframework.integration.dsl.StandardIntegrationFlow;
 import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -59,21 +60,18 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 
 	private final Lock registerFlowsLock = new ReentrantLock();
 
-	private ConfigurableListableBeanFactory beanFactory;
-
-	private BeanDefinitionRegistry beanDefinitionRegistry;
+	private DefaultListableBeanFactory beanFactory;
 
 	StandardIntegrationFlowContext() {
 	}
 
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
-		Assert.isInstanceOf(ConfigurableListableBeanFactory.class, beanFactory,
+		Assert.isInstanceOf(DefaultListableBeanFactory.class, beanFactory,
 				"To use Spring Integration Java DSL the 'beanFactory' has to be an instance of " +
 						"'ConfigurableListableBeanFactory'. " +
 						"Consider using 'GenericApplicationContext' implementation.");
-		this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
-		this.beanDefinitionRegistry = (BeanDefinitionRegistry) this.beanFactory;
+		this.beanFactory = (DefaultListableBeanFactory) beanFactory;
 	}
 
 	/**
@@ -107,7 +105,7 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 						"An existing IntegrationFlowRegistration must be destroyed before overriding.");
 			}
 
-			integrationFlow = registerFlowBean(integrationFlow, flowId, builder.source);
+			integrationFlow = registerFlowBean(integrationFlow, flowId, builder.source, builder.description);
 		}
 		finally {
 			this.registerFlowsLock.unlock();
@@ -116,7 +114,8 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 		builder.integrationFlow = integrationFlow;
 
 		final String theFlowId = flowId;
-		builder.additionalBeans.forEach((key, value) -> registerBean(key, value, theFlowId));
+		builder.additionalBeans.forEach((key, value) ->
+				registerBean(key, value, theFlowId, builder.source, builder.description));
 
 		IntegrationFlowRegistration registration =
 				new StandardIntegrationFlowRegistration(integrationFlow, this, flowId);
@@ -133,38 +132,37 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 		return registration;
 	}
 
-	private IntegrationFlow registerFlowBean(IntegrationFlow flow, String flowId, @Nullable Object source) {
-		AbstractBeanDefinition beanDefinition =
-				BeanDefinitionBuilder.genericBeanDefinition(IntegrationFlow.class, () -> flow)
-						.getRawBeanDefinition();
-		beanDefinition.setSource(source);
-		this.beanDefinitionRegistry.registerBeanDefinition(flowId, beanDefinition);
-		return this.beanFactory.getBean(flowId, IntegrationFlow.class);
+	private IntegrationFlow registerFlowBean(IntegrationFlow flow, @Nullable String beanName,
+			@Nullable Object source, @Nullable String description) {
+
+		IntegrationFlow flowToRegister = flow;
+
+		if (!(flow instanceof StandardIntegrationFlow) && !(flow instanceof IntegrationFlowAdapter)) {
+			flowToRegister = new IntegrationFlowComponentSourceAwareAdapter(flow);
+		}
+
+		return registerBean(flowToRegister, beanName, null, source, description);
 	}
 
 	@SuppressWarnings("unchecked")
-	private void registerBean(Object bean, @Nullable String beanNameArg, String parentName) {
-		String beanName = beanNameArg;
-		if (beanName == null) {
-			beanName = generateBeanName(bean, parentName);
-		}
+	private <B> B registerBean(B bean, @Nullable String beanNameArg, @Nullable String parentName,
+			@Nullable Object source, @Nullable String description) {
 
-		AbstractBeanDefinition beanDefinition =
-				BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) bean.getClass(), () -> bean)
-						.getRawBeanDefinition();
+		String beanName = beanNameArg != null ? beanNameArg : generateBeanName(bean, parentName);
 
 		if (parentName != null) {
-			AbstractBeanDefinition parentBeanDefinition =
-					(AbstractBeanDefinition) this.beanFactory.getBeanDefinition(parentName);
-			Object source = parentBeanDefinition.getSource();
-			if (source instanceof MethodMetadata) {
-				source = "bean method " + ((MethodMetadata) source).getMethodName();
-			}
-			beanDefinition.setSource(source);
 			this.beanFactory.registerDependentBean(parentName, beanName);
 		}
-		this.beanDefinitionRegistry.registerBeanDefinition(beanName, beanDefinition);
-		this.beanFactory.getBean(beanName);
+		if (bean instanceof ComponentSourceAware componentSourceAware) {
+			if (source != null && componentSourceAware.getComponentSource() == null) {
+				componentSourceAware.setComponentSource(source);
+			}
+			if (description != null && componentSourceAware.getComponentDescription() == null) {
+				componentSourceAware.setComponentDescription(description);
+			}
+		}
+		this.beanFactory.registerSingleton(beanName, bean);
+		return (B) this.beanFactory.initializeBean(bean, beanName);
 	}
 
 	/**
@@ -192,7 +190,7 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 
 			removeDependantBeans(flowId);
 
-			this.beanDefinitionRegistry.removeBeanDefinition(flowId);
+			this.beanFactory.destroySingleton(flowId);
 		}
 		else {
 			throw new IllegalStateException("An IntegrationFlow with the id "
@@ -203,12 +201,13 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 	private void removeDependantBeans(String parentName) {
 		String[] dependentBeans = this.beanFactory.getDependentBeans(parentName);
 		for (String beanName : dependentBeans) {
-			removeDependantBeans(beanName);
-			this.beanDefinitionRegistry.removeBeanDefinition(beanName);
-			String[] aliases = this.beanDefinitionRegistry.getAliases(beanName);
+			this.beanFactory.destroyBean(this.beanFactory.getBean(beanName));
+			this.beanFactory.destroySingleton(beanName);
+			String[] aliases = this.beanFactory.getAliases(beanName);
 			for (String alias : aliases) {
-				this.beanDefinitionRegistry.removeAlias(alias);
+				this.beanFactory.removeAlias(alias);
 			}
+			removeDependantBeans(beanName);
 		}
 	}
 
@@ -216,9 +215,8 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 	 * Obtain a {@link MessagingTemplate} with its default destination set to the input channel
 	 * of the {@link IntegrationFlow} for provided {@code flowId}.
 	 * <p> Any {@link IntegrationFlow} bean (not only manually registered) can be used for this method.
-	 * <p> If {@link IntegrationFlow} doesn't start with the
-	 * {@link org.springframework.messaging.MessageChannel}, the
-	 * {@link IllegalStateException} is thrown.
+	 * <p> If {@link IntegrationFlow} doesn't start with the {@link org.springframework.messaging.MessageChannel},
+	 * the {@link IllegalStateException} is thrown.
 	 * @param flowId the bean name to obtain the input channel from
 	 * @return the {@link MessagingTemplate} instance
 	 */
@@ -275,6 +273,9 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 		@Nullable
 		private Object source;
 
+		@Nullable
+		private String description;
+
 		StandardIntegrationFlowRegistrationBuilder(IntegrationFlow integrationFlow) {
 			this.integrationFlow = integrationFlow;
 		}
@@ -338,6 +339,12 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 			return this;
 		}
 
+		@Override
+		public IntegrationFlowRegistrationBuilder setDescription(String description) {
+			this.description = description;
+			return this;
+		}
+
 		@Override
 		public IntegrationFlowRegistrationBuilder useFlowIdAsPrefix() {
 			this.idAsPrefix = true;
@@ -364,4 +371,62 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 
 	}
 
+	private static final class IntegrationFlowComponentSourceAwareAdapter
+			implements IntegrationFlow, ComponentSourceAware {
+
+		private final IntegrationFlow delegate;
+
+		private Object beanSource;
+
+		private String beanDescription;
+
+		IntegrationFlowComponentSourceAwareAdapter(IntegrationFlow delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public void setComponentSource(Object source) {
+			this.beanSource = source;
+		}
+
+		@Override
+		public Object getComponentSource() {
+			return this.beanSource;
+		}
+
+		@Override
+		public void setComponentDescription(String description) {
+			this.beanDescription = description;
+		}
+
+		@Override
+		public String getComponentDescription() {
+			return this.beanDescription;
+		}
+
+		@Nullable
+		@Override
+		public String getBeanName() {
+			return null;
+		}
+
+		@Override
+		public void setBeanName(String name) {
+
+		}
+
+		@Override
+		public void configure(IntegrationFlowDefinition<?> flow) {
+			this.delegate.configure(flow);
+			StandardIntegrationFlow standardIntegrationFlow = ((IntegrationFlowBuilder) flow).get();
+			if (this.beanSource != null) {
+				standardIntegrationFlow.setComponentSource(this.beanSource);
+			}
+			if (this.beanDescription != null) {
+				standardIntegrationFlow.setComponentDescription(this.beanDescription);
+			}
+		}
+
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java

@@ -48,6 +48,7 @@ import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -59,6 +60,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
  * @since 5.1.3
  */
 @SpringJUnitConfig
+@DirtiesContext
 public class GatewayDslTests {
 
 	@Autowired
@@ -96,9 +98,8 @@ public class GatewayDslTests {
 		String exceptionMessage = ((Exception) receive.getPayload()).getMessage();
 		assertThat(exceptionMessage)
 				.contains("message has been rejected in filter")
-				.contains("defined in: " +
-						"'org.springframework.integration.dsl.gateway.GatewayDslTests$ContextConfiguration'; " +
-						"from source: 'bean method gatewayRequestFlow'");
+				.contains("from source: 'public org.springframework.integration.dsl.IntegrationFlow " +
+						"org.springframework.integration.dsl.gateway.GatewayDslTests$ContextConfiguration.gatewayRequestFlow()'");
 	}
 
 	@Autowired

spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java

@@ -21,8 +21,10 @@ import java.time.Instant;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -34,9 +36,7 @@ import java.util.function.Supplier;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
 
-import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.BeanCreationNotAllowedException;
-import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
@@ -52,7 +52,6 @@ import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableIntegrationManagement;
 import org.springframework.integration.config.EnableMessageHistory;
-import org.springframework.integration.config.IntegrationManagementConfigurer;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
@@ -112,9 +111,6 @@ public class ManualFlowTests {
 	@Autowired
 	private SmartLifecycleRoleController roleController;
 
-	@Autowired
-	private IntegrationManagementConfigurer integrationManagementConfigurer;
-
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testWithAnonymousMessageProducerStart() {
@@ -212,7 +208,6 @@ public class ManualFlowTests {
 				this.beanFactory.getBean(flowRegistrationId + BeanFactoryHandler.class.getName() + "#0",
 						BeanFactoryHandler.class);
 		assertThat(bean).isSameAs(additionalBean);
-		assertThat(bean.beanFactory).isSameAs(this.beanFactory);
 		bean = this.beanFactory.getBean(flowRegistrationId + "." + "anId.handler", BeanFactoryHandler.class);
 
 		MessagingTemplate messagingTemplate = flowRegistration.getMessagingTemplate();
@@ -486,23 +481,28 @@ public class ManualFlowTests {
 
 	@Test
 	public void testDisabledBeansOverride() {
-		assertThatExceptionOfType(BeanCreationException.class)
+		assertThatExceptionOfType(BeanDefinitionOverrideException.class)
 				.isThrownBy(() ->
 						this.integrationFlowContext.registration(f -> f.channel(c -> c.direct("doNotOverrideChannel")))
 								.register())
-				.isExactlyInstanceOf(BeanCreationException.class)
-				.withCauseExactlyInstanceOf(BeanDefinitionOverrideException.class)
 				.withMessageContaining("Invalid bean definition with name 'doNotOverrideChannel'");
 	}
 
 	@Test
 	public void testBeanDefinitionInfoInTheException() {
+		Map<?, ?> mergedBeanDefinitions =
+				TestUtils.getPropertyValue(this.beanFactory, "mergedBeanDefinitions", Map.class);
+
+		int mergedBeanDefinitionsSizeBeforeRandomFlow = mergedBeanDefinitions.size();
+
 		IntegrationFlow testFlow = f -> f.<String, String>transform(String::toUpperCase);
 		Method source = ReflectionUtils.findMethod(ManualFlowTests.class, "testBeanDefinitionInfoInTheException");
 		IntegrationFlowRegistration flowRegistration =
 				this.integrationFlowContext.registration(testFlow)
 						.setSource(source)
+						.id("flow#" + UUID.randomUUID())
 						.register();
+
 		assertThatExceptionOfType(MessageTransformationException.class)
 				.isThrownBy(() -> flowRegistration.getInputChannel().send(new GenericMessage<>(new Date())))
 				.withCauseExactlyInstanceOf(ClassCastException.class)
@@ -510,6 +510,8 @@ public class ManualFlowTests {
 				.withStackTraceContaining("java.util.Date cannot be cast to");
 
 		flowRegistration.destroy();
+
+		assertThat(mergedBeanDefinitions.size()).isEqualTo(mergedBeanDefinitionsSizeBeforeRandomFlow);
 	}
 
 	@Configuration
@@ -565,20 +567,17 @@ public class ManualFlowTests {
 
 	private final class BeanFactoryHandler extends AbstractReplyProducingMessageHandler {
 
-		@Autowired
-		private BeanFactory beanFactory;
-
 		private boolean destroyed;
 
 		@Override
 		protected Object handleRequestMessage(Message<?> requestMessage) {
-			Objects.requireNonNull(this.beanFactory);
+			Objects.requireNonNull(getBeanFactory());
 			return requestMessage;
 		}
 
 		@Override
 		protected void doInit() {
-			this.beanFactory.getClass(); // ensure wiring before afterPropertiesSet()
+			getBeanFactory().getClass(); // ensure wiring before afterPropertiesSet()
 		}
 
 		@Override