Ndtyjky

Any ideas would be appreciated, I am trying write test as;

@ExtendWith(SpringExtension.class)

@EmbeddedKafka(count = 1, controlledShutdown = true, topics = { "input", "output" }, brokerProperties = { "broker.id=2",

    "listeners=PLAINTEXT://127.0.0.1:9092" })

class BindingTest {



@Autowired

private ApplicationContext applicationContext;



@Autowired

private EmbeddedKafkaBroker embeddedKafka;



@Autowired

private CustomBindings cBindings;



/**

 * @throws java.lang.Exception

 */

@BeforeEach

void setUp() throws Exception {

}



/**

 * @throws java.lang.Exception

 */

@AfterEach

void tearDown() throws Exception {

    embeddedKafka.getKafkaServers().forEach(b -> b.shutdown());

}



@Test

void test0() {

    String KEY = "KEY";

    String testMessage = "TESTMESSAGE";

    Message<String> message = MessageBuilder.withPayload(testMessage)

            .setHeader(KafkaHeaders.MESSAGE_KEY, KEY).build();

    cBindings.output().send(message);



}



@SpringBootApplication

@EnableBinding(CustomBindings.class)

public static class BindingApplication {



}

}

and

spring.cloud.stream.bindings.output.destination=output
spring.cloud.stream.bindings.output.contentType=application/json

spring.cloud.stream.bindings.output.producer.header-mode=raw
spring.cloud.stream.bindings.output.producer.use-native-encoding=true

spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde==org.apache.kafka.common.serialization.StringSerializer
spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde==org.apache.kafka.common.serialization.StringSerializer

Still gets

error occurred in message handler [org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$ProducerConfigurationMessageHandler@71f0806b]; nested exception is org.apache.kafka.common.errors.SerializationException: Can't convert value of class [B to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer, failedMessage=GenericMessage [payload=byte[8], headers={id=c91897bc-2e4e-0a74-bc05-17fb31b690f6, kafka_messageKey=KEY, contentType=application/json, timestamp=1542295539746}]

Which doesnt make sense to me

From the looks of it, this is not a Kafka Streams application, but a regular Spring Cloud Stream application with the Kafka binder. Therefore, you don't need these two properties.

spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde==org.apache.kafka.common.serialization.StringSerializer spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde==org.apache.kafka.common.serialization.StringSerializer


In addition, in order to fix your error, you need to remove this line from your config: spring.cloud.stream.bindings.output.producer.use-native-encoding=true.

By setting native encoding to true, you are asking Kafka to do the serialization which is going to rely on the default ByteArraySerializer. If you really intended native serialization, you need to set the appropriate value serializer (StringSerializer). But since this is a test, I suggest you remove this property and see if your test passes.