title: Spring cloud stream + After a producer error subscription to the IN channel breaks
body: <p>We are using spring-cloud-stream-binder-kafka 3.1.1 version.
  Looking to handle intermittent kafka publishing failures.</p>
  <p>Below is our existing configuration,</p>
  <pre><code>spring:
    cloud:
      function:
        definition: execute;supplyPayment;
        routing-expression:
      stream:
        kafka:
          default:
            producer:
              errorChannelEnabled: true
          binder:
            brokers: localhost:9094
            producer-properties:
              retries: 20
              max.in.flight.requests.per.connection: 10
              linger.ms: 100
              request.timeout.ms: 3000
              batch.size: 20000
              retry.backoff.ms: 1000
        bindings:
          execute-in-0:
            destination: test-in-source
            group: consumer-dedupe
            concurrency: 2
          execute-out-0:
            destination: test-out-topic
            producer:
              error-channel-enabled: true
  </code></pre>
  <p>Able to get hold of the producer error using ServiceActivator.</p>
  <pre><code>    @ServiceActivator(inputChannel = &quot;errorChannel&quot;)
      public void handle(final ErrorMessage em) {
          log.error(&quot;Error caught&quot; + em.toString());
      }
  </code></pre>
  <p>Was expecting no exception to be thrown as we have the ServiceActivator implemented, but we see TimeoutException thrown</p>
  <pre><code>org.apache.kafka.common.errors.TimeoutException: Topic test-out-topic not present in metadata after 60000 ms.
  2021-12-09|06:11:25.267 ERROR o.s.c.s.f.FunctionConfiguration$FunctionToDestinationBinder - intuit_tid= provider_id= entity_type= Failure was detected during execution of the reactive function 'execute'
  org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$ProducerConfigurationMessageHandler@48ea8766]; nested exception is org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.TimeoutException: Topic test-out-topic not present in metadata after 60000 ms., failedMessage=GenericMessage [payload=byte[1046], headers={intuit_originatingip=192.19.0.1, entity_version=v1, entity_processing_status=SUCCESS, PIPELINE_ID=1, target-protocol=kafka, ROUTING_INFO=RoutingInfo(currentPipelineNode=PipelineNode{id=0, type=null, sourceTopic=null, pipelineNodeStatus=null, errorHandlerClassName=null, errorTopicName=null}, executedPipelineNodeList=[PipelineNode{id=1, type=PROCESSOR, sourceTopic=idx-payment-source, pipelineNodeStatus=SUCCESS, errorHandlerClassName=com.intuit.fdp.enrichment.messagemesh.errorhandler.impl.ContinueProcessingErrorHandler, errorTopicName=null}, PipelineNode{id=2, type=PROCESSOR, sourceTopic=idx-payment-source, pipelineNodeStatus=SUCCESS, errorHandlerClassName=com.intuit.fdp.enrichment.messagemesh.errorhandler.impl.ContinueProcessingErrorHandler, errorTopicName=null}]), id=d9be7b62-711d-2200-2da7-775bc9ed4bbd, contentType=application/json, kafka_receivedTimestamp=1639059024747, intuit_realmid=123148525896109, intuit_iddomain=test, timestamp=1639059025243, test=testValue, intuit_tid=978bd234-48cc-4914-8afc-9c84be5c467e, intuit_offeringid=Intuit.platform.qbo.dtx.ui, deliveryAttempt=1, kafka_timestampType=CREATE_TIME, intuit_locale=en_US, kafka_receivedTopic=idx-payment-source, kafka_offset=2, entity_type=PaymentTransactionEvent, entity_namespace=com.intuit.idx.event.payment.v1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@597cf885, intuit_country=US, provider_id=1234, intuit_corpid=Jacinda, kafka_receivedPartitionId=0, kafka_groupId=consumer-dedupe}]
      at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:192)
      Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
  Assembly trace from producer [reactor.core.publisher.FluxPeekFuseable] :
      reactor.core.publisher.Flux.doOnNext
      org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder.lambda$bindFunctionToDestinations$9(FunctionConfiguration.java:501)
  Error has been observed at the following site(s):
      |_ Flux.doOnNext ⇢ at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder.lambda$bindFunctionToDestinations$9(FunctionConfiguration.java:501)
  Stack trace:
          at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:192)
          at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:65)
          at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder$SendingHandler.handleMessageInternal(AbstractMessageChannelBinder.java:1041)
          at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:56)
          at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
          at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
          at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
          at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
          at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317)
          at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272)
          at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder.lambda$null$7(FunctionConfiguration.java:518)
          at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:196)
          at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:210)
          at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:127)
          at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:210)
          at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:127)
          at reactor.core.publisher.FluxFlatMap$FlatMapMain.tryEmit(FluxFlatMap.java:542)
          at reactor.core.publisher.FluxFlatMap$FlatMapInner.onNext(FluxFlatMap.java:1006)
          at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
          at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:180)
          at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
          at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1815)
          at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249)
          at reactor.core.publisher.MonoMetricsFuseable$MetricsFuseableSubscriber.onNext(MonoMetricsFuseable.java:132)
          at 
  ---
  Caused by: org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.TimeoutException: Topic test-out-topic not present in metadata after 60000 ms.
  </code></pre>
  <p>And also the errorHandler is logging the producer error,</p>
  <pre><code>ErrorMessage [payload=org.springframework.integration.kafka.support.KafkaSendFailureException: nested exception is org.springframework.kafka.core.KafkaProducerException: Failed to send; nested exception is org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for test-out-topic-0:120000 ms has passed since batch creation, failedMessage=GenericMessage [payload=byte[1044], headers={intuit_originatingip=192.19.0.1, entity_version=v1, entity_processing_status=SUCCESS, PIPELINE_ID=1, target-protocol=kafka, ROUTING_INFO=RoutingInfo(currentPipelineNode=PipelineNode{id=0, type=null, sourceTopic=null, pipelineNodeStatus=null, errorHandlerClassName=null, errorTopicName=null}, executedPipelineNodeList=[PipelineNode{id=1, type=PROCESSOR, sourceTopic=idx-payment-source, pipelineNodeStatus=SUCCESS, errorHandlerClassName=com.intuit.fdp.enrichment.messagemesh.errorhandler.impl.ContinueProcessingErrorHandler, errorTopicName=null}, PipelineNode{id=2, type=PROCESSOR, sourceTopic=idx-payment-source, pipelineNodeStatus=SUCCESS, errorHandlerClassName=com.intuit.fdp.enrichment.messagemesh.errorhandler.impl.ContinueProcessingErrorHandler, errorTopicName=null}]), id=5ea18c46-1074-41bd-8ac6-adfc1aba7811, contentType=application/json, kafka_receivedTimestamp=1639058994738, intuit_realmid=123148525896109, intuit_iddomain=test, timestamp=1639058995246, test=testValue, intuit_tid=0a19feb2-e90c-4a3d-b9ef-94294bbd5ebd, intuit_offeringid=Intuit.platform.qbo.dtx.ui, deliveryAttempt=1, kafka_timestampType=CREATE_TIME, intuit_locale=en_US, kafka_receivedTopic=idx-payment-source, kafka_offset=1, entity_type=PaymentTransactionEvent, entity_namespace=com.intuit.idx.event.payment.v1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@597cf885, intuit_country=US, provider_id=1234, intuit_corpid=Angela, kafka_receivedPartitionId=0, kafka_groupId=consumer-dedupe}] [record=ProducerRecord(topic=test-out-topic, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = intuit_originatingip, value = [49, 57, 50, 46, 49, 57, 46, 48, 46, 49]), RecordHeader(key = entity_version, value = [118, 49]), RecordHeader(key = entity_processing_status, value = [83, 85, 67, 67, 69, 83, 83]), RecordHeader(key = PIPELINE_ID, value = [49]), RecordHeader(key = target-protocol, value = [107, 97, 102, 107, 97]), RecordHeader(key = ROUTING_INFO, value = [123, 34, 99, 117, 114, 114, 101, 110, 116, 80, 105, 112, 101, 108, 105, 110, 101, 78, 111, 100, 101, 34, 58, 123, 34, 105, 100, 34, 58, 48, 44, 34, 103, 114, 111, 117, 112, 73, 100, 34, 58, 110, 117, 108, 108, 44, 34, 97, 114, 116, 105, 102, 97, 99, 116, 73, 100, 34, 58, 110, 117, 108, 108, 44, 34, 118, 101, 114, 115, 105, 111, 110, 34, 58, 110, 117, 108, 108, 44, 34, 116, 121, 112, 101, 34, 58, 110, 117, 108, 108, 44, 34, 114, 117, 110, 116, 105, 109, 101, 73, 100, 34, 58, 110, 117, 108, 108, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 115, 34, 58, 91, 93, 44, 34, 115, 111, 117, 114, 99, 101, 84, 111, 112, 105, 99, 34, 58, 110, 117, 108, 108, 44, 34, 112, 105, 112, 101, 108, 105, 110, 101, 78, 111, 100, 101, 83, 116, 97, 116, 117, 115, 34, 58, 110, 117, 108, 108, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 67, 108, 97, 115, 115, 78, 97, 109, 101, 34, 58, 110, 117, 108, 108, 44, 34, 101, 114, 114, 111, 114, 84, 111, 112, 105, 99, 78, 97, 109, 101, 34, 58, 110, 117, 108, 108, 125, 44, 34, 101, 120, 101, 99, 117, 116, 101, 100, 80, 105, 112, 101, 108, 105, 110, 101, 78, 111, 100, 101, 76, 105, 115, 116, 34, 58, 91, 123, 34, 105, 100, 34, 58, 49, 44, 34, 103, 114, 111, 117, 112, 73, 100, 34, 58, 34, 99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 109, 101, 115, 115, 97, 103, 101, 109, 101, 115, 104, 34, 44, 34, 97, 114, 116, 105, 102, 97, 99, 116, 73, 100, 34, 58, 34, 100, 101, 100, 117, 112, 101, 34, 44, 34, 118, 101, 114, 115, 105, 111, 110, 34, 58, 34, 48, 46, 49, 46, 49, 57, 34, 44, 34, 116, 121, 112, 101, 34, 58, 34, 80, 82, 79, 67, 69, 83, 83, 79, 82, 34, 44, 34, 114, 117, 110, 116, 105, 109, 101, 73, 100, 34, 58, 34, 105, 100, 120, 45, 100, 101, 100, 117, 112, 101, 34, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 115, 34, 58, 91, 93, 44, 34, 115, 111, 117, 114, 99, 101, 84, 111, 112, 105, 99, 34, 58, 34, 105, 100, 120, 45, 112, 97, 121, 109, 101, 110, 116, 45, 115, 111, 117, 114, 99, 101, 34, 44, 34, 112, 105, 112, 101, 108, 105, 110, 101, 78, 111, 100, 101, 83, 116, 97, 116, 117, 115, 34, 58, 34, 83, 85, 67, 67, 69, 83, 83, 34, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 67, 108, 97, 115, 115, 78, 97, 109, 101, 34, 58, 34, 99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 102, 100, 112, 46, 101, 110, 114, 105, 99, 104, 109, 101, 110, 116, 46, 109, 101, 115, 115, 97, 103, 101, 109, 101, 115, 104, 46, 101, 114, 114, 111, 114, 104, 97, 110, 100, 108, 101, 114, 46, 105, 109, 112, 108, 46, 67, 111, 110, 116, 105, 110, 117, 101, 80, 114, 111, 99, 101, 115, 115, 105, 110, 103, 69, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 34, 44, 34, 101, 114, 114, 111, 114, 84, 111, 112, 105, 99, 78, 97, 109, 101, 34, 58, 110, 117, 108, 108, 125, 44, 123, 34, 105, 100, 34, 58, 50, 44, 34, 103, 114, 111, 117, 112, 73, 100, 34, 58, 34, 99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 109, 101, 115, 115, 97, 103, 101, 109, 101, 115, 104, 34, 44, 34, 97, 114, 116, 105, 102, 97, 99, 116, 73, 100, 34, 58, 34, 112, 101, 114, 115, 105, 115, 116, 34, 44, 34, 118, 101, 114, 115, 105, 111, 110, 34, 58, 34, 48, 46, 49, 46, 49, 57, 34, 44, 34, 116, 121, 112, 101, 34, 58, 34, 80, 82, 79, 67, 69, 83, 83, 79, 82, 34, 44, 34, 114, 117, 110, 116, 105, 109, 101, 73, 100, 34, 58, 34, 105, 100, 120, 45, 100, 101, 100, 117, 112, 101, 34, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 115, 34, 58, 91, 123, 34, 101, 114, 114, 111, 114, 67, 111, 100, 101, 34, 58, 34, 52, 48, 48, 34, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 67, 108, 97, 115, 115, 78, 97, 109, 101, 34, 58, 34, 99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 102, 100, 112, 46, 101, 110, 114, 105, 99, 104, 109, 101, 110, 116, 46, 109, 101, 115, 115, 97, 103, 101, 109, 101, 115, 104, 46, 101, 114, 114, 111, 114, 104, 97, 110, 100, 108, 101, 114, 46, 105, 109, 112, 108, 46, 80, 117, 98, 108, 105, 115, 104, 84, 111, 69, 114, 114, 111, 114, 84, 111, 112, 105, 99, 69, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 34, 44, 34, 101, 114, 114, 111, 114, 84, 111, 112, 105, 99, 78, 97, 109, 101, 34, 58, 34, 105, 100, 120, 45, 101, 114, 114, 111, 114, 45, 116, 111, 112, 105, 99, 34, 125, 44, 123, 34, 101, 114, 114, 111, 114, 67, 111, 100, 101, 34, 58, 34, 42, 34, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 67, 108, 97, 115, 115, 78, 97, 109, 101, 34, 58, 34, 99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 102, 100, 112, 46, 101, 110, 114, 105, 99, 104, 109, 101, 110, 116, 46, 109, 101, 115, 115, 97, 103, 101, 109, 101, 115, 104, 46, 101, 114, 114, 111, 114, 104, 97, 110, 100, 108, 101, 114, 46, 105, 109, 112, 108, 46, 83, 116, 111, 112, 80, 114, 111, 99, 101, 115, 115, 105, 110, 103, 69, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 34, 44, 34, 101, 114, 114, 111, 114, 84, 111, 112, 105, 99, 78, 97, 109, 101, 34, 58, 110, 117, 108, 108, 125, 93, 44, 34, 115, 111, 117, 114, 99, 101, 84, 111, 112, 105, 99, 34, 58, 34, 105, 100, 120, 45, 112, 97, 121, 109, 101, 110, 116, 45, 115, 111, 117, 114, 99, 101, 34, 44, 34, 112, 105, 112, 101, 108, 105, 110, 101, 78, 111, 100, 101, 83, 116, 97, 116, 117, 115, 34, 58, 34, 83, 85, 67, 67, 69, 83, 83, 34, 44, 34, 101, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 67, 108, 97, 115, 115, 78, 97, 109, 101, 34, 58, 34, 99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 102, 100, 112, 46, 101, 110, 114, 105, 99, 104, 109, 101, 110, 116, 46, 109, 101, 115, 115, 97, 103, 101, 109, 101, 115, 104, 46, 101, 114, 114, 111, 114, 104, 97, 110, 100, 108, 101, 114, 46, 105, 109, 112, 108, 46, 67, 111, 110, 116, 105, 110, 117, 101, 80, 114, 111, 99, 101, 115, 115, 105, 110, 103, 69, 114, 114, 111, 114, 72, 97, 110, 100, 108, 101, 114, 34, 44, 34, 101, 114, 114, 111, 114, 84, 111, 112, 105, 99, 78, 97, 109, 101, 34, 58, 110, 117, 108, 108, 125, 93, 125]), RecordHeader(key = contentType, value = [97, 112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 106, 115, 111, 110]), RecordHeader(key = intuit_realmid, value = [49, 50, 51, 49, 52, 56, 53, 50, 53, 56, 57, 54, 49, 48, 57]), RecordHeader(key = intuit_iddomain, value = [116, 101, 115, 116]), RecordHeader(key = test, value = [116, 101, 115, 116, 86, 97, 108, 117, 101]), RecordHeader(key = intuit_tid, value = [48, 97, 49, 57, 102, 101, 98, 50, 45, 101, 57, 48, 99, 45, 52, 97, 51, 100, 45, 98, 57, 101, 102, 45, 57, 52, 50, 57, 52, 98, 98, 100, 53, 101, 98, 100]), RecordHeader(key = intuit_offeringid, value = [73, 110, 116, 117, 105, 116, 46, 112, 108, 97, 116, 102, 111, 114, 109, 46, 113, 98, 111, 46, 100, 116, 120, 46, 117, 105]), RecordHeader(key = intuit_locale, value = [101, 110, 95, 85, 83]), RecordHeader(key = entity_type, value = [80, 97, 121, 109, 101, 110, 116, 84, 114, 97, 110, 115, 97, 99, 116, 105, 111, 110, 69, 118, 101, 110, 116]), RecordHeader(key = entity_namespace, value = [99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 105, 100, 120, 46, 101, 118, 101, 110, 116, 46, 112, 97, 121, 109, 101, 110, 116, 46, 118, 49]), RecordHeader(key = intuit_country, value = [85, 83]), RecordHeader(key = provider_id, value = [49, 50, 51, 52]), RecordHeader(key = intuit_corpid, value = [65, 110, 103, 101, 108, 97]), RecordHeader(key = spring_json_header_types, value = [123, 34, 105, 110, 116, 117, 105, 116, 95, 111, 114, 105, 103, 105, 110, 97, 116, 105, 110, 103, 105, 112, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 101, 110, 116, 105, 116, 121, 95, 118, 101, 114, 115, 105, 111, 110, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 116, 101, 115, 116, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 116, 105, 100, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 111, 102, 102, 101, 114, 105, 110, 103, 105, 100, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 100, 101, 108, 105, 118, 101, 114, 121, 65, 116, 116, 101, 109, 112, 116, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 101, 110, 116, 105, 116, 121, 95, 112, 114, 111, 99, 101, 115, 115, 105, 110, 103, 95, 115, 116, 97, 116, 117, 115, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 80, 73, 80, 69, 76, 73, 78, 69, 95, 73, 68, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 73, 110, 116, 101, 103, 101, 114, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 108, 111, 99, 97, 108, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 116, 97, 114, 103, 101, 116, 45, 112, 114, 111, 116, 111, 99, 111, 108, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 101, 110, 116, 105, 116, 121, 95, 116, 121, 112, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 101, 110, 116, 105, 116, 121, 95, 110, 97, 109, 101, 115, 112, 97, 99, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 82, 79, 85, 84, 73, 78, 71, 95, 73, 78, 70, 79, 34, 58, 34, 99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 102, 100, 112, 46, 101, 110, 114, 105, 99, 104, 109, 101, 110, 116, 46, 112, 105, 112, 101, 108, 105, 110, 101, 46, 112, 114, 111, 99, 101, 115, 115, 111, 114, 115, 46, 82, 111, 117, 116, 105, 110, 103, 73, 110, 102, 111, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 99, 111, 117, 110, 116, 114, 121, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 112, 114, 111, 118, 105, 100, 101, 114, 95, 105, 100, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 99, 111, 114, 112, 105, 100, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 99, 111, 110, 116, 101, 110, 116, 84, 121, 112, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 114, 101, 97, 108, 109, 105, 100, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 105, 100, 100, 111, 109, 97, 105, 110, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 125])], isReadOnly = true), key=null, value=[B@dfd024f, timestamp=null)], headers={id=1823c929-83f7-4b15-358f-821b3ae843b4, timestamp=1639059115250}]
  </code></pre>
  <p>Looks like since the Timeout exception is thrown, the subscriber for the in channel is lost and seeing the below logged by the errorChannel,</p>
  <pre><code>ErrorMessage [payload=org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application-1.execute-in-0'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[761], headers={intuit_originatingip=192.19.0.1, entity_version=[B@24abc8d8, test=[B@677453b9, intuit_tid=[B@20a8288, intuit_offeringid=[B@3feabc33, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, intuit_locale=en_US, kafka_receivedTopic=idx-payment-source, kafka_offset=5, entity_type=[B@45fd5e1e, entity_namespace=[B@513483, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@597cf885, intuit_country=US, provider_id=[B@5619f8c3, intuit_corpid=[B@3c78cd17, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTimestamp=1639059114780, kafka_groupId=consumer-dedupe, intuit_realmid=[B@135a7928, intuit_iddomain=test}], failedMessage=GenericMessage [payload=byte[761], headers={intuit_originatingip=192.19.0.1, entity_version=[B@24abc8d8, test=[B@677453b9, intuit_tid=[B@20a8288, intuit_offeringid=[B@3feabc33, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, intuit_locale=en_US, kafka_receivedTopic=idx-payment-source, kafka_offset=5, entity_type=[B@45fd5e1e, entity_namespace=[B@513483, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@597cf885, intuit_country=US, provider_id=[B@5619f8c3, intuit_corpid=[B@3c78cd17, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTimestamp=1639059114780, kafka_groupId=consumer-dedupe, intuit_realmid=[B@135a7928, intuit_iddomain=test}], headers={kafka_data=ConsumerRecord(topic = idx-payment-source, partition = 0, leaderEpoch = 0, offset = 5, CreateTime = 1639059114780, serialized key size = -1, serialized value size = 1186, headers = RecordHeaders(headers = [RecordHeader(key = intuit_originatingip, value = [49, 57, 50, 46, 49, 57, 46, 48, 46, 49]), RecordHeader(key = entity_version, value = [118, 49]), RecordHeader(key = test, value = [116, 101, 115, 116, 86, 97, 108, 117, 101]), RecordHeader(key = intuit_tid, value = [56, 49, 55, 53, 52, 55, 51, 54, 45, 101, 102, 97, 55, 45, 52, 102, 53, 48, 45, 98, 101, 101, 57, 45, 102, 50, 54, 48, 48, 56, 101, 49, 53, 48, 101, 48]), RecordHeader(key = intuit_offeringid, value = [73, 110, 116, 117, 105, 116, 46, 112, 108, 97, 116, 102, 111, 114, 109, 46, 113, 98, 111, 46, 100, 116, 120, 46, 117, 105]), RecordHeader(key = intuit_locale, value = [101, 110, 95, 85, 83]), RecordHeader(key = entity_type, value = [80, 97, 121, 109, 101, 110, 116, 84, 114, 97, 110, 115, 97, 99, 116, 105, 111, 110, 69, 118, 101, 110, 116]), RecordHeader(key = entity_namespace, value = [99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 105, 100, 120, 46, 101, 118, 101, 110, 116, 46, 112, 97, 121, 109, 101, 110, 116, 46, 118, 49]), RecordHeader(key = intuit_country, value = [85, 83]), RecordHeader(key = provider_id, value = [84, 101, 115, 116, 80, 114, 111, 118, 49, 50, 51, 52]), RecordHeader(key = intuit_corpid, value = [74, 97, 99, 105, 110, 100, 97]), RecordHeader(key = contentType, value = [97, 112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 106, 115, 111, 110]), RecordHeader(key = intuit_realmid, value = [49, 50, 51, 49, 52, 56, 53, 50, 53, 56, 57, 54, 49, 48, 57]), RecordHeader(key = intuit_iddomain, value = [116, 101, 115, 116]), RecordHeader(key = spring_json_header_types, value = [123, 34, 105, 110, 116, 117, 105, 116, 95, 111, 114, 105, 103, 105, 110, 97, 116, 105, 110, 103, 105, 112, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 99, 111, 117, 110, 116, 114, 121, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 108, 111, 99, 97, 108, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 99, 111, 110, 116, 101, 110, 116, 84, 121, 112, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 105, 100, 100, 111, 109, 97, 105, 110, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 125])], isReadOnly = false), key = null, value = [B@3609e9b), id=99a23404-032b-363c-ae7a-2e5f6cd83ef8, sourceData=ConsumerRecord(topic = idx-payment-source, partition = 0, leaderEpoch = 0, offset = 5, CreateTime = 1639059114780, serialized key size = -1, serialized value size = 1186, headers = RecordHeaders(headers = [RecordHeader(key = intuit_originatingip, value = [49, 57, 50, 46, 49, 57, 46, 48, 46, 49]), RecordHeader(key = entity_version, value = [118, 49]), RecordHeader(key = test, value = [116, 101, 115, 116, 86, 97, 108, 117, 101]), RecordHeader(key = intuit_tid, value = [56, 49, 55, 53, 52, 55, 51, 54, 45, 101, 102, 97, 55, 45, 52, 102, 53, 48, 45, 98, 101, 101, 57, 45, 102, 50, 54, 48, 48, 56, 101, 49, 53, 48, 101, 48]), RecordHeader(key = intuit_offeringid, value = [73, 110, 116, 117, 105, 116, 46, 112, 108, 97, 116, 102, 111, 114, 109, 46, 113, 98, 111, 46, 100, 116, 120, 46, 117, 105]), RecordHeader(key = intuit_locale, value = [101, 110, 95, 85, 83]), RecordHeader(key = entity_type, value = [80, 97, 121, 109, 101, 110, 116, 84, 114, 97, 110, 115, 97, 99, 116, 105, 111, 110, 69, 118, 101, 110, 116]), RecordHeader(key = entity_namespace, value = [99, 111, 109, 46, 105, 110, 116, 117, 105, 116, 46, 105, 100, 120, 46, 101, 118, 101, 110, 116, 46, 112, 97, 121, 109, 101, 110, 116, 46, 118, 49]), RecordHeader(key = intuit_country, value = [85, 83]), RecordHeader(key = provider_id, value = [84, 101, 115, 116, 80, 114, 111, 118, 49, 50, 51, 52]), RecordHeader(key = intuit_corpid, value = [74, 97, 99, 105, 110, 100, 97]), RecordHeader(key = contentType, value = [97, 112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 106, 115, 111, 110]), RecordHeader(key = intuit_realmid, value = [49, 50, 51, 49, 52, 56, 53, 50, 53, 56, 57, 54, 49, 48, 57]), RecordHeader(key = intuit_iddomain, value = [116, 101, 115, 116]), RecordHeader(key = spring_json_header_types, value = [123, 34, 105, 110, 116, 117, 105, 116, 95, 111, 114, 105, 103, 105, 110, 97, 116, 105, 110, 103, 105, 112, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 99, 111, 117, 110, 116, 114, 121, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 108, 111, 99, 97, 108, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 99, 111, 110, 116, 101, 110, 116, 84, 121, 112, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 105, 110, 116, 117, 105, 116, 95, 105, 100, 100, 111, 109, 97, 105, 110, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 125])], isReadOnly = false), key = null, value = [B@3609e9b), timestamp=1639059117896}] for original GenericMessage [payload=byte[761], headers={intuit_originatingip=192.19.0.1, entity_version=[B@24abc8d8, test=[B@677453b9, intuit_tid=[B@20a8288, intuit_offeringid=[B@3feabc33, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, intuit_locale=en_US, kafka_receivedTopic=idx-payment-source, kafka_offset=5, entity_type=[B@45fd5e1e, entity_namespace=[B@513483, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@597cf885, intuit_country=US, provider_id=[B@5619f8c3, intuit_corpid=[B@3c78cd17, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTimestamp=1639059114780, kafka_groupId=consumer-dedupe, intuit_realmid=[B@135a7928, intuit_iddomain=test}]
  </code></pre>
  <p>Once this occurs, even after the kafka issue is resolved, the application is not able to recover unless restarted.</p>
  <p>Here is the sample project, <a href="https://github.com/ssruthisree/reactive-processor-kafka" rel="nofollow noreferrer">https://github.com/ssruthisree/reactive-processor-kafka</a> aggregate is my Function that is publishing. To replicate the error steps done: 1) Start appln 2) Verify messages consumed by aggregate and published to transform topic. 3) Delete the transform topic 4) Exception thrown from Producer not caught by Flux's onError.</p>
  <p>How can we avoid the subscription failure on producer errors? And continue consuming once kafka issue is resolved without application restart?</p>