NullPointerException occurs when applying functionRouting in Spring Cloud Stream

Question

I want to use Apache Kafka with Spring Cloud Stream(3.0) in my project. In general, when supplier and consumer are registered as @bean, Producing and Conuming is fine without any problem, but if only functionRouter is applied as below, NullPointerException occurs on the consumer side. What on earth is the cause? Wouldn't serialization be working properly?

supplier:

@Bean
fun upperCase(): Supplier<Message<MessageDto>?> = Supplier { createEvent() }

private fun createEvent(): Message<MessageDto>? {
    return if (supplierEnabled)
        MessageBuilder.withPayload(MessageDto(1, "I send the message!"))
            .setHeader("to_process", true)
            .build()
    else
        null
}

consumer:

@Bean
fun first(): Consumer<MessageDto> = Consumer {
    log.info("Received(process): {}", it)
}

@Bean
fun second(): Consumer<MessageDto> = Consumer {
    log.info("Received(handle): {}", it)
}

producer application.yml:

spring:
  application:
    name: producer
  cloud:
    stream:
      bindings:
        upperCase-out-0:
          destination: memberChangeTopic
          content-type: application/json
        producer:
          partitionKeyExpression: payload.id
          partitionCount: 10
      kafka:
        binder:
          brokers: localhost
          defaultBrokerPort: 9092

consumer appllication.yml:

server:
  port: 8081
spring:
  application:
    name: consumer
  cloud:
    stream:
      bindings:
        functionRouter-in-0:
          destination: memberChangeTopic
          group: consumer-a
      function:
        routing:
          enabled: true
    function:
      routing-expression: (headers['to_process']!=null && headers['to_process']==true) ? 'first':'second'

error trace:

2021-02-05 09:40:44.947 ERROR 98560 --- [container-0-C-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@18545e81]; nested exception is java.lang.NullPointerException, failedMessage=GenericMessage [payload=byte[47], headers={kafka_offset=24, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1d55aff4, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, to_process=true, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=memberChangeTopic, kafka_receivedTimestamp=1612485254857, kafka_groupId=consumer-a, target-protocol=kafka}]

question from:https://stackoverflow.com/questions/66056133/nullpointerexception-occurs-when-applying-functionrouting-in-spring-cloud-stream

Answer 1

Waitting for answers