title: Spring-Cloud-Stream-Kafka-Binder functional style ignores custom De/Serializer and/or useNativeEncoding?
body: <p>We're just upgrading to the 3.0.0-Release of Spring-Cloud-Stream and encounter the following problems:</p>
  
  <p>When using the functional style like this:</p>
  
  <pre><code>public class EventProcessor {
  
      private final PriceValidator priceValidator;
  
      @Bean
      public Function&lt;Flux&lt;EnrichedValidationRequest&gt;, Flux&lt;ValidationResult&gt;&gt; validate() {
          return enrichedValidationRequestFlux -&gt; enrichedValidationRequestFlux
                  .map(ProcessingContext::new)
                  .flatMap(priceValidator::validateAndMap);
      }
  }
  </code></pre>
  
  <p>The application.yaml looks like this:</p>
  
  <pre><code>spring.cloud.stream:
    default-binder: kafka
    kafka:
      binder:
        brokers: ${kafka.broker.prod}
        auto-create-topics: false
    function.definition: validate
  
  # INPUT: enrichedValidationRequests
  spring.cloud.stream.bindings.validate-in-0:
    destination: ${kafka.topic.${spring.application.name}.input.enrichedValidationRequests}
    group: ${spring.application.name}.${STAGE:NOT_SET}
    consumer:
      useNativeDecoding: true
  
  
  spring.cloud.stream.kafka.bindings.validate-in-0:
    consumer:
      configuration:
        key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
        value.deserializer: de.pricevalidator.deserializer.EnrichedValidationRequestDeserializer
  
  
  # OUTPUT: validationResults
  spring.cloud.stream.bindings.validate-out-0:
    destination: validationResultsTmp
    producer:
      useNativeEncoding: true
  
  spring.cloud.stream.kafka.bindings.validate-out-0:
    producer:
      compression.type: lz4
      messageKeyExpression: payload.offerKey
      configuration:
        key.serializer: org.apache.kafka.common.serialization.StringSerializer
        value.serializer: de.pricevalidator.serializer.ValidationResultSerializer
  </code></pre>
  
  <p>It seems like the serialization is done twice - when we intercept the messages that get produced in the kafka topic, the consumer would just display them as JSON (strings), but now it's an unreadable byte[]. Also, the downstream consumers in production can't deserialize the messages anymore. Strangely enough, the deserialization of the input messages seems to work just fine, no matter what we put into the consumer properties (either on binder or on default kafka level)
  We have a feeling, this bug "is back", but we can't find the exact spot in the code: <a href="https://github.com/spring-cloud/spring-cloud-stream/issues/1536" rel="nofollow noreferrer">https://github.com/spring-cloud/spring-cloud-stream/issues/1536</a></p>
  
  <p>Our (ugly) workaround:</p>
  
  <pre><code>@Slf4j
  @Configuration
  public class KafkaMessageConverterConfiguration {
  
      @ConditionalOnProperty(value = "spring.cloud.stream.default-binder", havingValue = "kafka")
      @Bean
      public MessageConverter validationResultConverter(BinderTypeRegistry binder, ObjectMapper objectMapper) {
          return new AbstractMessageConverter(MimeType.valueOf("application/json")) {
              @Override
              protected boolean supports(final Class&lt;?&gt; clazz) {
                  return ValidationResult.class.isAssignableFrom(clazz);
              }
  
              @Override
              protected Object convertToInternal(final Object payload, final MessageHeaders headers, final Object conversionHint) {
                  return payload;
              }
          };
      }
  }
  </code></pre>
  
  <p>Is there a "proper" way of setting the custom serializer or to get the native encoding as it was before?</p>