| /** |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You 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 |
| * <p> |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * <p> |
| * 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.apache.kafka.streams.kstream.internals; |
| |
| |
| import org.apache.kafka.common.errors.SerializationException; |
| import org.apache.kafka.common.serialization.Deserializer; |
| import org.apache.kafka.common.serialization.Serde; |
| import org.apache.kafka.common.serialization.Serdes; |
| import org.apache.kafka.common.serialization.Serializer; |
| import org.apache.kafka.streams.kstream.KStream; |
| import org.apache.kafka.streams.kstream.KStreamBuilder; |
| import org.apache.kafka.test.KStreamTestDriver; |
| import org.junit.Test; |
| |
| import java.io.ByteArrayOutputStream; |
| import java.io.PrintStream; |
| import java.nio.charset.Charset; |
| import java.util.Map; |
| |
| import static org.junit.Assert.assertEquals; |
| |
| public class KeyValuePrinterProcessorTest { |
| |
| private String topicName = "topic"; |
| private Serde<String> stringSerde = Serdes.String(); |
| private Serde<byte[]> bytesSerde = Serdes.ByteArray(); |
| private ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| private KStreamBuilder builder = new KStreamBuilder(); |
| private PrintStream printStream = new PrintStream(baos); |
| |
| |
| @Test |
| public void testPrintKeyValueDefaultSerde() throws Exception { |
| |
| KeyValuePrinter<String, String> keyValuePrinter = new KeyValuePrinter<>(printStream); |
| String[] suppliedKeys = {"foo", "bar", null}; |
| String[] suppliedValues = {"value1", "value2", "value3"}; |
| String[] expectedValues = {"foo , value1", "bar , value2", "null , value3"}; |
| |
| |
| KStream<String, String> stream = builder.stream(stringSerde, stringSerde, topicName); |
| stream.process(keyValuePrinter); |
| |
| KStreamTestDriver driver = new KStreamTestDriver(builder); |
| for (int i = 0; i < suppliedKeys.length; i++) { |
| driver.process(topicName, suppliedKeys[i], suppliedValues[i]); |
| } |
| |
| String[] capturedValues = new String(baos.toByteArray(), Charset.forName("UTF-8")).split("\n"); |
| |
| for (int i = 0; i < capturedValues.length; i++) { |
| assertEquals(capturedValues[i], expectedValues[i]); |
| } |
| } |
| |
| |
| @Test |
| public void testPrintKeyValueWithProvidedSerde() throws Exception { |
| |
| Serde<MockObject> mockObjectSerde = Serdes.serdeFrom(new MockSerializer(), new MockDeserializer()); |
| KeyValuePrinter<String, MockObject> keyValuePrinter = new KeyValuePrinter<>(printStream, stringSerde, mockObjectSerde); |
| KStream<String, MockObject> stream = builder.stream(stringSerde, mockObjectSerde, topicName); |
| |
| stream.process(keyValuePrinter); |
| |
| KStreamTestDriver driver = new KStreamTestDriver(builder); |
| |
| String suppliedKey = null; |
| byte[] suppliedValue = "{\"name\":\"print\", \"label\":\"test\"}".getBytes(Charset.forName("UTF-8")); |
| |
| driver.process(topicName, suppliedKey, suppliedValue); |
| String expectedPrintedValue = "null , name:print label:test"; |
| String capturedValue = new String(baos.toByteArray(), Charset.forName("UTF-8")).trim(); |
| |
| assertEquals(capturedValue, expectedPrintedValue); |
| |
| } |
| |
| private static class MockObject { |
| public String name; |
| public String label; |
| |
| public MockObject() { |
| } |
| |
| MockObject(String name, String label) { |
| this.name = name; |
| this.label = label; |
| } |
| |
| @Override |
| public String toString() { |
| return "name:" + name + " label:" + label; |
| } |
| } |
| |
| |
| private static class MockDeserializer implements Deserializer<MockObject> { |
| |
| private com.fasterxml.jackson.databind.ObjectMapper objectMapper = new com.fasterxml.jackson.databind.ObjectMapper(); |
| |
| @Override |
| public void configure(Map<String, ?> configs, boolean isKey) { |
| |
| } |
| |
| @Override |
| public MockObject deserialize(String topic, byte[] data) { |
| MockObject mockObject; |
| try { |
| mockObject = objectMapper.readValue(data, MockObject.class); |
| } catch (Exception e) { |
| throw new SerializationException(e); |
| } |
| return mockObject; |
| } |
| |
| @Override |
| public void close() { |
| |
| } |
| } |
| |
| |
| private static class MockSerializer implements Serializer<MockObject> { |
| private final com.fasterxml.jackson.databind.ObjectMapper objectMapper = new com.fasterxml.jackson.databind.ObjectMapper(); |
| |
| @Override |
| public void configure(Map<String, ?> configs, boolean isKey) { |
| |
| } |
| |
| @Override |
| public byte[] serialize(String topic, MockObject data) { |
| try { |
| return objectMapper.writeValueAsBytes(data); |
| } catch (Exception e) { |
| throw new SerializationException("Error serializing JSON message", e); |
| } |
| } |
| |
| @Override |
| public void close() { |
| |
| } |
| } |
| |
| |
| } |