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);
+ }
+
+}