Spring Cloud Stream Kotlin Consumer Problem

Question

Spring Cloud Stream Kotlin Consumer Problem

I'm trying to use Spring Cloud Stream from Kotlin. I wrote a simple consumer as follows:

@Bean
fun log(): Consumer<Person> {
    return Consumer<Person> { person: Person -> println("Received: ${person.name}") }
}

Spring Cloud Stream does not find this consumer and quits immediately. It doesn't give an error, but just terminates.

If I write the very same consumer in Java as in the following, Stream Cloud Stream finds this consumer and it consumes input.

@Bean
public Consumer<Person> log() {
    return person -> {
        System.out.println("Received: " + person.getName());
    };
}

Why cannot Spring Cloud Stream find my consumer in Kotlin?

Here's the Spring Boot Console output:

2020-05-26 08:01:29.167  INFO 78463 --- [           main] c.a.s.t.TestBinderApplicationKt          : Starting TestBinderApplicationKt on mburbidg-macOS with PID 78463 (/Users/mburbidg/test-binder/build/classes/kotlin/main started by mburbidg in /Users/mburbidg/test-binder)
2020-05-26 08:01:29.181  INFO 78463 --- [           main] c.a.s.t.TestBinderApplicationKt          : No active profile set, falling back to default profiles: default
2020-05-26 08:01:30.109  INFO 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
2020-05-26 08:01:30.113  INFO 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'taskScheduler' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created.
2020-05-26 08:01:30.114 DEBUG 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : The '#jsonPath' SpEL function cannot be registered: there is no jayway json-path.jar on the classpath.
2020-05-26 08:01:30.114 DEBUG 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : SpEL function '#xpath' isn't registered: there is no spring-integration-xml.jar on the classpath.
2020-05-26 08:01:30.116  INFO 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
2020-05-26 08:01:30.145  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.147  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration$IntegrationJmxConfiguration' of type [org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration$IntegrationJmxConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.153  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration' of type [org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.156  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'mbeanServer' of type [com.sun.jmx.mbeanserver.JmxMBeanServer] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.166  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.167  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationDisposableAutoCreatedBeans' of type [org.springframework.integration.config.annotation.Disposables] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.468  INFO 78463 --- [           main] o.s.s.c.ThreadPoolTaskScheduler          : Initializing ExecutorService 'taskScheduler'
2020-05-26 08:01:30.484  INFO 78463 --- [           main] c.f.c.c.BeanFactoryAwareFunctionRegistry : Looking up function '' with acceptedOutputTypes: []
2020-05-26 08:01:30.601 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering beans for JMX exposure on startup
2020-05-26 08:01:30.602  INFO 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel errorChannel
2020-05-26 08:01:30.603 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Located managed bean 'org.springframework.integration:type=MessageChannel,name=errorChannel': registering with JMX server as MBean [org.springframework.integration:type=MessageChannel,name=errorChannel]
2020-05-26 08:01:30.660  INFO 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel nullChannel
2020-05-26 08:01:30.660 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Located managed bean 'org.springframework.integration:type=MessageChannel,name=nullChannel': registering with JMX server as MBean [org.springframework.integration:type=MessageChannel,name=nullChannel]
2020-05-26 08:01:30.678  INFO 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageHandler errorLogger
2020-05-26 08:01:30.679 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Located managed bean 'org.springframework.integration:type=MessageHandler,name=errorLogger,bean=internal': registering with JMX server as MBean [org.springframework.integration:type=MessageHandler,name=errorLogger,bean=internal]
2020-05-26 08:01:30.700 DEBUG 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : 
Spring Integration global properties:

spring.integration.channels.autoCreate=true
spring.integration.taskScheduler.poolSize=10
spring.integration.readOnly.headers=
spring.integration.channels.maxUnicastSubscribers=0x7fffffff
spring.integration.endpoints.noAutoStartup=
spring.integration.messagingTemplate.throwExceptionOnLateReply=false
spring.integration.channels.maxBroadcastSubscribers=0x7fffffff

2020-05-26 08:01:30.701 DEBUG 78463 --- [           main] .s.i.c.GlobalChannelInterceptorProcessor : No global channel interceptors.
2020-05-26 08:01:30.703  INFO 78463 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2020-05-26 08:01:30.703  INFO 78463 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application.errorChannel' has 1 subscriber(s).
2020-05-26 08:01:30.703  INFO 78463 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
2020-05-26 08:01:30.716  INFO 78463 --- [           main] c.a.s.t.TestBinderApplicationKt          : Started TestBinderApplicationKt in 2.32 seconds (JVM running for 3.12)
2020-05-26 08:01:30.722  INFO 78463 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : Removing {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2020-05-26 08:01:30.723  INFO 78463 --- [extShutdownHook] o.s.i.channel.PublishSubscribeChannel    : Channel 'application.errorChannel' has 0 subscriber(s).
2020-05-26 08:01:30.723  INFO 78463 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : stopped bean '_org.springframework.integration.errorLogger'
2020-05-26 08:01:30.724  INFO 78463 --- [extShutdownHook] o.s.s.c.ThreadPoolTaskScheduler          : Shutting down ExecutorService 'taskScheduler'
2020-05-26 08:01:30.726 DEBUG 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Unregistering JMX-exposed beans on shutdown
2020-05-26 08:01:30.726 DEBUG 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Unregistering JMX-exposed beans
2020-05-26 08:01:30.727  INFO 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Summary on shutdown: bean 'errorChannel'
2020-05-26 08:01:30.727  INFO 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Summary on shutdown: nullChannel
2020-05-26 08:01:30.727  INFO 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Summary on shutdown: bean '_org.springframework.integration.errorLogger.handler' for component '_org.springframework.integration.errorLogger'

Process finished with exit code 0

kotlin

spring-cloud-stream

consumer

asked on Stack Overflow May 26, 2020 by

Michael Burbidge

2 Answers

I just tested it with this; and had no problems...

@SpringBootApplication
open class So62025232Application {
    @Bean
    open fun input(): Consumer<String> {
        return Consumer { x: String? -> println(x) }
    }

    @Bean
    open fun runner(template: RabbitTemplate): ApplicationRunner {
        return ApplicationRunner { args: ApplicationArguments? ->
            template.convertAndSend("input-in-0.foo", "foo")
            Thread.sleep(2000)
        }
    }

    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            val ctx = SpringApplication.run(So62025232Application::class.java, *args)
            println(ctx.getBeansOfType(Consumer::class.java))
            ctx.close()
        }
    }
}

