Add a test to json serialize both a key and value in a stream.

Author: elliotkennedy

spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsJsonSerializationTests.java

@@ -0,0 +1,207 @@
+/*
+ * 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.kafka.kstream;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+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.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.Consumed;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Printed;
+import org.apache.kafka.streams.kstream.Produced;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafkaStreams;
+import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.kafka.kstream.KafkaStreamsJsonSerializationTests.JsonObjectKey.jsonObjectKeySerde;
+import static org.springframework.kafka.kstream.KafkaStreamsJsonSerializationTests.JsonObjectValue.jsonObjectValueSerde;
+
+/**
+ * @author Elliot Kennedy
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+@EmbeddedKafka(partitions = 1,
+		topics = {KafkaStreamsJsonSerializationTests.INPUT_TOPIC,
+				KafkaStreamsJsonSerializationTests.OUTPUT_TOPIC})
+public class KafkaStreamsJsonSerializationTests {
+
+	public static final String INPUT_TOPIC = "input-topic";
+	public static final String OUTPUT_TOPIC = "output-topic";
+
+	@Autowired
+	private KafkaTemplate<JsonObjectKey, JsonObjectValue> template;
+
+	@Autowired
+	private KafkaEmbedded kafkaEmbedded;
+
+	private Consumer<JsonObjectKey, JsonObjectValue> outputTopicConsumer;
+
+	@Before
+	public void setup() throws Exception {
+		Map<String, Object> consumerProps =
+				KafkaTestUtils.consumerProps(UUID.randomUUID().toString(), "false", this.kafkaEmbedded);
+		consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10000);
+
+		DefaultKafkaConsumerFactory<JsonObjectKey, JsonObjectValue> kafkaConsumerFactory =
+				new DefaultKafkaConsumerFactory<>(consumerProps, jsonObjectKeySerde().deserializer(), jsonObjectValueSerde().deserializer());
+		outputTopicConsumer = kafkaConsumerFactory.createConsumer();
+		kafkaEmbedded.consumeFromAnEmbeddedTopic(outputTopicConsumer, OUTPUT_TOPIC);
+	}
+
+	@Test
+	public void testJsonSerialiazation() throws Exception {
+		template.sendDefault(new JsonObjectKey(25), new JsonObjectValue("twenty-five"));
+
+		ConsumerRecords<JsonObjectKey, JsonObjectValue> outputTopicRecords = KafkaTestUtils.getRecords(outputTopicConsumer);
+
+		assertThat(outputTopicRecords.count()).isEqualTo(1);
+		ConsumerRecord<JsonObjectKey, JsonObjectValue> output = outputTopicRecords.iterator().next();
+		assertThat(output.key()).isInstanceOf(JsonObjectKey.class);
+		assertThat(output.key().getKey()).isEqualTo(25);
+		assertThat(output.value()).isInstanceOf(JsonObjectValue.class);
+		assertThat(output.value().getValue()).isEqualTo("twenty-five");
+	}
+
+	public static class JsonObjectKey {
+
+		private final Integer key;
+
+		@JsonCreator
+		public JsonObjectKey(@JsonProperty(value = "key", required = true) Integer key) {
+			this.key = key;
+		}
+
+		public Integer getKey() {
+			return key;
+		}
+
+		public static Serde<JsonObjectKey> jsonObjectKeySerde() {
+			return new JsonSerde<>(JsonObjectKey.class);
+		}
+
+		@Override
+		public String toString() {
+			return "JsonObjectKey{" +
+					"key=" + key +
+					'}';
+		}
+	}
+
+	public static class JsonObjectValue {
+
+		private final String value;
+
+		@JsonCreator
+		public JsonObjectValue(@JsonProperty(value = "value", required = true) String value) {
+			this.value = value;
+		}
+
+		public String getValue() {
+			return value;
+		}
+
+		public static Serde<JsonObjectValue> jsonObjectValueSerde() {
+			return new JsonSerde<>(JsonObjectValue.class);
+		}
+
+		@Override
+		public String toString() {
+			return "JsonObjectValue{" +
+					"value='" + value + '\'' +
+					'}';
+		}
+	}
+
+	@Configuration
+	@EnableKafkaStreams
+	public static class Config {
+
+		@Value("${" + KafkaEmbedded.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
+		private String brokerAddresses;
+
+		@Bean
+		public ProducerFactory<JsonObjectKey, JsonObjectValue> producerFactory() {
+			return new DefaultKafkaProducerFactory<>(producerConfigs());
+		}
+
+		@Bean
+		public Map<String, Object> producerConfigs() {
+			Map<String, Object> senderProps = KafkaTestUtils.senderProps(this.brokerAddresses);
+			senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+			senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+			return senderProps;
+		}
+
+		@Bean
+		public KafkaTemplate<?, ?> kafkaTemplate() {
+			KafkaTemplate<JsonObjectKey, JsonObjectValue> kafkaTemplate = new KafkaTemplate<>(producerFactory());
+			kafkaTemplate.setDefaultTopic(INPUT_TOPIC);
+			return kafkaTemplate;
+		}
+
+		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
+		public StreamsConfig kStreamsConfigs() {
+			Map<String, Object> props = new HashMap<>();
+			props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
+			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
+			return new StreamsConfig(props);
+		}
+
+		@Bean
+		@SuppressWarnings("unchecked")
+		public KStream<JsonObjectKey, JsonObjectValue> testJsonSerializationStream(StreamsBuilder streamsBuilder) {
+			KStream<JsonObjectKey, JsonObjectValue> testStream = streamsBuilder
+					.stream(INPUT_TOPIC, Consumed.with(jsonObjectKeySerde(), jsonObjectValueSerde()));
+
+			testStream.print(Printed.toSysOut());
+			testStream.to(OUTPUT_TOPIC, Produced.with(jsonObjectKeySerde(), jsonObjectValueSerde()));
+
+			return testStream;
+		}
+	}
+
+}