title: Kafka stream does not retry on deserialisation error
body: <p>Spring cloud Kafka stream does not retry upon deserialization error even after specific configuration. The expectation is, it should retry based on the configured retry policy and at the end push the failed message to DLQ. </p>
  
  <p>Configuration as below.</p>
  
  <pre><code>spring.cloud.stream.bindings.input_topic.consumer.maxAttempts=7
  spring.cloud.stream.bindings.input_topic.consumer.backOffInitialInterval=500
  spring.cloud.stream.bindings.input_topic.consumer.backOffMultiplier=10.0
  spring.cloud.stream.bindings.input_topic.consumer.backOffMaxInterval=100000
  spring.cloud.stream.bindings.iinput_topic.consumer.defaultRetryable=true
  </code></pre>
  
  <pre><code>public interface MyStreams {
  
      String INPUT_TOPIC = "input_topic";
      String INPUT_TOPIC2 = "input_topic2";
      String ERROR = "apperror";
      String OUTPUT = "output";
  
      @Input(INPUT_TOPIC)
      KStream&lt;String, InObject&gt; inboundTopic();
  
      @Input(INPUT_TOPIC2)
      KStream&lt;Object, InObject&gt; inboundTOPIC2();
  
      @Output(OUTPUT)
      KStream&lt;Object, outObject&gt; outbound();
  
      @Output(ERROR)
      MessageChannel outboundError();
  }
  
  @StreamListener(MyStreams.INPUT_TOPIC)
      @SendTo(MyStreams.OUTPUT)
      public KStream&lt;Key, outObject&gt; processSwft(KStream&lt;Key, InObject&gt; myStream) {
          return myStream.mapValues(this::transform);
      }
  </code></pre>
  
  <p>The metadataRetryOperations in KafkaTopicProvisioner.java is always null and hence it creates a new RetryTemplate in the <code>afterPropertiesSet()</code>.</p>
  
  <pre><code>public KafkaTopicProvisioner(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties, KafkaProperties kafkaProperties) {
          Assert.isTrue(kafkaProperties != null, "KafkaProperties cannot be null");
          this.adminClientProperties = kafkaProperties.buildAdminProperties();
          this.configurationProperties = kafkaBinderConfigurationProperties;
          this.normalalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties, kafkaBinderConfigurationProperties);
      }
  
      public void setMetadataRetryOperations(RetryOperations metadataRetryOperations) {
          this.metadataRetryOperations = metadataRetryOperations;
      }
  
      public void afterPropertiesSet() throws Exception {
          if (this.metadataRetryOperations == null) {
              RetryTemplate retryTemplate = new RetryTemplate();
              SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
              simpleRetryPolicy.setMaxAttempts(10);
              retryTemplate.setRetryPolicy(simpleRetryPolicy);
              ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
              backOffPolicy.setInitialInterval(100L);
              backOffPolicy.setMultiplier(2.0D);
              backOffPolicy.setMaxInterval(1000L);
              retryTemplate.setBackOffPolicy(backOffPolicy);
              this.metadataRetryOperations = retryTemplate;
          }
  
      }
  </code></pre>