Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9854: Special error for "too early message production"

Browse Source 
Fixes: https://github.com/spring-projects/spring-integration/issues/9854

The well-known `Dispatcher has no subscribers` is not very informative
when a message is produced from early application context initialization phase

* Add internal `ApplicationRunningController` bean to handle early `start()` event
* Check for this bean status from the `AbstractMessageChannel.send()`
* Throw specific `MessageDispatchingException` to indicate that the message was produced from a wrong place
* Adjust `ApplicationEventListeningMessageProducer` logic for `ContextStoppedEvent` & `ContextClosedEvent`
to indicate that `AbstractMessageChannel` bean might not dispatch a message because the application context is not running
This commit is contained in:
Artem Bilan
2025-02-20 17:45:20 -05:00
parent 84a3be1022
commit 2296e4798e
10 changed files with 213 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-2025 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.
@@ -34,10 +34,14 @@ import io.micrometer.observation.Observation;
import io.micrometer.observation.ObservationRegistry;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.core.OrderComparator;
import org.springframework.core.log.LogAccessor;
import org.springframework.integration.IntegrationPattern;
import org.springframework.integration.IntegrationPatternType;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.history.MessageHistory;
@@ -110,6 +114,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
private volatile String fullChannelName;
private volatile boolean applicationRunning;
private volatile Lifecycle applicationRunningController;
@Override
public String getComponentType() {
return "channel";
@@ -319,6 +327,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
public boolean send(Message<?> messageArg, long timeout) {
Assert.notNull(messageArg, "message must not be null");
Assert.notNull(messageArg.getPayload(), "message payload must not be null");
assertApplicationRunning(messageArg);
Message<?> message = messageArg;
if (this.shouldTrack) {
message = MessageHistory.write(message, this, getMessageBuilderFactory());
@@ -335,6 +344,39 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
}
private void assertApplicationRunning(Message<?> message) {
if (!this.applicationRunning) {
ApplicationContext applicationContext = getApplicationContext();
this.applicationRunning =
applicationContext == null ||
!applicationContext.containsBean(
IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME);
if (!this.applicationRunning) {
if (((ConfigurableApplicationContext) applicationContext).isActive()) {
this.applicationRunningController =
applicationContext.getBean(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME,
Lifecycle.class);
this.applicationRunning = this.applicationRunningController.isRunning();
}
}
}
if (this.applicationRunning && this.applicationRunningController != null) {
this.applicationRunning = this.applicationRunningController.isRunning();
}
if (!this.applicationRunning) {
throw new MessageDispatchingException(message,
"""
The application context is not ready to dispatch messages. \
It has to be refreshed or started first. \
Also, messages must not be emitted from initialization phase, \
like 'afterPropertiesSet()', '@PostConstruct' or bean definition methods. \
Consider to use 'SmartLifecycle.start()' instead.""");
}
}
private boolean sendWithObservation(Message<?> message, long timeout) {
MutableMessage<?> messageToSend = MutableMessage.of(message);
Observation observation = IntegrationObservation.PRODUCER.observation(
@@ -343,15 +385,15 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
() -> new MessageSenderContext(messageToSend, getComponentName()),
this.observationRegistry);
Boolean observe = observation.observe(() -> {
Message<?> messageToSendInternal = messageToSend;
if (message instanceof ErrorMessage errorMessage) {
messageToSendInternal =
new ErrorMessage(errorMessage.getPayload(),
messageToSend.getHeaders(),
errorMessage.getOriginalMessage());
}
return sendInternal(messageToSendInternal, timeout);
});
Message<?> messageToSendInternal = messageToSend;
if (message instanceof ErrorMessage errorMessage) {
messageToSendInternal =
new ErrorMessage(errorMessage.getPayload(),
messageToSend.getHeaders(),
errorMessage.getOriginalMessage());
}
return sendInternal(messageToSendInternal, timeout);
});
return Boolean.TRUE.equals(observe);
}

62
spring-integration-core/src/main/java/org/springframework/integration/config/ApplicationRunningController.java Normal file
View File

