diff --git a/pom.xml b/pom.xml index 6e293443a..a8d6060fb 100644 --- a/pom.xml +++ b/pom.xml @@ -129,6 +129,12 @@ + + org.springframework.cloud + spring-cloud-stream-schema + ${spring-cloud-stream.version} + test + diff --git a/spring-cloud-stream-binder-kafka-streams/pom.xml b/spring-cloud-stream-binder-kafka-streams/pom.xml index 4880ddd6f..3f03769ec 100644 --- a/spring-cloud-stream-binder-kafka-streams/pom.xml +++ b/spring-cloud-stream-binder-kafka-streams/pom.xml @@ -13,6 +13,10 @@ 2.1.0.BUILD-SNAPSHOT + + 1.8.2 + + org.springframework.cloud @@ -76,5 +80,40 @@ test test + + + org.springframework.cloud + spring-cloud-stream-schema + test + + + org.apache.avro + avro + ${avro.version} + provided + + + + + + org.apache.avro + avro-maven-plugin + ${avro.version} + + + generate-sources + + schema + protocol + idl-protocol + + + src/test/resources/avro + + + + + + \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index ea7d415fa..fb25bb8cc 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -36,6 +36,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; +import org.springframework.cloud.stream.binder.kafka.streams.serde.CompositeNonNativeSerde; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; import org.springframework.cloud.stream.config.BindingServiceConfiguration; @@ -161,6 +162,11 @@ public class KafkaStreamsBinderSupportAutoConfiguration { KafkaStreamsBindingInformationCatalogue, binderConfigurationProperties); } + @Bean + public CompositeNonNativeSerde compositeNonNativeSerde(CompositeMessageConverterFactory compositeMessageConverterFactory) { + return new CompositeNonNativeSerde(compositeMessageConverterFactory); + } + @Bean public KStreamBoundElementFactory kStreamBoundElementFactory(BindingServiceProperties bindingServiceProperties, KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java index fefef6da4..fddcef7e0 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java @@ -21,6 +21,8 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.processor.Processor; @@ -92,7 +94,7 @@ public class KafkaStreamsMessageConversionDelegate { } /** - * Deserialize incoming {@link KStream} based on contentType. + * Deserialize incoming {@link KStream} based on content type. * * @param valueClass on KStream value * @param bindingTarget inbound KStream target @@ -101,6 +103,9 @@ public class KafkaStreamsMessageConversionDelegate { @SuppressWarnings("unchecked") public KStream deserializeOnInbound(Class valueClass, KStream bindingTarget) { MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered(); + final PerRecordContentTypeHolder perRecordContentTypeHolder = new PerRecordContentTypeHolder(); + + resolvePerRecordContentType(bindingTarget, perRecordContentTypeHolder); //Deserialize using a branching strategy KStream[] branch = bindingTarget.branch( @@ -113,16 +118,26 @@ public class KafkaStreamsMessageConversionDelegate { if (o2 != null) { if (valueClass.isAssignableFrom(o2.getClass())) { keyValueThreadLocal.set(new KeyValue<>(o, o2)); - } else if (o2 instanceof Message) { - if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) { - keyValueThreadLocal.set(new KeyValue<>(o, ((Message) o2).getPayload())); - } else { - convertAndSetMessage(o, valueClass, messageConverter, (Message) o2); + } + else if (o2 instanceof Message) { + Message m1 = (Message) o2; + if (perRecordContentTypeHolder.contentType != null) { + m1 = MessageBuilder.fromMessage(m1).setHeader("contentType", perRecordContentTypeHolder.contentType).build(); } - } else if (o2 instanceof String || o2 instanceof byte[]) { - Message message = MessageBuilder.withPayload(o2).build(); + + if (valueClass.isAssignableFrom(m1.getPayload().getClass())) { + keyValueThreadLocal.set(new KeyValue<>(o, m1.getPayload())); + } + else { + convertAndSetMessage(o, valueClass, messageConverter, m1); + } + } + else if (o2 instanceof String || o2 instanceof byte[]) { + Message message = perRecordContentTypeHolder.contentType != null ? MessageBuilder.withPayload(o2) + .setHeader("contentType", perRecordContentTypeHolder.contentType).build() : MessageBuilder.withPayload(o2).build(); convertAndSetMessage(o, valueClass, messageConverter, message); - } else { + } + else { keyValueThreadLocal.set(new KeyValue<>(o, o2)); } isValidRecord = true; @@ -150,6 +165,45 @@ public class KafkaStreamsMessageConversionDelegate { }); } + private static class PerRecordContentTypeHolder { + + String contentType; + + void setContentType(String contentType) { + this.contentType = contentType; + } + } + + @SuppressWarnings("unchecked") + private void resolvePerRecordContentType(KStream outboundBindTarget, PerRecordContentTypeHolder perRecordContentTypeHolder) { + outboundBindTarget.process(() -> new Processor() { + + ProcessorContext context; + + @Override + public void init(ProcessorContext context) { + this.context = context; + } + + @Override + public void process(Object key, Object value) { + final Headers headers = context.headers(); + final Iterable
contentTypes = headers.headers("contentType"); + if (contentTypes != null && contentTypes.iterator().hasNext()) { + final String contentType = new String(contentTypes.iterator().next().value()); + //remove leading and trailing quotes + final String cleanContentType = StringUtils.replace(contentType, "\"", ""); + perRecordContentTypeHolder.setContentType(cleanContentType); + } + } + + @Override + public void close() { + + } + }); + } + private void convertAndSetMessage(Object o, Class valueClass, MessageConverter messageConverter, Message msg) { Object messageConverted = messageConverter.fromMessage(msg, valueClass); if (messageConverted == null) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java new file mode 100644 index 000000000..995e97615 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java @@ -0,0 +1,210 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams.serde; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serializer; + +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; + +/** + * A {@link Serde} implementation that wraps the list of {@link MessageConverter}s + * from {@link CompositeMessageConverterFactory}. + * + * The primary motivation for this class is to provide an avro based {@link Serde} that is + * compatible with the schema registry that Spring Cloud Stream provides. When using the + * schema registry support from Spring Cloud Stream in a Kafka Streams binder based application, + * the applications can deserialize the incoming Kafka Streams records using the built in + * Avro {@link MessageConverter}. However, this same message conversion approach will not work + * downstream in other operations in the topology for Kafka Streams as some of them needs a + * {@link Serde} instance that can talk to the Spring Cloud Stream provided Schema Registry. + * This implementation will solve that problem. + * + * Only Avro and JSON based converters are exposed as binder provided {@link Serde} implementations currently. + * + * Users of this class must call the {@link CompositeNonNativeSerde#configure(Map, boolean)} method + * to configure the {@link Serde} object. At the very least the configuration map must include a key + * called "valueClass" to indicate the type of the target object for deserialization. If any other + * content type other than JSON is needed (only Avro is available now other than JSON), that needs + * to be included in the configuration map with the key "contentType". For example, + * + * 
+ * Map config = new HashMap<>();
+ * config.put("valueClass", Foo.class);
+ * config.put("contentType", "application/avro");
+ *
+ * + * Then use the above map when calling the configure method. + * + * This class is only intended to be used when writing a Spring Cloud Stream Kafka Streams application + * that uses Spring Cloud Stream schema registry for schema evolution. + * + * An instance of this class is provided as a bean by the binder configuration and typically the applications + * can autowire that bean. This is the expected usage pattern of this class. + * + * @author Soby Chacko + */ +public class CompositeNonNativeSerde implements Serde { + + private static final String CONTENT_TYPE_HEADER = "contentType"; + + private static final String VALUE_CLASS_HEADER = "valueClass"; + + private static final String AVRO_FORMAT = "avro"; + + private static final MimeType DEFAULT_AVRO_MIME_TYPE = new MimeType("application", "*+" + AVRO_FORMAT); + + private final CompositeNonNativeDeserializer compositeNonNativeDeserializer; + + private final CompositeNonNativeSerializer compositeNonNativeSerializer; + + public CompositeNonNativeSerde(CompositeMessageConverterFactory compositeMessageConverterFactory) { + this.compositeNonNativeDeserializer = new CompositeNonNativeDeserializer<>(compositeMessageConverterFactory); + this.compositeNonNativeSerializer = new CompositeNonNativeSerializer<>(compositeMessageConverterFactory); + } + + @Override + public void configure(Map configs, boolean isKey) { + this.compositeNonNativeDeserializer.configure(configs, isKey); + this.compositeNonNativeSerializer.configure(configs, isKey); + } + + @Override + public void close() { + //No-op + } + + @Override + public Serializer serializer() { + return this.compositeNonNativeSerializer; + } + + @Override + public Deserializer deserializer() { + return this.compositeNonNativeDeserializer; + } + + private static MimeType resolveMimeType(Map configs) { + if (configs.containsKey(CONTENT_TYPE_HEADER)){ + String contentType = (String)configs.get(CONTENT_TYPE_HEADER); + if (DEFAULT_AVRO_MIME_TYPE.equals(MimeTypeUtils.parseMimeType(contentType))) { + return DEFAULT_AVRO_MIME_TYPE; + } + else if(contentType.contains("avro")) { + return MimeTypeUtils.parseMimeType("application/avro"); + } + else { + return new MimeType("application", "json", StandardCharsets.UTF_8); + } + } + else { + return new MimeType("application", "json", StandardCharsets.UTF_8); + } + } + + /** + * Custom {@link Deserializer} that uses the {@link CompositeMessageConverterFactory}. + * + * @param Parameterized target type for deserialization + */ + private static class CompositeNonNativeDeserializer implements Deserializer { + + private final MessageConverter messageConverter; + + private MimeType mimeType; + + private Class valueClass; + + CompositeNonNativeDeserializer(CompositeMessageConverterFactory compositeMessageConverterFactory) { + this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered(); + } + + @Override + public void configure(Map configs, boolean isKey) { + Assert.isTrue(configs.containsKey(VALUE_CLASS_HEADER), "Deserializers must provide a configuration for valueClass."); + final Object valueClass = configs.get(VALUE_CLASS_HEADER); + Assert.isTrue(valueClass instanceof Class, "Deserializers must provide a valid value for valueClass."); + this.valueClass = (Class) valueClass; + this.mimeType = resolveMimeType(configs); + } + + @SuppressWarnings("unchecked") + @Override + public U deserialize(String topic, byte[] data) { + Message message = MessageBuilder.withPayload(data) + .setHeader(CONTENT_TYPE_HEADER, this.mimeType.toString()).build(); + U messageConverted = (U)messageConverter.fromMessage(message, this.valueClass); + if (messageConverted == null) { + throw new IllegalStateException("Deserialization failed."); + } + return messageConverted; + } + + @Override + public void close() { + //No-op + } + } + + /** + * Custom {@link Serializer} that uses the {@link CompositeMessageConverterFactory}. + * + * @param Parameterized type for serialization + */ + private static class CompositeNonNativeSerializer implements Serializer { + + private final MessageConverter messageConverter; + private MimeType mimeType; + + CompositeNonNativeSerializer(CompositeMessageConverterFactory compositeMessageConverterFactory) { + this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered(); + } + + @Override + public void configure(Map configs, boolean isKey) { + this.mimeType = resolveMimeType(configs); + } + + @Override + public byte[] serialize(String topic, V data) { + Message message = MessageBuilder.withPayload(data).build(); + Map headers = new HashMap<>(message.getHeaders()); + headers.put(MessageHeaders.CONTENT_TYPE, this.mimeType.toString()); + MessageHeaders messageHeaders = new MessageHeaders(headers); + final Object payload = messageConverter.toMessage(message.getPayload(), + messageHeaders).getPayload(); + return (byte[])payload; + } + + @Override + public void close() { + //No-op + } + } +} 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 new file mode 100644 index 000000000..cbf629f86 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/PerRecordAvroContentTypeTests.java @@ -0,0 +1,165 @@ +/* + * Copyright 2017-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * 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.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +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.annotation.StreamMessageConverter; +import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; +import org.springframework.cloud.stream.binder.kafka.streams.integration.utils.TestAvroSerializer; +import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter; +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 + 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.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 + @StreamMessageConverter + public MessageConverter sensorMessageConverter() throws IOException { + return new AvroSchemaMessageConverter(); + } + } +} 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 new file mode 100644 index 000000000..1a22dab6b --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/utils/TestAvroSerializer.java @@ -0,0 +1,61 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * 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.stream.schema.avro.AvroSchemaMessageConverter; +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. + * + * @author Soby Chacko + * + * @param Target type to serialize + */ +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(); + 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/CompositeNonNativeSerdeTest.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerdeTest.java new file mode 100644 index 000000000..42573fd79 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerdeTest.java @@ -0,0 +1,69 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * 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.Test; + +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter; +import org.springframework.messaging.converter.MessageConverter; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Refer {@link CompositeNonNativeSerde} for motivations. + * + * @author Soby Chacko + */ +public class CompositeNonNativeSerdeTest { + + @Test + @SuppressWarnings("unchecked") + 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()); + CompositeMessageConverterFactory compositeMessageConverterFactory = + new CompositeMessageConverterFactory(messageConverters, new ObjectMapper()); + CompositeNonNativeSerde compositeNonNativeSerde = new CompositeNonNativeSerde(compositeMessageConverterFactory); + + Map configs = new HashMap<>(); + configs.put("valueClass", Sensor.class); + configs.put("contentType", "application/avro"); + compositeNonNativeSerde.configure(configs, false); + final byte[] serialized = compositeNonNativeSerde.serializer().serialize(null, sensor); + + final Object deserialized = compositeNonNativeSerde.deserializer().deserialize(null, serialized); + + assertThat(deserialized).isEqualTo(sensor); + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/resources/avro/sensor.avsc b/spring-cloud-stream-binder-kafka-streams/src/test/resources/avro/sensor.avsc new file mode 100644 index 000000000..c0e060d3d --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/resources/avro/sensor.avsc @@ -0,0 +1,11 @@ +{ + "namespace" : "com.example", + "type" : "record", + "name" : "Sensor", + "fields" : [ + {"name":"id","type":"string"}, + {"name":"temperature", "type":"float", "default":0.0}, + {"name":"acceleration", "type":"float","default":0.0}, + {"name":"velocity","type":"float","default":0.0} + ] +}