title: how to set concurrency (or other configurations) for ConcurrentKafkaListenerContainerFactory per StreamListener
body: <p>We have scenario where our application(spring boot, spring-cloud-stream based) listens to multiple Kafka topics (TOPIC_A with 3 partitions, TOPIC_B with 1 partition,TOPIC_C with 10 partitions) i.e. 3 @StreamListener methods.</p>
  
  <pre><code>   @StreamListener(TopicASink.INPUT)
      public void processTopicA(Message&lt;String&gt; msg) {
          logger.info("** recieved message: {} ", msg.getPayload());
        // do some processing
      }
  
     @StreamListener(TopicBSink.INPUT)
      public void processTopicB(Message&lt;String&gt; msg) {
          logger.info("** recieved message: {} ", msg.getPayload());
         // do some processing
      }
  
     @StreamListener(TopicCSink.INPUT)
      public void processTopicC(Message&lt;String&gt; msg) {
          logger.info("** recieved message: {} ", msg.getPayload());
        // do some processing
      }
  </code></pre>
  
  <p>We need customize error handling and retry mechanism hence achieving this by configuring ConcurrentKafkaListenerContainerFactory bean.</p>
  
  <pre><code>       @Bean
          public ConcurrentKafkaListenerContainerFactory concurrentKafkaListenerContainerFactory(ConsumerFactory&lt;Object,Object&gt; consumerFactory) {
              ConcurrentKafkaListenerContainerFactory&lt;Object, Object&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
              factory.setConcurrency(2);   // we need to customize this per topic based on number of partitions
              factory.setConsumerFactory(consumerFactory);
              RetryTemplate retryTemplate = new RetryTemplate();
              retryTemplate.setRetryPolicy(new SimpleRetryPolicy(10));
              factory.setRetryTemplate(retryTemplate);
              factory.setErrorHandler(new SeekToCurrentErrorHandler(new FixedBackOff(FixedBackOff.DEFAULT_INTERVAL, 10)));
              return factory;
          }
  </code></pre>
  
  <p>Problem is now we need some properties of the KafkaListenerContainer's to vary per @StreamListener (i.e. per topic in this case) , say to have a concurrency of 3 for TOPIC_A, 10 for TOPIC_C etc. instead of common concurrency set on the factory or set SeekToCurrentErrorHandler for TOPIC_A,TOPIC_C but not for TOPIC_B (or different ErrorHandler for some topics).</p>
  
  <p>How can this be achieved per container level?</p>
  
  <hr>
  
  <p>stack trace after trying solution with solution with reflection shared below</p>
  
  <pre><code>o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException: Exception thrown while invoking com.jta.poc.kafkapoc.KafkaStreamPocApplication$MessageProcessor#processInput[1 args]; nested exception is com.jta.poc.kafkapoc.MyNewRetryableException, failedMessage=GenericMessage [payload=byte[35], headers={kafka_timestampType=CREATE_TIME, kafka_receivedTopic=new_input_topic, spanTraceId=e3382bf49eaa5343, spanId=e3382bf49eaa5343, nativeHeaders={spanTraceId=[e3382bf49eaa5343], spanId=[efc90644fc4c7dee], spanSampled=[0], X-B3-TraceId=[e3382bf49eaa5343], X-B3-SpanId=[efc90644fc4c7dee], X-B3-ParentSpanId=[e3382bf49eaa5343], spanParentSpanId=[e3382bf49eaa5343], X-B3-Sampled=[0]}, kafka_offset=26, X-B3-SpanId=e3382bf49eaa5343, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2a011bf8, X-B3-Sampled=0, X-B3-TraceId=e3382bf49eaa5343, id=3c86f652-f16e-2f59-1a59-f3d8601849f0, kafka_receivedPartitionId=1, spanSampled=0, kafka_receivedTimestamp=1586250896206, kafka_acknowledgment=Acknowledgment for ConsumerRecord(topic = new_input_topic, partition = 1, offset = 26, CreateTime = 1586250896206, serialized key size = -1, serialized value size = 35, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = [B@68df9f80), contentType=application/json, timestamp=1586274368357}]
      at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:63)
      at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)
      at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:158)
      at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
      at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:132)
      at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105)
      at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:445)
      at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:394)
      at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181)
      at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160)
      at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
      at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)
      at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203)
      at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$300(KafkaMessageDrivenChannelAdapter.java:70)
      at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:387)
      at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:364)
      at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.lambda$onMessage$0(RetryingMessageListenerAdapter.java:120)
      at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
      at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:211)
      at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:114)
      at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:40)
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1071)
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1051)
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:998)
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:866)
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:724)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: com.jta.poc.kafkapoc.MyNewRetryableException
      at com.jta.poc.kafkapoc.KafkaStreamPocApplication$MessageProcessor.consumeMessage(KafkaStreamPocApplication.java:164)
      at com.jta.poc.kafkapoc.KafkaStreamPocApplication$MessageProcessor.lambda$processInput$0(KafkaStreamPocApplication.java:107)
      at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
      at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:164)
      at com.jta.poc.kafkapoc.KafkaStreamPocApplication$MessageProcessor.processInput(KafkaStreamPocApplication.java:105)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:181)
      at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:114)
      at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:55)
      ... 29 more
  </code></pre>