Unhandled converter exception ends in stderr instead of log [INT-4552]

[spring-operator]

Alen Turkovic opened INT-4552 and commented

I am not sure if this is a UdpSyslogReceivingChannelAdapter or UnicastReceivingChannelAdapter specific problem, but in case MessageConverter throws an exception, the exception message ends in stderr instead of the log.

Here is a sample application to reproduce:

 

package com.alturkovic.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.integration.syslog.MessageConverter;
import org.springframework.integration.syslog.inbound.UdpSyslogReceivingChannelAdapter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@SpringBootApplication
public class UnhandledExceptionApplication {

 public static void main(final String[] args) {
   SpringApplication.run(UnhandledExceptionApplication.class, args);
 }

 @Bean
 public TaskExecutor executor() {
   final ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
   threadPoolTaskExecutor.setCorePoolSize(5);
   return threadPoolTaskExecutor;
 }

 @Bean
 public UdpSyslogReceivingChannelAdapter udpReceiver() {
   final UdpSyslogReceivingChannelAdapter adapter = new UdpSyslogReceivingChannelAdapter();
   adapter.setUdpAdapter(receiver());
   adapter.setOutputChannelName("loggingChannel");
   adapter.setConverter(alwaysFailingConverter());
   return adapter;
 }


 @Bean
 public UnicastReceivingChannelAdapter receiver() {
   final UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(1997);
   adapter.setTaskExecutor(executor());
   return adapter;
 }

 @Bean
 public IntegrationFlow loggingFlow() {
   return IntegrationFlows.from("loggingChannel")
     .log()
     .get();
 }

 @Bean
 public MessageConverter alwaysFailingConverter() {
   return syslog -> {
     throw new IllegalStateException();
   };
 }
}

If I send a packet to port 1997, IllegalStateException ends up in stderr.

Exception in thread "executor-2" org.springframework.messaging.MessagingException: Failed to send Message; nested exception is java.lang.IllegalStateException, failedMessage=GenericMessage [payload=byte[16], headers={ip_packetAddress=10.10.20.120/10.10.20.120:53470, ip_address=10.10.20.120, id=0e03d0a8-241c-19e3-f5ad-543baa707af1, ip_port=53470, ip_hostname=10.10.20.120, timestamp=1542113178100}]
 at org.springframework.integration.syslog.inbound.SyslogReceivingChannelAdapterSupport.convertAndSend(SyslogReceivingChannelAdapterSupport.java:97)
 at org.springframework.integration.syslog.inbound.UdpSyslogReceivingChannelAdapter.lambda$onInit$0(UdpSyslogReceivingChannelAdapter.java:68)
 at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:70)
 at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:64)
 at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
 at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
 at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
 at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
 at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203)
 at org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter.doSend(UnicastReceivingChannelAdapter.java:206)
 at org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter.lambda$asyncSendMessage$0(UnicastReceivingChannelAdapter.java:179)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalStateException
 at com.alturkovic.example.StreamConfigurationEngineExampleApplication.lambda$alwaysFailingConverter$0(UnhandledExceptionApplication.java:55)
 at org.springframework.integration.syslog.inbound.SyslogReceivingChannelAdapterSupport.convertAndSend(SyslogReceivingChannelAdapterSupport.java:93)
 ... 13 more

 

---

Referenced from: pull request #2627

Backported to: 5.0.10

[spring-operator]

Gary Russell commented

Because of the nature of UDP, we perform an async send so we can get back to receive more data at the earliest opportunity. We can't rely on the user configuring an ExecutorChannel.

We need a try/catch/log in the Runnable lambda.

try {
	taskExecutor.execute(() -> doSend(packet));
}
catch (RejectedExecutionException e) {
	if (logger.isDebugEnabled()) {
		logger.debug("Adapter stopped, sending on main thread");
	}
	doSend(packet);
}