spring cloud dataflow input source kafka

Question

I have problem with a common task and i can find any solutions or help (maybe some properties i need to pass for this to work ?) I use local server 1.3.0.M2 and create simple stream

dataflow:>stream create --name test --definition ":bosstds > log" --deploy

In log i got this :

2017-09-28 12:31:00.644 INFO 5156 --- [ -C-1] o.a.k.c.c.internals.AbstractCoordinator : Successfully joined group test with generation 1 2017-09-28 12:31:00.646 INFO 5156 --- [ -C-1] o.a.k.c.c.internals.ConsumerCoordinator : Setting newly assigned partitions [bosstds-0] for group test 2017-09-28 12:31:00.671 INFO 5156 --- [ -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3 : partitions assigned:[bosstds-0] 2017-09-28 12:37:08.898 ERROR 5156 --- [ -L-1] o.s.c.s.b.k.KafkaMessageChannelBinder : Could not convert message: 74657374 java.lang.StringIndexOutOfBoundsException: String index out of range: 103 at java.lang.String.checkBounds(String.java:385) ~[na:1.8.0_144] at java.lang.String.(String.java:425) ~[na:1.8.0_144] at org.springframework.cloud.stream.binder.EmbeddedHeaderUtils.oldExtractHeaders(EmbeddedHeaderUtils.java:154) ~[spring-cloud-stream-1.3.0.M2.jar!/:1.3.0.M2] at org.springframework.cloud.stream.binder.EmbeddedHeaderUtils.extractHeaders(EmbeddedHeaderUtils.java:115) ~[spring-cloud-stream-1.3.0.M2.jar!/:1.3.0.M2]

message is produced with kafka-console-producer.sh --broker-list localhost:9092 --topic bosstds and simply send line "test"

any suggestions ?

SCDF is a simple orchestration tool. All the action lies in the Spring Cloud Stream applications. It'd help if you can edit the post with the version of log-sink and Kafka that you're using. — Sabby Anandan

Vinicius Carvalho Vinicius Carvalho · Accepted Answer · 2017-10-01T15:36:26

SCS embed headers on kafka in order to get this working set header mode to raw. You need to do that when interfacing with external apps not using SCS

spring cloud dataflow input source kafka

2 Answers