@@ -0,0 +1,62 @@
/*
* Copyright 2025 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.config;
import org.springframework.context.SmartLifecycle;
/**
* An infrastructure bean to hold the status of the application context when
* it is ready for interaction: refreshed or started.
* <p>
* Well-known {@link org.springframework.context.ConfigurableApplicationContext#isRunning()}
* (or {@link org.springframework.context.event.ContextRefreshedEvent})
* is good for target applications, when all the beans are already started,
* but most of Spring Integration channel adapters initiate their logic
* from the {@link SmartLifecycle#start()} implementation, so it would be false report
* that application is not running during start.
* <p>
* This implementation uses {@value Integer#MIN_VALUE} for its phase to be started as early as possible.
*
* @author Artem Bilan
*
* @since 6.5
*/
class ApplicationRunningController implements SmartLifecycle {
private volatile boolean running;
@Override
public void start() {
this.running = true;
}
@Override
public void stop() {
this.running = false;
}
@Override
public boolean isRunning() {
return this.running;
}
@Override
public int getPhase() {
return Integer.MIN_VALUE;
}
}

14
spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2024 the original author or authors.
* Copyright 2002-2025 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.
@@ -129,6 +129,7 @@ public class DefaultConfiguringBeanFactoryPostProcessor implements BeanDefinitio
registerListMessageHandlerMethodFactory();
registerIntegrationConfigurationReport();
registerControlBusCommandRegistry();
registerApplicationRunningController();
}
@Override
@@ -453,6 +454,17 @@ public class DefaultConfiguringBeanFactoryPostProcessor implements BeanDefinitio
}
}
private void registerApplicationRunningController() {
if (!this.beanFactory.containsBean(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME)) {
BeanDefinitionBuilder builder =
BeanDefinitionBuilder.genericBeanDefinition(ApplicationRunningController.class)
.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
this.registry.registerBeanDefinition(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME,
builder.getBeanDefinition());
}
}
private static BeanDefinitionBuilder createMessageHandlerMethodFactoryBeanDefinition(boolean listCapable) {
return BeanDefinitionBuilder.genericBeanDefinition(IntegrationMessageHandlerMethodFactory.class)
.addConstructorArgValue(listCapable)

4
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2024 the original author or authors.
* Copyright 2002-2025 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.
@@ -102,6 +102,8 @@ public abstract class IntegrationContextUtils {
public static final String CONTROL_BUS_COMMAND_REGISTRY_BEAN_NAME = "controlBusCommandRegistry";
public static final String APPLICATION_RUNNING_CONTROLLER_BEAN_NAME = "applicationRunningController";
/**
* The default timeout for blocking operations like send and receive messages.
* @since 6.1

2
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -334,7 +334,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint
return ErrorMessageUtils.getAttributeAccessor(message, null);
}
private MessageChannel getRequiredOutputChannel() {
protected MessageChannel getRequiredOutputChannel() {
MessageChannel messageChannel = getOutputChannel();
Assert.state(messageChannel != null, "The 'outputChannel' or `outputChannelName` must be configured");
return messageChannel;

24
spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2024 the original author or authors.
* Copyright 2002-2025 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,10 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.IntegrationContextUtils;
@@ -43,8 +45,10 @@ import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.ClassUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.mock;
/**
@@ -67,14 +71,21 @@ public class ApplicationContextMessageBusTests {
}
@Test
public void endpointRegistrationWithInputChannelReference() {
public void endpointRegistrationWithInputChannelReference() throws ClassNotFoundException {
this.context.registerBean(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME,
BeanUtils.instantiateClass(
ClassUtils.forName(
IntegrationContextUtils.BASE_PACKAGE + ".config.ApplicationRunningController", null)));
QueueChannel sourceChannel = new QueueChannel();
sourceChannel.setApplicationContext(this.context);
QueueChannel targetChannel = new QueueChannel();
this.context.registerChannel("sourceChannel", sourceChannel);
this.context.registerChannel("targetChannel", targetChannel);
Message<String> message = MessageBuilder.withPayload("test")
.setReplyChannelName("targetChannel").build();
sourceChannel.send(message);
assertThatExceptionOfType(MessageDispatchingException.class)
.isThrownBy(() -> sourceChannel.send(message))
.withMessageStartingWith("The application context is not ready to dispatch messages.");
AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
@Override
@@ -88,8 +99,15 @@ public class ApplicationContextMessageBusTests {
endpoint.setBeanFactory(mock(BeanFactory.class));
this.context.registerEndpoint("testEndpoint", endpoint);
this.context.refresh();
sourceChannel.send(message);
Message<?> result = targetChannel.receive(10000);
assertThat(result.getPayload()).isEqualTo("test");
this.context.stop();
assertThatExceptionOfType(MessageDispatchingException.class)
.isThrownBy(() -> sourceChannel.send(message))
.withMessageStartingWith("The application context is not ready to dispatch messages.");
}
@Test

34
spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2024 the original author or authors.
* Copyright 2002-2025 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.
@@ -26,8 +26,8 @@ import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.ContextStoppedEvent;
import org.springframework.context.event.GenericApplicationListener;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.core.ResolvableType;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -49,10 +49,10 @@ import org.springframework.util.Assert;
public class ApplicationEventListeningMessageProducer extends ExpressionMessageProducerSupport
implements GenericApplicationListener {
private volatile Set<ResolvableType> eventTypes;
private ApplicationEventMulticaster applicationEventMulticaster;
private volatile Set<ResolvableType> eventTypes;
private volatile long stoppedAt;
/**
@@ -106,12 +106,17 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
@Override
public void onApplicationEvent(ApplicationEvent event) {
if (isActive() || ((event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent)
&& stoppedRecently())) {
boolean contextFinished = event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent;
if (isActive() || (contextFinished && stoppedRecently())) {
Object source = event.getSource();
if (source instanceof Message<?>) {
sendMessage((Message<?>) source);
if (contextFinished && getRequiredOutputChannel() instanceof AbstractMessageChannel) {
logger.warn("Messages for 'ContextStoppedEvent' or 'ContextClosedEvent' cannot be dispatched " +
"via 'AbstractMessageChannel' beans: the application context is in the finished state." +
"Consider to use custom 'MessageChannel' implementation without dispatching logic.");
}
if (event.getSource() instanceof Message<?> message) {
sendMessage(message);
}
else {
Message<?> message;
@@ -128,8 +133,8 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
}
private Object extractObjectToSend(Object root) {
if (root instanceof PayloadApplicationEvent) {
return ((PayloadApplicationEvent<?>) root).getPayload();
if (root instanceof PayloadApplicationEvent<?> payloadApplicationEvent) {
return payloadApplicationEvent.getPayload();
}
return evaluatePayloadExpression(root);
}
@@ -167,14 +172,9 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
return false;
}
@Override
public boolean supportsSourceType(Class<?> sourceType) {
return true;
}
@Override
public int getOrder() {
return Ordered.LOWEST_PRECEDENCE;
return HIGHEST_PRECEDENCE;
}
@Override

44
spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2024 the original author or authors.
* Copyright 2002-2025 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.
@@ -18,6 +18,9 @@ package org.springframework.integration.event.inbound;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
@@ -341,13 +344,7 @@ public class ApplicationEventListeningMessageProducerTests {
}
private static class TestApplicationListener implements ApplicationListener<ApplicationEvent> {
private final AtomicInteger counter;
private TestApplicationListener(AtomicInteger counter) {
this.counter = counter;
}
private record TestApplicationListener(AtomicInteger counter) implements ApplicationListener<ApplicationEvent> {
@Override
public void onApplicationEvent(ApplicationEvent event) {
@@ -356,4 +353,35 @@ public class ApplicationEventListeningMessageProducerTests {
}
static final class ContextEventsChannel implements PollableChannel {
private final BlockingQueue<Message<?>> internalQueue = new LinkedBlockingQueue<>();
@Override
public Message<?> receive() {
return receive(-1);
}
@Override
public Message<?> receive(long timeout) {
try {
return this.internalQueue.poll(timeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
@Override
public boolean send(Message<?> message, long timeout) {
try {
return this.internalQueue.offer(message, timeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
}

11
spring-integration-event/src/test/java/org/springframework/integration/event/inbound/applicationEventInboundChannelAdapterTests.xml
View File

@@ -1,13 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd"
xmlns:int="http://www.springframework.org/schema/integration">
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
<int:channel id="channel">
<int:queue capacity="5"/>
</int:channel>
<bean id="channel"
class="org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducerTests$ContextEventsChannel"/>
<bean id="adapter" class="org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer">
<property name="outputChannel" ref="channel"/>

4
src/reference/antora/modules/ROOT/pages/whats-new.adoc
View File

@@ -27,6 +27,10 @@ The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentExcept
Instead, such a collection is wrapped into a single reply message.
See xref:aggregator.adoc[Aggregator] for more information.
The `AbstractMessageChannel` beans now throw a special `MessageDispatchingException` when an attempt to send a message to not running application is done.
In general, it is a design error to try to produce a message from `afterPropertiesSet()`, `@PostConstruct` or bean definition methods.
The `SmartLifecycle.start()` is preferred way for this kind of logic, or better to do that via inbound channel adapters.
[[x6.5-correlation-changes]]
== The `discardIndividuallyOnExpiry` Option For Correlation Handlers