title: Understanding Spring Cloud Stream content types
body: <p>I wrote a custom processor similar to <a href="https://stackoverflow.com/q/35500759/944849">this one</a>. In particular, the processor takes an InputDto and returns json. Following guidance in the Q&amp;A, my custom processor has an application.properties file with this content:</p>
  
  <pre><code>spring.cloud.stream.bindings.input.content-type=application/x-java-object;com.company.InputDto
  spring.cloud.stream.bindings.output.content-type=application/json
  </code></pre>
  
  <p>From <a href="https://stackoverflow.com/q/41781351/944849">this Q&amp;A</a> I created a <code>spring.integration.properties</code> file with this line:</p>
  
  <pre><code>spring.integration.readOnly.headers=contentType
  </code></pre>
  
  <p>And I have working automated integration tests. So far so good.</p>
  
  <p>I created a stream in the SCDF shell including my processor. <code>time</code> and <code>httpclient</code> are working well, so I didn't show detailed args for those. Added newlines here &amp; throughout for readability.</p>
  
  <pre><code>stream create --name test --definition "time &lt;args&gt; 
    | httpclient &lt;args&gt; 
    | splitter --expression=\"#jsonPath(payload,'$..[0:]')\" 
    | myprocessor 
    | log"
  </code></pre>
  
  <p>I have debug logging enabled. <code>httpclient</code> produces messages with <code>contentType=text/plain</code> and payload:</p>
  
  <pre><code>payload=[{"name":"first","url":"url1"},{"name":"second","url":"url2"}]
  </code></pre>
  
  <p>The <code>splitter</code> creates two messages like this, per logs (as expected):</p>
  
  <pre><code>message: GenericMessage [payload={name=first, url=url1}, 
    headers={...contentType=text/plain, ... }]
  
  message: GenericMessage [payload={name=second, url=url2}, 
    headers={...contentType=text/plain, ... }]
  </code></pre>
  
  <p>And the custom processor I wrote fails with this exception:</p>
  
  <pre><code>org.springframework.messaging.converter.MessageConversionException: Cannot 
  convert from [java.util.LinkedHashMap] to [com.company.InputDto] for 
  GenericMessage [payload={name=first, url=url1}, headers={sequenceNumber=1, 
     kafka_offset=xx, sequenceSize=x, correlationId=yy, id=zzz, 
     kafka_receivedPartitionId=0, contentType=text/plain, 
     kafka_receivedTopic=test.splitter, timestamp=1503591622851}]
      at 
   org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:142) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE] 
  ....
  </code></pre>
  
  <p>I'm not sure where the <code>LinkedHashMap</code> is coming from. I tried changing the application.properties file to:</p>
  
  <pre><code>spring.cloud.stream.bindings.input.content-type=application/json;com.company.InputDto
  </code></pre>
  
  <p>but no help. I also tried adding </p>
  
  <pre><code>--spring.cloud.stream.bindings.output.contentType='application/json'
  </code></pre>
  
  <p>to the <code>splitter</code> when creating the stream (following directions in a <a href="https://github.com/spring-cloud/spring-cloud-dataflow-samples/tree/master/streaming/http-to-cassandra" rel="nofollow noreferrer">sample app</a>), but still get the exception.</p>
  
  <p>I've spent hours and just cannot see what I'm missing. Appreciate any assistance.</p>
  
  <p>My custom processor uses Spring Cloud Dalston.SR3. I'm using SCDF Server and shell 1.3.0.M1. Using the Kafka binder.</p>
  
  <hr>
  
  <p>Update, more info. I looked at the code in the Splitter starter and wrote a little test case to simulate what it's doing.</p>
  
  <pre><code>final ExpressionEvaluatingSplitter splitter = new ExpressionEvaluatingSplitter(
      new SpelExpressionParser().parseExpression(
          "T(com.jayway.jsonpath.JsonPath).read(payload, '$..[0:]')"));
  final PollableChannel channel = new QueueChannel();
  splitter.setOutputChannel(channel);
  splitter.handleMessage(new GenericMessage&lt;Object&gt;(
    "[{\"name\":\"first\",\"url\":\"url1\"},
      {\"name\":\"second\",\"url\":\"url2\"}]"));
  final Message&lt;?&gt; message = channel.receive(10);
  System.out.println("payload class: " + message.getPayload().getClass());
  System.out.println("payload: " + message.getPayload());
  System.out.println(channel.receive(10));
  </code></pre>
  
  <p>This produces output:</p>
  
  <pre><code>payload class: class java.util.LinkedHashMap
  payload: {name=first, url=url1}
  GenericMessage [payload={name=second, url=url2}, headers={sequenceNumber=2,
    correlationId=xx, id=yy, sequenceSize=2, timestamp=1503609649518}]
  </code></pre>
  
  <p>Aha, the <code>LinkedHashMap</code>! Now I just need to convince the splitter to send the output as plain text or json, not a Map.</p>
  
  <hr>
  
  <p>Update 2. I've been able to duplicate this issue without using any custom processors.</p>
  
  <pre><code>stream create --name test --definition "time &lt;args&gt;
    | httpclient &lt;args&gt;
    | splitter --expression=\"#jsonPath(payload,'$.[0:]')\" 
           --outputType=\"application/json\"
    | transform --expression=\"new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(payload) != null\" 
           --outputType=\"application/json\"
    | log"
  </code></pre>
  
  <p>When run, the splitter log file contains this exception (abridged), the infamous "payload must not be null" error:</p>
  
  <pre><code>2017-08-25 13:09:30,322 DEBUG -kafka-listener-2 o.s.i.c.DirectChannel:411 - preSend on channel 'output', message: GenericMessage [payload={name=first, url=url1}, headers={sequenceNumber=1, kafka_offset=xx sequenceSize=2, correlationId=yy, id=zz, kafka_receivedPartitionId=0, contentType=text/plain, kafka_receivedTopic=test.httpclient, timestamp=1503680970322}]
  2017-08-25 13:09:30,328 ERROR -kafka-listener-2 o.s.k.l.LoggingErrorHandler:37 - Error while processing: ConsumerRecord(topic = test.httpclient, partition = 0, offset = 52606, CreateTime = 1503680967030, checksum = 2166925613, serialized key size = -1, serialized value size = 470, key = null, value = [B@6567451e)
  org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'output'; nested exception is java.lang.IllegalArgumentException: payload must not be null
      at     org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:449) ~[spring-integration-core-.3.8.RELEASE.jar!/:4.3.8.RELEASE]
  </code></pre>
  
  <p>It looks like the splitter is having difficulty converting a LinkedHashMap to JSON. Any way to make the conversion happen?</p>
  
  <p>I also tried setting the outputType of the httpclient processor to be explicitly <code>application/json</code> and it didn't seem to make a difference. (Following the <a href="https://docs.spring.io/spring-cloud-dataflow/docs/1.3.0.M1/reference/htmlsingle/#passing_content_type_properties" rel="nofollow noreferrer">docs</a>. The example shows the outputType values with shell quotes, I also tried without, no difference.)</p>
  
  <p>Apps were loaded into SCDF server using command (replaced '.' in bit-ly so SO would accept link)</p>
  
  <pre><code>app import --uri http://bit-ly/Bacon-RELEASE-stream-applications-kafka-10-maven
  </code></pre>
  
  <p>Noticed a few more things. The debug logging shows the content type as <code>text/plain</code> throughout, so it appears to not be picking up the content type I set explicitly. Also, the errors go away if I remove the <code>transform</code> processor. I see the data in the log, but not in JSON format, just this:</p>
  
  <pre><code>{name=first, url=url1}
  </code></pre>