Функция Spring Cloud и обмен недоставленными письмами

Я пытаюсь написать реактивную службу Spring Cloud Function с использованием RabbitMQ, которая будет потреблять одну очередь и производить обмен.

У меня 2 вопроса.

1. Почему я получаю ошибку ниже в логах.
2. Как мне отклонить запрос с doOnError? DoOnError имеет доступ только к throwable, но не к сообщению об отклонении.

Вот код приложения. Он скопирован из этого вопроса Spring Reactive Stream - Unexpected Shutdown

@SpringBootApplication
public class StreamApplication {

  public static void main(String[] args) {
    SpringApplication.runStreamApplication args);
  }

  @Bean
  public Function<Flux<Message<String>>, Flux<Message<String>>> transform() {

    return inbound -> inbound.map(msg -> {
      try {
        System.out.println("ACKING MESSAGE");
        Channel channel = msg.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
        channel.basicAck(msg.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class), false);
      }
      catch (IOException e) {
        e.printStackTrace();
      }
      return msg;
    });
  }
}

Вот application.yml. Он имеет 2 разных связывателя для входящего из очереди и исходящего для обмена.

spring:
  cloud:
    stream:
      override-cloud-connectors: true
      rabbit:
        bindings:
          events-processor:
            producer:
              bindQueue: false
              declareExchange: false
              routing-key-expression: headers.eventType
          events:
            consumer:
              acknowledge-mode: MANUAL   
              prefetch: 10
              #auto-bind-dlq: false
              dead-letter-exchange: dead-letter-exchange
              bindQueue: false
              declareExchange: false
      function:
        definition: transform
        bindings:
          transform-in-0: events
          transform-out-0: events-processor
      bindings:
        events:
          destination: queue
          binder: consumerrabbit
          group: events-processor
        events-processor:
          destination: activity-events
          binder: producerrabbit
      binders:
        producerrabbit:
          defaultCandidate: false
          inheritEnvironment: false
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: host
                port: port
                username: username
                password: password
                virtual-host: virtual-host
        consumerrabbit:
          defaultCandidate: true
          inheritEnvironment: false
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: host
                port: port
                username: username
                password: password
                virtual-host: virtual-host

Вот журналы запуска и того, когда приложение получает событие. Я не уверен, почему во время запуска у канала есть подписчик, но как только он получает сообщение, он говорит, что у него нет подписчиков.

2020-09-07 12:02:04.076  INFO 10652 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.events' has 1 subscriber(s).
2020-09-07 18:02:08.848  INFO 10652 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.events-processor' has 1 subscriber(s).
2020-09-07 12:02:05.081  INFO 10652 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel events
2020-09-07 12:02:05.145  INFO 10652 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel errorChannel
2020-09-07 12:02:05.178  INFO 10652 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel events-processor
...................other startup log messages...............
2020-09-07 18:05:09.896  INFO 10652 --- [nts-processor-1] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.events' has 0 subscriber(s).
ACKING MESSAGE
2020-09-07 18:05:12.900 ERROR 10652 --- [nts-processor-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application-1.events'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload={
  "data": {
  
  }
}, headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedExchange=source, amqp_deliveryTag=1, deliveryAttempt=3, amqp_consumerQueue=source.events-processor, amqp_channel=Cached Rabbit Channel: AMQChannel(amqp:), conn: Proxy@43e869ea Shared Rabbit Connection: SimpleConnection@213451a4 [delegate=amqp:], amqp_redelivered=false, id=35cdfa7f-7f75-5502-aede-6ec9569145f0, amqp_consumerTag=amq.ctag-Z87p6nKN4PLJKDbZQpXnVw, sourceData=(Body:'[B@20dc391b(byte[401])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=NON_PERSISTENT, redelivered=false, receivedExchange=source, receivedRoutingKey=, deliveryTag=1, consumerTag=amq.ctag-Z87p6nKN4PLJKDbZQpXnVw, consumerQueue=source.events-processor]), contentType=application/json, timestamp=1599501912897}], failedMessage=GenericMessage [payload={
  "data": {
  
  }
}, headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedExchange=source, amqp_deliveryTag=1, deliveryAttempt=3, amqp_consumerQueue=source.events-processor, amqp_channel=Cached Rabbit Channel: AMQChannel(amqp://H), conn: Proxy@43e869ea Shared Rabbit Connection: SimpleConnection@213451a4 [delegate=amqp://], amqp_redelivered=false, id=35cdfa7f-7f75-5502-aede-6ec9569145f0, amqp_consumerTag=amq.ctag-Z87p6nKN4PLJKDbZQpXnVw, sourceData=(Body:'[B@20dc391b(byte[401])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=NON_PERSISTENT, redelivered=false, receivedExchange=source, receivedRoutingKey=, deliveryTag=1, consumerTag=amq.ctag-Z87p6nKN4PLJKDbZQpXnVw, consumerQueue=source.events-processor]), contentType=application/json, timestamp=1599501912897}]