New sample for Kafka native serialization with functions

kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplication.java

@@ -0,0 +1,24 @@
+package com.example.kafkanativeserialization;
+
+import java.util.function.Function;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+@SpringBootApplication
+public 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;
+		};
+	}
+
+}

kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/MyJsonDeserializer.java

@@ -0,0 +1,6 @@
+package com.example.kafkanativeserialization;
+
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+public class MyJsonDeserializer extends JsonDeserializer<Person> {
+}

kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/MyJsonSerializer.java

@@ -0,0 +1,19 @@
+package com.example.kafkanativeserialization;
+
+import org.apache.kafka.common.header.Headers;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.lang.Nullable;
+
+public class MyJsonSerializer extends JsonSerializer<Object> {
+
+	@Override
+	@Nullable
+	public byte[] serialize(String topic, Headers headers, @Nullable Object data) {
+		return super.serialize(topic, headers, data);
+	}
+	@Override
+	@Nullable
+	public byte[] serialize(String topic, @Nullable Object data) {
+		return super.serialize(topic, data);
+	}
+}

kafka-native-serialization/src/main/java/com/example/kafkanativeserialization/Person.java

@@ -0,0 +1,17 @@
+package com.example.kafkanativeserialization;
+
+public class Person {
+
+	private String name;
+	public Person() {
+	}
+	public Person(String name) {
+		this.setName(name);
+	}
+	public String getName() {
+		return name;
+	}
+	public void setName(String name) {
+		this.name = name;
+	}
+}

kafka-native-serialization/src/main/resources/application.yml

@@ -0,0 +1,17 @@
+spring.cloud:
+  stream:
+    function.definition: process
+    bindings:
+      process-in-0:
+        destination: topic1
+        group: group1
+      process-out-0:
+        destination: topic2
+        producer:
+          useNativeEncoding: true
+    kafka:
+      bindings:
+        process-out-0:
+          producer:
+            configuration:
+              value.serializer: com.example.kafkanativeserialization.MyJsonSerializer

kafka-native-serialization/src/test/java/com/example/kafkanativeserialization/KafkaNativeSerializationApplicationTests.java

@@ -0,0 +1,63 @@
+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.common.serialization.StringSerializer;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+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.utils.KafkaTestUtils;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+@RunWith(SpringRunner.class)
+public 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());
+	}
+
+	@Test
+	public void testSendReceive() {
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
+		senderProps.put("value.serializer", 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());
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		consumerProps.put("value.deserializer", 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(5));
+		consumer.commitSync();
+
+		assertThat(records.count()).isEqualTo(1);
+		assertThat(new String(records.iterator().next().value().getName())).isEqualTo("foo");
+	}
+}