30 Second Timeout breaks Coffee Java DSL Sample Application

[swiss-chris]

In what version(s) of Spring Integration are you seeing this issue?

6.4.0

Describe the bug

Running the org.springframework.integration.samples.dsl.cafe.lambda.Application looks fine for the first 30 seconds and then prints this

...
scheduling-1 prepared cold drink #25 for order #25: iced 3 shot MOCHA
scheduling-1 prepared cold drink #26 for order #26: iced 3 shot MOCHA
scheduling-1 prepared cold drink #27 for order #27: iced 3 shot MOCHA
scheduling-1 prepared cold drink #28 for order #28: iced 3 shot MOCHA
scheduling-1 prepared cold drink #29 for order #29: iced 3 shot MOCHA
23:33:33.222 [pool-2-thread-79] ERROR o.s.i.handler.LoggingHandler - org.springframework.messaging.MessageDeliveryException: Failed to send message to channel 'bean 'orders.subFlow#0.channel#1'; defined in: 'org.springframework.integration.samples.dsl.cafe.lambda.Application'; from source: 'bean method orders'' within timeout: 30000, failedMessage=GenericMessage [payload=iced 3 shot MOCHA, headers={replyChannel=nullChannel, sequenceNumber=2, errorChannel=, sequenceSize=2, correlationId=8e1a32b3-d0a5-c92a-8d70-06247011ed26, id=79cc6c1d-14fe-3185-6c84-9946a2b6e7de, timestamp=1720819983215}]

To Reproduce

Run the app

Expected behavior

No Exception

Sample

https://github.com/spring-projects/spring-integration-samples/blob/main/dsl/cafe-dsl/src/main/java/org/springframework/integration/samples/dsl/cafe/lambda/Application.java

[swiss-chris]

I tried setting spring.integration.endpoints.defaultTimeout=-1 inside META-INF/spring.integration.properties, but no luck (see also https://stackoverflow.com/q/78742274/349169)

[artembilan]

So, as Andy pointed out in other issue, the workaround is to use this bean:

    @Bean
    static BeanPostProcessor defaultEndpointTimeout() {
        return new BeanPostProcessor() {
            @Override
            public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
                if (IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME.equals(beanName)) {
                    ((IntegrationProperties) bean).setEndpointsDefaultTimeout(-1);
                }
                return bean;
            }
        };
    }

You also need to pay attention that by default Spring Boot provides for us a TaskScheduler with one thread in its pool: https://docs.spring.io/spring-boot/reference/features/task-execution-and-scheduling.html.

So, you might need to consider to change that pool size to 10, for example. This way scheduled tasks for polling endpoints are not going to wait for each other when they are part of the same flow. That's exactly why we have introduced that timeout: to not block the waiting thread forever. But apparently when we pile message into QueueChannel, those default 30 seconds are not enough to handle workload. Therefore we fail with the mentioned error from time to time.

I can provide the fix for the mention sample. But is that a really goal of your request?

[swiss-chris]

Hi Artem Thanks for the investigation. Concerning this issue, I assumed the samples should continue to run without errors as a proof of concept or blueprint how Spring Integration DSL can be used. Indeed our Flows are somewhat loosely based on the Cafe DSL Example in that they also start long processes (longer than 30 Seconds) within a single/main flow (rather than a sequence of separate flows passed through explicit channels). So it would be helpful for me to see in the Cafe DSL Example how we need to configure a Spring Integration Flow to work under these conditions and not run into timeouts. I thought the global defaultTimeout setting might also be a useful solution to use here.