diff --git a/spring-cloud-stream-binder-kafka-streams/pom.xml b/spring-cloud-stream-binder-kafka-streams/pom.xml index d474fd99c..eb90b24fa 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 00c745a3b..1617d8302 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,171 +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.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 - 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 6bbf3180e..752faeefd 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 82c68ff0b..099902ec6 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,61 +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.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") - 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); +// } +// +//}