title: StreamBridge problem creating message with custom MessageConverter
body: <p>We are migrating our system of events to the functional programming model. We followed the next &quot;<a href="https://stackoverflow.com/questions/65978055/enablebinding-output-input-deprecated-since-version-of-3-1-of-spring-cloud-str">guide</a>&quot; and it worked quite well for the consumers but the producers with StreamBridge are not creating the messages properly.</p>
  <p>We have the next error:</p>
  <pre><code>java.lang.ClassCastException: class com.streamdemo.domain.Event cannot be cast to class java.lang.String (com.streamdemo.domain.Event is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')
  </code></pre>
  <p>We are working with <code>2021.0.0</code> spring-cloud version.</p>
  <p>with a simple configuration:</p>
  <pre><code>spring.cloud.stream:
    bindings:
      demo-channel-out-0:
        producer:
          use-native-encoding: true
        content-type: application/json
  </code></pre>
  <p>This is our custom <code>MessageConverter</code> which is used in all our microservices and for this reason, it is important to keep the format of the message:</p>
  <pre><code>public class StreamMessageConverter extends AbstractMessageConverter {
  
    private static final Logger log = LoggerFactory.getLogger(StreamMessageConverter.class);
    private static final MimeType JSON_MIME_TYPE = MimeType.valueOf(&quot;application/json&quot;);
    private final Class&lt;?&gt; clazz;
    private final ObjectMapper objectMapper;
  
    public StreamMessageConverter(ObjectMapper objectMapper) {
      this(objectMapper, Object.class);
    }
  
    public StreamMessageConverter(ObjectMapper objectMapper, Class&lt;?&gt; clazz) {
      super(JSON_MIME_TYPE);
      this.objectMapper = objectMapper;
      this.clazz = clazz;
    }
  
    protected boolean supports(Class&lt;?&gt; clazz) {
      return clazz.equals(clazz);
    }
  
    @Nullable
    protected Object convertFromInternal(Message&lt;?&gt; message, Class&lt;?&gt; targetClass, @Nullable Object conversionHint) {
      Object payload = message.getPayload();
  
      try {
        return payload instanceof byte[] ? this.objectMapper.readValue((byte[]) payload, targetClass)
            : this.objectMapper.readValue((String) payload, targetClass);
      } catch (IOException var6) {
        log.info(&quot;Unable to read json payload as object&quot;, var6);
        return null;
      }
    }
  
    @Nullable
    protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {
      try {
        return this.objectMapper.writeValueAsString(payload).getBytes(&quot;UTF-8&quot;);
      } catch (IOException var5) {
        log.info(&quot;Unable to write payload as json&quot;, var5);
        return null;
      }
    }
  }
  </code></pre>
  <p>and also we are using <code>StreamBridge</code> to produce the events:</p>
  <pre><code>@Service
  public class MessageBroker {
  
    private final StreamBridge source;
  
    public MessageBroker(StreamBridge source) {
      this.source = source;
    }
  
    public void send(Event event) {
      source.send(&quot;demo-channel-out-0&quot;, event);
    }
  }
  </code></pre>
  <p>When the <code>source.send</code> method is called, internally it calls to <code>StreamMessageConverter.convertFromInternal</code> when we think it should not do it. In the previous model (<code>@EnableBinding, @StreamListener</code>), only <code>StreamMessageConverter.convertToInternal</code> was called in order to serialize and send the message to the Kafka topic.</p>
  <p>We created <a href="https://github.com/natete/spring-cloud-stream-bridge-demo" rel="nofollow noreferrer">this demo project</a> where you can reproduce the error.</p>