diff --git a/samples/kafka-native-serialization/README.adoc b/samples/kafka-native-serialization/README.adoc index d5d4f3a54..cd0a17d97 100644 --- a/samples/kafka-native-serialization/README.adoc +++ b/samples/kafka-native-serialization/README.adoc @@ -1,11 +1,32 @@ -Spring Cloud Stream Kafka and native encoding -============================================== +== Spring Cloud Stream Kafka with native encoding -In this *Spring Cloud Stream* sample, we demonstrate native encoding with Kafka and functions. -Spring Cloud Stream will skip the regular message conversion on the outbound and let Kafka natively perform serialization. +This sample demonstrates native message encoding with Kafka. -There is a test provided where we verify the native conversion done by Kafka. This test uses `EmbeddedKafka`. -You can also run the application against a real Kafka cluster to see it in action. +By default, message encoding is performed transparently by the framework based on contentType. However, when native encoding is used, the default encoding is disabled and instead handled by the client library. It is the responsibility of the consumer to use an appropriate decoder to deserialize the inbound message and the responsibility of the producer to use an appropriate encoder to serialize the outbound message. + +More simply put, Spring Cloud Stream will skip the regular message conversion and let Kafka natively perform de/serialization. + +=== Application +The app consists of a simple function that takes in a string name and returns a Person object for the name. + +* The function input comes from `topic1` and uses the default framework encoding. +* The function output goes to `topic2` and uses native encoding (`JsonSerializer`). + +NOTE: By design, looking at the function code provides none of the above context. However, it can easily be seen where it is configured in link:./src/main/resources/application.yml[application.yml]. +=== Building +To build the app simply execute the following command: +[source,bash] +---- +./mvnw clean install +---- + +=== Running +The sample can be run directly but there is no source of events to the input topic. There is a link:./src/test/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplicationTests.java[@SpringBootTest] provided that exercises the functionality though. The test does the following: + +* produces a Kafka event w/ a simple string payload (the name) to the input topic +* consumes from the output topic using a custom `JsonDeserializer` + +This works because the function output was natively encoded with the `JsonSerializer` and therefore a JSON representation of the Person was sent to the output topic. diff --git a/samples/kafka-native-serialization/pom.xml b/samples/kafka-native-serialization/pom.xml index bb746ecb3..e7576ce3e 100644 --- a/samples/kafka-native-serialization/pom.xml +++ b/samples/kafka-native-serialization/pom.xml @@ -5,7 +5,6 @@ com.example kafka-native-serialization kafka-native-serialization - Demo project for Spring Boot org.springframework.cloud @@ -14,10 +13,6 @@ - - org.springframework.boot - spring-boot-starter-actuator - org.springframework.boot spring-boot-starter-web @@ -30,10 +25,6 @@ org.springframework.cloud spring-cloud-stream-binder-kafka - - org.springframework.kafka - spring-kafka - org.springframework.boot @@ -45,11 +36,6 @@ spring-kafka-test test - - org.junit.vintage - junit-vintage-engine - test - diff --git a/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplication.java b/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplication.java index 85c259ead..5ed4072d4 100644 --- a/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplication.java +++ b/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplication.java @@ -22,19 +22,15 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; -@SpringBootApplication -public class KafkaNativeSerializationApplication { +@SpringBootApplication(proxyBeanMethods = false) +class KafkaNativeSerializationApplication { public static void main(String[] args) { SpringApplication.run(KafkaNativeSerializationApplication.class, args); } @Bean - public Function process() { - return str -> { - Person item = new Person(str); - return item; - }; + Function process() { + return Person::new; } - } diff --git a/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/Person.java b/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/Person.java index 63f39183c..909dc319f 100644 --- a/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/Person.java +++ b/samples/kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/Person.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2022 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. @@ -19,15 +19,24 @@ package com.example.kafkanativeserialization; public class Person { private String name; + public Person() { } + public Person(String name) { - this.setName(name); + this.name = name; } + public String getName() { return name; } + public void setName(String name) { this.name = name; } + + @Override + public java.lang.String toString() { + return "Person(" + name + ")"; + } } diff --git a/samples/kafka-native-serialization/src/main/resources/application.yml b/samples/kafka-native-serialization/src/main/resources/application.yml index f9352322c..c9453c625 100644 --- a/samples/kafka-native-serialization/src/main/resources/application.yml +++ b/samples/kafka-native-serialization/src/main/resources/application.yml @@ -1,17 +1,20 @@ -spring.cloud: - stream: - function.definition: process - bindings: - process-in-0: - destination: topic1 - group: group1 - process-out-0: - destination: topic2 - producer: - useNativeEncoding: true - kafka: +spring: + cloud: + stream: bindings: + process-in-0: + destination: topic1 + group: group1 process-out-0: + destination: topic2 producer: - configuration: - value.serializer: org.springframework.kafka.support.serializer.JsonSerializer \ No newline at end of file + # this says 'I am using native encoding' + useNativeEncoding: true + kafka: + bindings: + process-out-0: + producer: + configuration: + # this says 'I am using this serializer for my native encoding' + value.serializer: org.springframework.kafka.support.serializer.JsonSerializer + diff --git a/samples/kafka-native-serialization/src/test/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplicationTests.java b/samples/kafka-native-serialization/src/test/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplicationTests.java index 0212eca07..087a504cc 100644 --- a/samples/kafka-native-serialization/src/test/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplicationTests.java +++ b/samples/kafka-native-serialization/src/test/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplicationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2022 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. @@ -16,64 +16,60 @@ package com.example.kafkanativeserialization; -import java.time.Duration; -import java.util.Collections; import java.util.Map; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.StringSerializer; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; -import org.springframework.test.context.junit4.SpringRunner; +import static com.example.kafkanativeserialization.KafkaNativeSerializationApplicationTests.INPUT_TOPIC; +import static com.example.kafkanativeserialization.KafkaNativeSerializationApplicationTests.OUTPUT_TOPIC; import static org.assertj.core.api.Assertions.assertThat; @SpringBootTest -@RunWith(SpringRunner.class) -public class KafkaNativeSerializationApplicationTests { +@EmbeddedKafka( + topics = { INPUT_TOPIC, OUTPUT_TOPIC }, + bootstrapServersProperty = "spring.cloud.stream.kafka.binder.brokers", + controlledShutdown = true) +class KafkaNativeSerializationApplicationTests { - private static final String INPUT_TOPIC = "topic1"; - private static final String OUTPUT_TOPIC = "topic2"; - private static final String GROUP_NAME = "nativeSerializationTest"; - - @ClassRule - public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, OUTPUT_TOPIC); - - @BeforeClass - public static void setup() { - System.setProperty("spring.cloud.stream.kafka.binder.brokers", embeddedKafka.getEmbeddedKafka().getBrokersAsString()); - } + public static final String INPUT_TOPIC = "topic1"; + public static final String OUTPUT_TOPIC = "topic2"; @Test - public void testSendReceive() { - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka()); - senderProps.put("value.serializer", StringSerializer.class); + void functionOutputPersonWithNativeEncoding(@Autowired EmbeddedKafkaBroker embeddedKafka) { + + // Send 'foo' to input topic + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); template.setDefaultTopic(INPUT_TOPIC); template.sendDefault("foo"); - Map consumerProps = KafkaTestUtils.consumerProps(GROUP_NAME, "false", embeddedKafka.getEmbeddedKafka()); + // Consume from output topic w/ custom JSON deserializer + Map consumerProps = KafkaTestUtils.consumerProps("nativeSerializationTest", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - consumerProps.put("value.deserializer", MyJsonDeserializer.class); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MyJsonDeserializer.class); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); - - Consumer consumer = cf.createConsumer(); - consumer.subscribe(Collections.singleton(OUTPUT_TOPIC)); - ConsumerRecords records = consumer.poll(Duration.ofSeconds(10)); - consumer.commitSync(); - - assertThat(records.count()).isEqualTo(1); - assertThat(new String(records.iterator().next().value().getName())).isEqualTo("foo"); + try (Consumer consumer = cf.createConsumer()) { + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, OUTPUT_TOPIC); + assertThat(KafkaTestUtils.getSingleRecord(consumer, OUTPUT_TOPIC, 10000L)) + .extracting(ConsumerRecord::value) + .extracting(Person::getName) + .isEqualTo("foo"); + } } }