title: Kafka producer sending invalid characters
body: <p>Using following code, I send Elasticsearch documents for indexing. I tried converting basic Objects to JSON and sent via producer. However, every message (as checked from the console) appends jibberish characters like - <em>��t�{"productId":2455</em></p>
  
  <pre><code>public boolean sendMessage()
  {
      PageRequest page = new PageRequest(0, 1); 
      Product p = product.findByName("Cream", page).getContent().get(0);
      String json = "";
      ObjectMapper mapper = new ObjectMapper();
      try {
          json = mapper.writeValueAsString(p);
      } catch (JsonProcessingException e1) {
          // TODO Auto-generated catch block
          e1.printStackTrace();
      }       
      logger.info("JSON = " + json);
  
      boolean status =  inputToKafka.send(org.springframework.integration.support.MessageBuilder.withPayload(json).build());
      try {
          Thread.sleep(10000);
      } catch (InterruptedException e) {
          e.printStackTrace();
      }
      return status;
  }
  </code></pre>
  
  <p>Outbound configuration</p>
  
  <pre><code> &lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;beans xmlns="http://www.springframework.org/schema/beans"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns:int="http://www.springframework.org/schema/integration"
         xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
         xmlns:task="http://www.springframework.org/schema/task"
         xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
          http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
          http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
          http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"&gt;
  
      &lt;int:channel id="inputToKafka"&gt;
          &lt;int:queue/&gt;
      &lt;/int:channel&gt;
  
      &lt;int-kafka:outbound-channel-adapter
              id="kafkaOutboundChannelAdapter"
              kafka-producer-context-ref="kafkaProducerContext"
              channel="inputToKafka"&gt;
          &lt;int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/&gt;
      &lt;/int-kafka:outbound-channel-adapter&gt;
  
      &lt;task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/&gt;
  
      &lt;int-kafka:producer-context id="kafkaProducerContext"&gt;
          &lt;int-kafka:producer-configurations&gt;
              &lt;int-kafka:producer-configuration broker-list="localhost:9092"
                                                topic="test_topic"
                                                compression-codec="default"/&gt;
          &lt;/int-kafka:producer-configurations&gt;
      &lt;/int-kafka:producer-context&gt;
  
      &lt;/beans&gt;
  </code></pre>
  
  <p>Any clue ?</p>
  
  <p>Plugin used: <a href="https://github.com/spring-projects/spring-integration-kafka" rel="nofollow">Spring Extension Kafka</a></p>