Polish kafka-native-serialization sample

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.
 

samples/kafka-native-serialization/pom.xml

@@ -5,7 +5,6 @@
 	<groupId>com.example</groupId>
 	<artifactId>kafka-native-serialization</artifactId>
 	<name>kafka-native-serialization</name>
-	<description>Demo project for Spring Boot</description>
 
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
@@ -14,10 +13,6 @@
 	</parent>
 
 	<dependencies>
-		<dependency>
-			<groupId>org.springframework.boot</groupId>
-			<artifactId>spring-boot-starter-actuator</artifactId>
-		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-web</artifactId>
@@ -30,10 +25,6 @@
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.kafka</groupId>
-			<artifactId>spring-kafka</artifactId>
-		</dependency>
 
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
@@ -45,11 +36,6 @@
 			<artifactId>spring-kafka-test</artifactId>
 			<scope>test</scope>
 		</dependency>
-		<dependency>
-			<groupId>org.junit.vintage</groupId>
-			<artifactId>junit-vintage-engine</artifactId>
-			<scope>test</scope>
-		</dependency>
 	</dependencies>
 
 </project>

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<String, Person> process() {
-		return str -> {
-			Person item = new Person(str);
-			return item;
-		};
+	Function<String, Person> process() {
+		return Person::new;
 	}
-
 }

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

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
+            # 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
+

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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
-		senderProps.put("value.serializer", StringSerializer.class);
+	void functionOutputPersonWithNativeEncoding(@Autowired EmbeddedKafkaBroker embeddedKafka) {
+
+		// Send 'foo' to input topic
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
 		DefaultKafkaProducerFactory<byte[], String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<byte[], String> template = new KafkaTemplate<>(pf, true);
 		template.setDefaultTopic(INPUT_TOPIC);
 		template.sendDefault("foo");
 
-		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(GROUP_NAME, "false", embeddedKafka.getEmbeddedKafka());
+		// Consume from output topic w/ custom JSON deserializer
+		Map<String, Object> 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<byte[], Person> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
-
-		Consumer<byte[], Person> consumer = cf.createConsumer();
-		consumer.subscribe(Collections.singleton(OUTPUT_TOPIC));
-		ConsumerRecords<byte[], Person> 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<byte[], Person> consumer = cf.createConsumer()) {
+			embeddedKafka.consumeFromAnEmbeddedTopic(consumer, OUTPUT_TOPIC);
+			assertThat(KafkaTestUtils.getSingleRecord(consumer, OUTPUT_TOPIC, 10000L))
+				.extracting(ConsumerRecord::value)
+				.extracting(Person::getName)
+				.isEqualTo("foo");
+		}
 	}
 }