1

I am building a Spring Cloud Stream Kafka processor app that will consume raw data with a String key and sometimes a null payload from a Kafka topic. I want to produce to another topic a String key and the null payload (known as a tombstone within Kafka). In order to use raw headers on the message, I need to output a byte[], but if I encode KafkaNull.INSTANCE into a byte[] it will literally output a String of the object hashcode.

If I try to send anything other than a byte[], I can't use raw headers.

What is the correct way to do this? Where would the headers even be if the payload is null? I set producer.headerMode=embeddedHeaders on deploy and that didn't seem to make a difference, still have the hashcode as the payload.

0

1 Answer 1

1

You can use the latter introduced useNativeEncoding property (which overlaps the raw headerMode in that it skips headers) but allows you use a native Kafka serializer.

See http://docs.spring.io/autorepo/docs/spring-cloud-stream-docs/Brooklyn.BUILD-SNAPSHOT/reference/htmlsingle/#_producer_properties or http://docs.spring.io/autorepo/docs/spring-cloud-stream-docs/Chelsea.BUILD-SNAPSHOT/reference/htmlsingle/#_producer_properties

7
  • 1
    Ah, right, of course, so a KafkaNull payload will work with that mode. spring-integration-kafka uses a "special" message payload KafkaNull.INSTANCE to represent a null payload. Mar 22, 2017 at 19:29
  • I think we should treat if it doesn't as a bug ... Maybe we can have a similar convention at the core level and do the translation in the binder. Mar 22, 2017 at 19:34
  • Thanks. I am using KafkaNull.Instance when I produce. How do I override the producer serializer? I added several configs and nothing seems to change it, both in the stream def and deployment props. In fact the consumer warned it picked up an extra config value.serializer which makes sense, but the producer is still set to use bytearray serializer. Mar 22, 2017 at 20:47
  • --spring.cloud.stream.output.producer.useNativeEncoding=true --spring.cloud.stream.kafka.binder.configuration.value.serializer=org.apache.kafka.common.serialization.StringSerializer --spring.cloud.stream.kafka.bindings.output.producer.configuration.value.serializer=org.apache.kafka.common.serialization.StringSerializer Mar 22, 2017 at 20:51
  • Tried this locally with the same set of configurations and it worked (Brooklyn.SR3) . A couple questions - is your channel named output ? How can you tell that the valuee serializer is not set? There will be two producer configurations logged - one is for querying metadata - you should check the second one. Mar 22, 2017 at 21:15

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.