From e549090787fc5e9ac27dc9811602f023e52b1e5b Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 23 Sep 2019 17:12:31 -0400 Subject: [PATCH] Restoring the avro tests for Kafka Streams --- .../pom.xml | 58 +-- .../PerRecordAvroContentTypeTests.java | 340 +++++++++--------- .../integration/utils/TestAvroSerializer.java | 94 ++--- .../MessageConverterDelegateSerdeTest.java | 120 +++---- 4 files changed, 306 insertions(+), 306 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/pom.xml b/spring-cloud-stream-binder-kafka-streams/pom.xml index eb90b24fa..d474fd99c 100644 --- a/spring-cloud-stream-binder-kafka-streams/pom.xml +++ b/spring-cloud-stream-binder-kafka-streams/pom.xml @@ -86,39 +86,39 @@ test - - - - - - - - - - - + + org.springframework.cloud + spring-cloud-schema-registry-client + test + + + org.apache.avro + avro + ${avro.version} + provided + - - - - - - - - - - - - - - - - - - + + org.apache.avro + avro-maven-plugin + ${avro.version} + + + generate-test-sources + + schema + + + ${project.basedir}/target/generated-test-sources + ${project.basedir}/target/generated-test-sources + ${project.basedir}/src/test/resources/avro + + + + \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/PerRecordAvroContentTypeTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/PerRecordAvroContentTypeTests.java index 1617d8302..f87eaf1dc 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/PerRecordAvroContentTypeTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/PerRecordAvroContentTypeTests.java @@ -14,173 +14,173 @@ * limitations under the License. */ -//package org.springframework.cloud.stream.binder.kafka.streams.integration; -// -//import java.io.IOException; -//import java.util.Map; -//import java.util.Random; -//import java.util.UUID; -// -//import com.example.Sensor; -//import org.apache.kafka.clients.consumer.Consumer; -//import org.apache.kafka.clients.consumer.ConsumerConfig; -//import org.apache.kafka.clients.consumer.ConsumerRecord; -//import org.apache.kafka.clients.producer.ProducerConfig; -//import org.apache.kafka.common.serialization.ByteArrayDeserializer; -//import org.apache.kafka.streams.KeyValue; -//import org.apache.kafka.streams.kstream.KStream; -//import org.junit.AfterClass; -//import org.junit.BeforeClass; -//import org.junit.ClassRule; -//import org.junit.Ignore; -//import org.junit.Test; -// -//import org.springframework.boot.SpringApplication; -//import org.springframework.boot.WebApplicationType; -//import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -//import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter; -//import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl; -//import org.springframework.cloud.stream.annotation.EnableBinding; -//import org.springframework.cloud.stream.annotation.Input; -//import org.springframework.cloud.stream.annotation.StreamListener; -//import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; -//import org.springframework.cloud.stream.binder.kafka.streams.integration.utils.TestAvroSerializer; -//import org.springframework.context.ConfigurableApplicationContext; -//import org.springframework.context.annotation.Bean; -//import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -//import org.springframework.kafka.core.DefaultKafkaProducerFactory; -//import org.springframework.kafka.core.KafkaTemplate; -//import org.springframework.kafka.test.EmbeddedKafkaBroker; -//import org.springframework.kafka.test.rule.EmbeddedKafkaRule; -//import org.springframework.kafka.test.utils.KafkaTestUtils; -//import org.springframework.messaging.Message; -//import org.springframework.messaging.converter.MessageConverter; -//import org.springframework.messaging.handler.annotation.SendTo; -//import org.springframework.messaging.support.MessageBuilder; -//import org.springframework.util.MimeTypeUtils; -// -//import static org.assertj.core.api.Assertions.assertThat; -// -// -///** -// * @author Soby Chacko -// */ -//public class PerRecordAvroContentTypeTests { -// -// @ClassRule -// public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, -// "received-sensors"); -// -// private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule -// .getEmbeddedKafka(); -// -// private static Consumer consumer; -// -// @BeforeClass -// public static void setUp() throws Exception { -// Map consumerProps = KafkaTestUtils.consumerProps("avro-ct-test", -// "false", embeddedKafka); -// -// // Receive the data as byte[] -// consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, -// ByteArrayDeserializer.class); -// -// consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); -// DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( -// consumerProps); -// consumer = cf.createConsumer(); -// embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "received-sensors"); -// } -// -// @AfterClass -// public static void tearDown() { -// consumer.close(); -// } -// -// @Test -// @Ignore -// public void testPerRecordAvroConentTypeAndVerifySerialization() throws Exception { -// SpringApplication app = new SpringApplication(SensorCountAvroApplication.class); -// app.setWebApplicationType(WebApplicationType.NONE); -// -// try (ConfigurableApplicationContext ignored = app.run("--server.port=0", -// "--spring.jmx.enabled=false", -// "--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=false", -// "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=false", -// "--spring.cloud.stream.bindings.input.destination=sensors", -// "--spring.cloud.stream.bindings.output.destination=received-sensors", -// "--spring.cloud.stream.bindings.output.contentType=application/avro", -// "--spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=per-record-avro-contentType-test", -// "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", -// "--spring.cloud.stream.kafka.streams.binder.brokers=" -// + embeddedKafka.getBrokersAsString())) { -// -// Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); -// // Use a custom avro test serializer -// senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, -// TestAvroSerializer.class); -// DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( -// senderProps); -// try { -// KafkaTemplate template = new KafkaTemplate<>(pf, true); -// -// Random random = new Random(); -// Sensor sensor = new Sensor(); -// sensor.setId(UUID.randomUUID().toString() + "-v1"); -// sensor.setAcceleration(random.nextFloat() * 10); -// sensor.setVelocity(random.nextFloat() * 100); -// sensor.setTemperature(random.nextFloat() * 50); -// // Send with avro content type set. -// Message message = MessageBuilder.withPayload(sensor) -// .setHeader("contentType", "application/avro").build(); -// template.setDefaultTopic("sensors"); -// template.send(message); -// -// // Serialized byte[] ^^ is received by the binding process and deserialzed -// // it using avro converter. -// // Then finally, the data will be output to a return topic as byte[] -// // (using the same avro converter). -// -// // Receive the byte[] from return topic -// ConsumerRecord cr = KafkaTestUtils -// .getSingleRecord(consumer, "received-sensors"); -// final byte[] value = cr.value(); -// -// // Convert the byte[] received back to avro object and verify that it is -// // the same as the one we sent ^^. -// AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(); -// -// Message receivedMessage = MessageBuilder.withPayload(value) -// .setHeader("contentType", -// MimeTypeUtils.parseMimeType("application/avro")) -// .build(); -// Sensor messageConverted = (Sensor) avroSchemaMessageConverter -// .fromMessage(receivedMessage, Sensor.class); -// assertThat(messageConverted).isEqualTo(sensor); -// } -// finally { -// pf.destroy(); -// } -// } -// } -// -// @EnableBinding(KafkaStreamsProcessor.class) -// @EnableAutoConfiguration -// static class SensorCountAvroApplication { -// -// @StreamListener -// @SendTo("output") -// public KStream process(@Input("input") KStream input) { -// // return the same Sensor object unchanged so that we can do test -// // verifications -// return input.map(KeyValue::new); -// } -// -// @Bean -// public MessageConverter sensorMessageConverter() throws IOException { -// return new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl()); -// } -// -// } -// -//} +package org.springframework.cloud.stream.binder.kafka.streams.integration; + +import java.io.IOException; +import java.util.Map; +import java.util.Random; +import java.util.UUID; + +import com.example.Sensor; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Ignore; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter; +import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; +import org.springframework.cloud.stream.binder.kafka.streams.integration.utils.TestAvroSerializer; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.MimeTypeUtils; + +import static org.assertj.core.api.Assertions.assertThat; + + +/** + * @author Soby Chacko + */ +public class PerRecordAvroContentTypeTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "received-sensors"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule + .getEmbeddedKafka(); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("avro-ct-test", + "false", embeddedKafka); + + // Receive the data as byte[] + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + ByteArrayDeserializer.class); + + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( + consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "received-sensors"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + @Ignore + public void testPerRecordAvroConentTypeAndVerifySerialization() throws Exception { + SpringApplication app = new SpringApplication(SensorCountAvroApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext ignored = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=false", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=false", + "--spring.cloud.stream.bindings.input.destination=sensors", + "--spring.cloud.stream.bindings.output.destination=received-sensors", + "--spring.cloud.stream.bindings.output.contentType=application/avro", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=per-record-avro-contentType-test", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString())) { + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + // Use a custom avro test serializer + senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + TestAvroSerializer.class); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( + senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + + Random random = new Random(); + Sensor sensor = new Sensor(); + sensor.setId(UUID.randomUUID().toString() + "-v1"); + sensor.setAcceleration(random.nextFloat() * 10); + sensor.setVelocity(random.nextFloat() * 100); + sensor.setTemperature(random.nextFloat() * 50); + // Send with avro content type set. + Message message = MessageBuilder.withPayload(sensor) + .setHeader("contentType", "application/avro").build(); + template.setDefaultTopic("sensors"); + template.send(message); + + // Serialized byte[] ^^ is received by the binding process and deserialzed + // it using avro converter. + // Then finally, the data will be output to a return topic as byte[] + // (using the same avro converter). + + // Receive the byte[] from return topic + ConsumerRecord cr = KafkaTestUtils + .getSingleRecord(consumer, "received-sensors"); + final byte[] value = cr.value(); + + // Convert the byte[] received back to avro object and verify that it is + // the same as the one we sent ^^. + AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(); + + Message receivedMessage = MessageBuilder.withPayload(value) + .setHeader("contentType", + MimeTypeUtils.parseMimeType("application/avro")) + .build(); + Sensor messageConverted = (Sensor) avroSchemaMessageConverter + .fromMessage(receivedMessage, Sensor.class); + assertThat(messageConverted).isEqualTo(sensor); + } + finally { + pf.destroy(); + } + } + } + + @EnableBinding(KafkaStreamsProcessor.class) + @EnableAutoConfiguration + static class SensorCountAvroApplication { + + @StreamListener + @SendTo("output") + public KStream process(@Input("input") KStream input) { + // return the same Sensor object unchanged so that we can do test + // verifications + return input.map(KeyValue::new); + } + + @Bean + public MessageConverter sensorMessageConverter() throws IOException { + return new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl()); + } + + } + +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/utils/TestAvroSerializer.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/utils/TestAvroSerializer.java index 752faeefd..6bbf3180e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/utils/TestAvroSerializer.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/utils/TestAvroSerializer.java @@ -14,50 +14,50 @@ * limitations under the License. */ -//package org.springframework.cloud.stream.binder.kafka.streams.integration.utils; -// -//import java.util.HashMap; -//import java.util.Map; -// -//import org.apache.kafka.common.serialization.Serializer; -// -//import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter; -//import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl; -//import org.springframework.messaging.Message; -//import org.springframework.messaging.MessageHeaders; -//import org.springframework.messaging.support.MessageBuilder; -// -///** -// * Custom avro serializer intended to be used for testing only. -// * -// * @param Target type to serialize -// * @author Soby Chacko -// */ -//public class TestAvroSerializer implements Serializer { -// -// public TestAvroSerializer() { -// } -// -// @Override -// public void configure(Map configs, boolean isKey) { -// -// } -// -// @Override -// public byte[] serialize(String topic, S data) { -// AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl()); -// Message message = MessageBuilder.withPayload(data).build(); -// Map headers = new HashMap<>(message.getHeaders()); -// headers.put(MessageHeaders.CONTENT_TYPE, "application/avro"); -// MessageHeaders messageHeaders = new MessageHeaders(headers); -// final Object payload = avroSchemaMessageConverter -// .toMessage(message.getPayload(), messageHeaders).getPayload(); -// return (byte[]) payload; -// } -// -// @Override -// public void close() { -// -// } -// -//} +package org.springframework.cloud.stream.binder.kafka.streams.integration.utils; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.common.serialization.Serializer; + +import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter; +import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; + +/** + * Custom avro serializer intended to be used for testing only. + * + * @param Target type to serialize + * @author Soby Chacko + */ +public class TestAvroSerializer implements Serializer { + + public TestAvroSerializer() { + } + + @Override + public void configure(Map configs, boolean isKey) { + + } + + @Override + public byte[] serialize(String topic, S data) { + AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl()); + Message message = MessageBuilder.withPayload(data).build(); + Map headers = new HashMap<>(message.getHeaders()); + headers.put(MessageHeaders.CONTENT_TYPE, "application/avro"); + MessageHeaders messageHeaders = new MessageHeaders(headers); + final Object payload = avroSchemaMessageConverter + .toMessage(message.getPayload(), messageHeaders).getPayload(); + return (byte[]) payload; + } + + @Override + public void close() { + + } + +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/MessageConverterDelegateSerdeTest.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/MessageConverterDelegateSerdeTest.java index 099902ec6..4bface439 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/MessageConverterDelegateSerdeTest.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/MessageConverterDelegateSerdeTest.java @@ -14,63 +14,63 @@ * limitations under the License. */ -//package org.springframework.cloud.stream.binder.kafka.streams.serde; -// -//import java.util.ArrayList; -//import java.util.HashMap; -//import java.util.List; -//import java.util.Map; -//import java.util.Random; -//import java.util.UUID; -// -//import com.example.Sensor; -//import com.fasterxml.jackson.databind.ObjectMapper; -//import org.junit.Ignore; -//import org.junit.Test; -// -//import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter; -//import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl; -//import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -//import org.springframework.messaging.converter.MessageConverter; -// -//import static org.assertj.core.api.Assertions.assertThat; -// -///** -// * Refer {@link MessageConverterDelegateSerde} for motivations. -// * -// * @author Soby Chacko -// */ -//public class MessageConverterDelegateSerdeTest { -// -// @Test -// @SuppressWarnings("unchecked") -// @Ignore -// public void testCompositeNonNativeSerdeUsingAvroContentType() { -// Random random = new Random(); -// Sensor sensor = new Sensor(); -// sensor.setId(UUID.randomUUID().toString() + "-v1"); -// sensor.setAcceleration(random.nextFloat() * 10); -// sensor.setVelocity(random.nextFloat() * 100); -// sensor.setTemperature(random.nextFloat() * 50); -// -// List messageConverters = new ArrayList<>(); -// messageConverters.add(new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl())); -// CompositeMessageConverterFactory compositeMessageConverterFactory = new CompositeMessageConverterFactory( -// messageConverters, new ObjectMapper()); -// MessageConverterDelegateSerde messageConverterDelegateSerde = new MessageConverterDelegateSerde( -// compositeMessageConverterFactory.getMessageConverterForAllRegistered()); -// -// Map configs = new HashMap<>(); -// configs.put("valueClass", Sensor.class); -// configs.put("contentType", "application/avro"); -// messageConverterDelegateSerde.configure(configs, false); -// final byte[] serialized = messageConverterDelegateSerde.serializer().serialize(null, -// sensor); -// -// final Object deserialized = messageConverterDelegateSerde.deserializer() -// .deserialize(null, serialized); -// -// assertThat(deserialized).isEqualTo(sensor); -// } -// -//} +package org.springframework.cloud.stream.binder.kafka.streams.serde; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.UUID; + +import com.example.Sensor; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.Ignore; +import org.junit.Test; + +import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter; +import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.messaging.converter.MessageConverter; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Refer {@link MessageConverterDelegateSerde} for motivations. + * + * @author Soby Chacko + */ +public class MessageConverterDelegateSerdeTest { + + @Test + @SuppressWarnings("unchecked") + @Ignore + public void testCompositeNonNativeSerdeUsingAvroContentType() { + Random random = new Random(); + Sensor sensor = new Sensor(); + sensor.setId(UUID.randomUUID().toString() + "-v1"); + sensor.setAcceleration(random.nextFloat() * 10); + sensor.setVelocity(random.nextFloat() * 100); + sensor.setTemperature(random.nextFloat() * 50); + + List messageConverters = new ArrayList<>(); + messageConverters.add(new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl())); + CompositeMessageConverterFactory compositeMessageConverterFactory = new CompositeMessageConverterFactory( + messageConverters, new ObjectMapper()); + MessageConverterDelegateSerde messageConverterDelegateSerde = new MessageConverterDelegateSerde( + compositeMessageConverterFactory.getMessageConverterForAllRegistered()); + + Map configs = new HashMap<>(); + configs.put("valueClass", Sensor.class); + configs.put("contentType", "application/avro"); + messageConverterDelegateSerde.configure(configs, false); + final byte[] serialized = messageConverterDelegateSerde.serializer().serialize(null, + sensor); + + final Object deserialized = messageConverterDelegateSerde.deserializer() + .deserialize(null, serialized); + + assertThat(deserialized).isEqualTo(sensor); + } + +}