spring.cloud.stream.bindings.input-in-0.group=foo

foo
{input=com.example.demo.So62025232Application$input$1@2a65bb85}

Java version

@SpringBootApplication
public class So62025232Application {

    public static void main(String[] args) {
        ConfigurableApplicationContext ctx = SpringApplication.run(So62025232Application.class, args);
        System.out.println(ctx.getBeansOfType(Consumer.class));
        ctx.close();
    }

    @Bean
    Consumer<String> input() {
        return System.out::println;
    }

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> {
            template.convertAndSend("input-in-0.foo", "foo");
            Thread.sleep(2000);
        };
    }

}

answered on Stack Overflow May 26, 2020 by

Gary Russell

Thanks! Your answer helped me figure out the problem. My app did look like this:

@SpringBootApplication
open class TestBinderApplication

fun main(args: Array<String>) {
    runApplication<TestBinderApplication>(*args)
}

@Bean
fun log(): Consumer<Person> {
    return Consumer<Person> { person: Person -> println("Received: ${person.name}") }

When I changed it to:

@SpringBootApplication
open class TestBinderApplication {
    @Bean
    open fun log(): Consumer<Person> {
        return Consumer<Person> { person: Person -> println("Received: ${person.name}") }
    }

    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            runApplication<TestBinderApplication>(*args)
        }
    }
}

It worked!

I'm new to Kotlin, so I'm not sure what the difference is.

answered on Stack Overflow May 26, 2020 by

Michael Burbidge

User contributions licensed under CC BY-SA 3.0