kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-4772: Exploit #peek to implement #print() and other methods
Date Wed, 17 May 2017 18:15:34 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 8c7e66313 -> 6910baf54


KAFKA-4772: Exploit #peek to implement #print() and other methods

I remove `KeyValuePrinter` and `KStreamForeach` two class, then implements them by `KStreamPeek`.
So, now `KStreamPeek` can do `KeyValuePrinter` and `KStreamForeach` job.

Author: jameschien <jameschien@staff.ruten.com.tw>
Author: JamesChien <jedichien@users.noreply.github.com>

Reviewers: Matthias J. Sax, Guozhang Wang

Closes #2955 from jedichien/trunk


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/6910baf5
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/6910baf5
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/6910baf5

Branch: refs/heads/trunk
Commit: 6910baf548ebad4c1530432e51be40793b4a4f10
Parents: 8c7e663
Author: James Chien <jameschien@staff.ruten.com.tw>
Authored: Wed May 17 11:15:31 2017 -0700
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Wed May 17 11:15:31 2017 -0700

----------------------------------------------------------------------
 .../streams/kstream/PrintForeachAction.java     |  59 ++++++
 .../kstream/internals/KStreamForeach.java       |  43 ----
 .../streams/kstream/internals/KStreamImpl.java  |   9 +-
 .../streams/kstream/internals/KStreamPeek.java  |   9 +-
 .../streams/kstream/internals/KStreamPrint.java |  89 ++++++++
 .../streams/kstream/internals/KTableImpl.java   |   9 +-
 .../kstream/internals/KeyValuePrinter.java      | 120 -----------
 .../kstream/internals/KStreamPeekTest.java      |   9 +-
 .../kstream/internals/KStreamPrintTest.java     |  91 ++++++++
 .../internals/KeyValuePrinterProcessorTest.java | 207 -------------------
 10 files changed, 262 insertions(+), 383 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/main/java/org/apache/kafka/streams/kstream/PrintForeachAction.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/PrintForeachAction.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/PrintForeachAction.java
new file mode 100644
index 0000000..3eb6d80
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/PrintForeachAction.java
@@ -0,0 +1,59 @@
+/*
+ * 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
+ *
+ *    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.apache.kafka.streams.kstream;
+
+import java.io.PrintWriter;
+
+public class PrintForeachAction<K, V> implements ForeachAction<K, V> {
+
+    private final String streamName;
+    private final PrintWriter printWriter;
+    
+    /**
+     * Print data message with given writer. The PrintWriter can be null in order to
+     * distinguish between {@code System.out} and the others. If the PrintWriter is {@code
PrintWriter(System.out)},
+     * then it would close {@code System.out} output stream.
+     * <p>
+     * Afterall, not to pass in {@code PrintWriter(System.out)} but {@code null} instead.
+     *
+     * @param printWriter Use {@code System.out.println} if {@code null}.
+     * @param streamName The given name will be printed.
+     */
+    public PrintForeachAction(final PrintWriter printWriter, final String streamName) {
+        this.printWriter = printWriter;
+        this.streamName = streamName;
+    }
+
+    @Override
+    public void apply(final K key, final V value) {
+        final String data = String.format("[%s]: %s, %s", streamName, key, value);
+        if (printWriter == null) {
+            System.out.println(data);
+        } else {
+            printWriter.println(data);
+        }
+    }
+
+    public void close() {
+        if (printWriter == null) {
+            System.out.flush();
+        } else {
+            printWriter.close();
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamForeach.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamForeach.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamForeach.java
deleted file mode 100644
index eb3189c..0000000
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamForeach.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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
- *
- *    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.apache.kafka.streams.kstream.internals;
-
-import org.apache.kafka.streams.kstream.ForeachAction;
-import org.apache.kafka.streams.processor.AbstractProcessor;
-import org.apache.kafka.streams.processor.Processor;
-import org.apache.kafka.streams.processor.ProcessorSupplier;
-
-class KStreamForeach<K, V> implements ProcessorSupplier<K, V> {
-
-    private final ForeachAction<K, V> action;
-
-    public KStreamForeach(ForeachAction<K, V> action) {
-        this.action = action;
-    }
-
-    @Override
-    public Processor<K, V> get() {
-        return new KStreamForeachProcessor();
-    }
-
-    private class KStreamForeachProcessor extends AbstractProcessor<K, V> {
-        @Override
-        public void process(K key, V value) {
-            action.apply(key, value);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
index b751294..38b0a85 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.errors.TopologyBuilderException;
 import org.apache.kafka.streams.kstream.ForeachAction;
+import org.apache.kafka.streams.kstream.PrintForeachAction;
 import org.apache.kafka.streams.kstream.JoinWindows;
 import org.apache.kafka.streams.kstream.KGroupedStream;
 import org.apache.kafka.streams.kstream.GlobalKTable;
@@ -193,7 +194,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K> implements
KStream<K, V
     public void print(Serde<K> keySerde, Serde<V> valSerde, String streamName)
{
         String name = topology.newName(PRINTING_NAME);
         streamName = (streamName == null) ? this.name : streamName;
-        topology.addProcessor(name, new KeyValuePrinter<>(keySerde, valSerde, streamName),
this.name);
+        topology.addProcessor(name, new KStreamPrint<>(new PrintForeachAction(null,
streamName), keySerde, valSerde), this.name);
     }
 
 
@@ -227,7 +228,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K> implements
KStream<K, V
         try {
             PrintWriter printWriter = null;
             printWriter = new PrintWriter(filePath, StandardCharsets.UTF_8.name());
-            topology.addProcessor(name, new KeyValuePrinter<>(printWriter, keySerde,
valSerde, streamName), this.name);
+            topology.addProcessor(name, new KStreamPrint<>(new PrintForeachAction(printWriter,
streamName), keySerde, valSerde), this.name);
         } catch (FileNotFoundException | UnsupportedEncodingException e) {
             String message = "Unable to write stream to file at [" + filePath + "] " + e.getMessage();
             throw new TopologyBuilderException(message);
@@ -315,7 +316,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K> implements
KStream<K, V
         Objects.requireNonNull(action, "action can't be null");
         String name = topology.newName(FOREACH_NAME);
 
-        topology.addProcessor(name, new KStreamForeach<>(action), this.name);
+        topology.addProcessor(name, new KStreamPeek<>(action, false), this.name);
     }
 
     @Override
@@ -323,7 +324,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K> implements
KStream<K, V
         Objects.requireNonNull(action, "action can't be null");
         final String name = topology.newName(PEEK_NAME);
 
-        topology.addProcessor(name, new KStreamPeek<>(action), this.name);
+        topology.addProcessor(name, new KStreamPeek<>(action, true), this.name);
 
         return new KStreamImpl<>(topology, name, sourceNodes, repartitionRequired);
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPeek.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPeek.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPeek.java
index 2e72f68..44d1d60 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPeek.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPeek.java
@@ -23,10 +23,12 @@ import org.apache.kafka.streams.processor.ProcessorSupplier;
 
 class KStreamPeek<K, V> implements ProcessorSupplier<K, V> {
 
+    private final boolean forwardDownStream;
     private final ForeachAction<K, V> action;
 
-    public KStreamPeek(final ForeachAction<K, V> action) {
+    public KStreamPeek(final ForeachAction<K, V> action, final boolean forwardDownStream)
{
         this.action = action;
+        this.forwardDownStream = forwardDownStream;
     }
 
     @Override
@@ -38,7 +40,10 @@ class KStreamPeek<K, V> implements ProcessorSupplier<K, V>
{
         @Override
         public void process(final K key, final V value) {
             action.apply(key, value);
-            context().forward(key, value);
+            if (forwardDownStream) {
+                context().forward(key, value);
+            }
         }
     }
+
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPrint.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPrint.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPrint.java
new file mode 100644
index 0000000..8447ae1
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamPrint.java
@@ -0,0 +1,89 @@
+/*
+ * 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
+ *
+ *    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.apache.kafka.streams.kstream.internals;
+
+import org.apache.kafka.streams.kstream.ForeachAction;
+import org.apache.kafka.streams.kstream.PrintForeachAction;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.AbstractProcessor;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.ProcessorSupplier;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+
+public class KStreamPrint<K, V> implements ProcessorSupplier<K, V> {
+
+    private final Serde<?> keySerde;
+    private final Serde<?> valueSerde;
+    private final ForeachAction<K, V> action;
+    
+    public KStreamPrint(final ForeachAction<K, V> action, final Serde<?> keySerde,
final Serde<?> valueSerde) {
+        this.action = action;
+        this.keySerde = keySerde;
+        this.valueSerde = valueSerde;
+    }
+
+    @Override
+    public Processor<K, V> get() {
+        return new KStreamPrintProcessor(keySerde, valueSerde);
+    }
+
+    private class KStreamPrintProcessor extends AbstractProcessor<K, V> {
+        
+        private Serde<?> keySerde;
+        private Serde<?> valueSerde;
+        private ProcessorContext context;
+        
+        public KStreamPrintProcessor(final Serde<?> keySerde, final Serde<?>
valueSerde) {
+            this.keySerde = keySerde;
+            this.valueSerde = valueSerde;
+        }
+
+        @Override
+        public void init(ProcessorContext context) {
+            this.context = context;
+            if (keySerde == null) {
+                this.keySerde = context.keySerde();
+            }
+            if (valueSerde == null) {
+                this.valueSerde = context.valueSerde();
+            }
+        }
+
+        @Override
+        public void process(final K key, final V value) {
+            final K deKey = (K) maybeDeserialize(key, keySerde.deserializer());
+            final V deValue = (V) maybeDeserialize(value, valueSerde.deserializer());
+            action.apply(deKey, deValue);
+        }
+
+        private Object maybeDeserialize(final Object keyOrValue, final Deserializer<?>
deserializer) {
+            if (keyOrValue instanceof byte[]) {
+                return deserializer.deserialize(this.context.topic(), (byte[]) keyOrValue);
+            }
+            return keyOrValue;
+        }
+        
+        @Override
+        public void close() {
+            if (action instanceof PrintForeachAction) {
+                ((PrintForeachAction) action).close();
+            }
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
index 96a0b2c..2103ff2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.errors.TopologyBuilderException;
 import org.apache.kafka.streams.kstream.ForeachAction;
+import org.apache.kafka.streams.kstream.PrintForeachAction;
 import org.apache.kafka.streams.kstream.KGroupedTable;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KStreamBuilder;
@@ -242,7 +243,7 @@ public class KTableImpl<K, S, V> extends AbstractStream<K>
implements KTable<K,
     public void print(Serde<K> keySerde, Serde<V> valSerde, String streamName)
{
         String name = topology.newName(PRINTING_NAME);
         streamName = (streamName == null) ? this.name : streamName;
-        topology.addProcessor(name, new KeyValuePrinter<>(keySerde, valSerde, streamName),
this.name);
+        topology.addProcessor(name, new KStreamPrint<>(new PrintForeachAction(null,
streamName), keySerde, valSerde), this.name);
     }
 
     @Override
@@ -273,7 +274,7 @@ public class KTableImpl<K, S, V> extends AbstractStream<K>
implements KTable<K,
         streamName = (streamName == null) ? this.name : streamName;
         try {
             PrintWriter printWriter = new PrintWriter(filePath, StandardCharsets.UTF_8.name());
-            topology.addProcessor(name, new KeyValuePrinter<>(printWriter, keySerde,
valSerde, streamName), this.name);
+            topology.addProcessor(name, new KStreamPrint<>(new PrintForeachAction(printWriter,
streamName), keySerde, valSerde), this.name);
         } catch (FileNotFoundException | UnsupportedEncodingException e) {
             String message = "Unable to write stream to file at [" + filePath + "] " + e.getMessage();
             throw new TopologyBuilderException(message);
@@ -284,12 +285,12 @@ public class KTableImpl<K, S, V> extends AbstractStream<K>
implements KTable<K,
     public void foreach(final ForeachAction<? super K, ? super V> action) {
         Objects.requireNonNull(action, "action can't be null");
         String name = topology.newName(FOREACH_NAME);
-        KStreamForeach<K, Change<V>> processorSupplier = new KStreamForeach<>(new
ForeachAction<K, Change<V>>() {
+        KStreamPeek<K, Change<V>> processorSupplier = new KStreamPeek<>(new
ForeachAction<K, Change<V>>() {
             @Override
             public void apply(K key, Change<V> value) {
                 action.apply(key, value.newValue);
             }
-        });
+        }, false);
         topology.addProcessor(name, processorSupplier, this.name);
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinter.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinter.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinter.java
deleted file mode 100644
index c1d4382..0000000
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinter.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * 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
- *
- *    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.apache.kafka.streams.kstream.internals;
-
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.streams.processor.AbstractProcessor;
-import org.apache.kafka.streams.processor.Processor;
-import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.processor.ProcessorSupplier;
-
-import java.io.PrintWriter;
-
-
-class KeyValuePrinter<K, V> implements ProcessorSupplier<K, V> {
-    private final PrintWriter printWriter;
-    private Serde<?> keySerde;
-    private Serde<?> valueSerde;
-    private String streamName;
-
-    KeyValuePrinter(PrintWriter printWriter, Serde<?> keySerde, Serde<?> valueSerde,
String streamName) {
-        this.keySerde = keySerde;
-        this.valueSerde = valueSerde;
-        this.streamName = streamName;
-        this.printWriter = printWriter;
-    }
-
-    KeyValuePrinter(PrintWriter printWriter, String streamName) {
-        this(printWriter, null, null, streamName);
-    }
-
-    KeyValuePrinter(Serde<?> keySerde, Serde<?> valueSerde, String streamName)
{
-        this(null, keySerde, valueSerde, streamName);
-    }
-
-    @Override
-    public Processor<K, V> get() {
-        return new KeyValuePrinterProcessor(this.printWriter, this.keySerde, this.valueSerde,
this.streamName);
-    }
-
-    private class KeyValuePrinterProcessor extends AbstractProcessor<K, V> {
-        private final PrintWriter printWriter;
-        private Serde<?> keySerde;
-        private Serde<?> valueSerde;
-        private ProcessorContext processorContext;
-        private String streamName;
-
-        private KeyValuePrinterProcessor(PrintWriter printWriter, Serde<?> keySerde,
Serde<?> valueSerde, String streamName) {
-            this.printWriter = printWriter;
-            this.keySerde = keySerde;
-            this.valueSerde = valueSerde;
-            this.streamName = streamName;
-        }
-
-        @Override
-        public void init(ProcessorContext context) {
-            this.processorContext = context;
-
-            if (this.keySerde == null) {
-                keySerde = this.processorContext.keySerde();
-            }
-
-            if (this.valueSerde == null) {
-                valueSerde = this.processorContext.valueSerde();
-            }
-        }
-
-        @Override
-        public void process(K key, V value) {
-            K keyToPrint = (K) maybeDeserialize(key, keySerde.deserializer());
-            V valueToPrint = (V) maybeDeserialize(value, valueSerde.deserializer());
-
-            println("[" + this.streamName + "]: " + keyToPrint + " , " + valueToPrint);
-
-            this.processorContext.forward(key, value);
-        }
-
-        private void println(String str) {
-            if (printWriter == null)
-                System.out.println(str);
-            else
-                printWriter.println(str);
-        }
-
-        private Object maybeDeserialize(Object receivedElement, Deserializer<?> deserializer)
{
-            if (receivedElement == null) {
-                return null;
-            }
-
-            if (receivedElement instanceof byte[]) {
-                return deserializer.deserialize(this.processorContext.topic(), (byte[]) receivedElement);
-            }
-
-            return receivedElement;
-        }
-
-        @Override
-        public void close() {
-            if (printWriter == null) {
-                System.out.flush();
-            } else {
-                printWriter.close();
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPeekTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPeekTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPeekTest.java
index 7991a92..df6f765 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPeekTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPeekTest.java
@@ -16,12 +16,14 @@
  */
 package org.apache.kafka.streams.kstream.internals;
 
+import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.ForeachAction;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KStreamBuilder;
 import org.apache.kafka.test.KStreamTestDriver;
+
 import org.junit.After;
 import org.junit.Test;
 
@@ -34,7 +36,8 @@ import static org.junit.Assert.fail;
 public class KStreamPeekTest {
 
     private final String topicName = "topic";
-
+    private final Serde<Integer> intSerd = Serdes.Integer();
+    private final Serde<String> stringSerd = Serdes.String();
     private KStreamTestDriver driver = null;
 
     @After
@@ -47,7 +50,7 @@ public class KStreamPeekTest {
     @Test
     public void shouldObserveStreamElements() {
         final KStreamBuilder builder = new KStreamBuilder();
-        final KStream<Integer, String> stream = builder.stream(Serdes.Integer(), Serdes.String(),
topicName);
+        final KStream<Integer, String> stream = builder.stream(intSerd, stringSerd,
topicName);
         final List<KeyValue<Integer, String>> peekObserved = new ArrayList<>(),
streamObserved = new ArrayList<>();
         stream.peek(collect(peekObserved)).foreach(collect(streamObserved));
 
@@ -66,7 +69,7 @@ public class KStreamPeekTest {
     @Test
     public void shouldNotAllowNullAction() {
         final KStreamBuilder builder = new KStreamBuilder();
-        final KStream<Integer, String> stream = builder.stream(Serdes.Integer(), Serdes.String(),
topicName);
+        final KStream<Integer, String> stream = builder.stream(intSerd, stringSerd,
topicName);
         try {
             stream.peek(null);
             fail("expected null action to throw NPE");

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPrintTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPrintTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPrintTest.java
new file mode 100644
index 0000000..c537e0a
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamPrintTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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
+ *
+ *    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.apache.kafka.streams.kstream.internals;
+
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.PrintForeachAction;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.test.KStreamTestDriver;
+
+import org.junit.Before;
+import org.junit.After;
+import org.junit.Test;
+
+import java.io.PrintWriter;
+import java.io.ByteArrayOutputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+public class KStreamPrintTest {
+
+    private final String topicName = "topic";
+    private final Serde<Integer> intSerd = Serdes.Integer();
+    private final Serde<String> stringSerd = Serdes.String();
+    private PrintWriter printWriter;
+    private ByteArrayOutputStream byteOutStream;
+    private KStreamTestDriver driver = null;
+
+
+    @Before
+    public void setUp() {
+        byteOutStream = new ByteArrayOutputStream();
+        printWriter = new PrintWriter(new OutputStreamWriter(byteOutStream, StandardCharsets.UTF_8));
+    }
+
+    @After
+    public void cleanup() {
+        if (driver != null) {
+            driver.close();
+        }
+    }
+    
+    @Test
+    public void testPrintKeyValueWithName() {
+        final KStreamPrint<Integer, String> kStreamPrint = new KStreamPrint<>(new
PrintForeachAction(printWriter, "test-stream"), intSerd, stringSerd);
+
+        final List<KeyValue<Integer, String>> inputRecords = Arrays.asList(
+                new KeyValue<>(0, "zero"),
+                new KeyValue<>(1, "one"),
+                new KeyValue<>(2, "two"),
+                new KeyValue<>(3, "three"));
+        
+        final String[] expectedResult = {"[test-stream]: 0, zero", "[test-stream]: 1, one",
"[test-stream]: 2, two", "[test-stream]: 3, three"};
+        
+        final KStreamBuilder builder = new KStreamBuilder();
+        final KStream<Integer, String> stream = builder.stream(intSerd, stringSerd,
topicName);
+        stream.process(kStreamPrint);
+        
+        driver = new KStreamTestDriver(builder);
+        for (KeyValue<Integer, String> record: inputRecords) {
+            driver.process(topicName, record.key, record.value);
+        }
+        printWriter.flush();
+        final String[] flushOutDatas = new String(byteOutStream.toByteArray(), Charset.forName("UTF-8")).split("\n");
+        for (int i = 0; i < flushOutDatas.length; i++) {
+            assertEquals(flushOutDatas[i], expectedResult[i]);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/6910baf5/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinterProcessorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinterProcessorTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinterProcessorTest.java
deleted file mode 100644
index 7de2b8b..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KeyValuePrinterProcessorTest.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * 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
- *
- *    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.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.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.ByteArrayOutputStream;
-import java.io.OutputStreamWriter;
-import java.io.PrintWriter;
-import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
-import java.util.Map;
-
-import static org.junit.Assert.assertEquals;
-
-public class KeyValuePrinterProcessorTest {
-
-    private final String topicName = "topic";
-    private final Serde<String> stringSerde = Serdes.String();
-    private ByteArrayOutputStream baos;
-    private KStreamBuilder builder;
-    private PrintWriter printWriter;
-    private KStreamTestDriver driver;
-
-    @Before
-    public void setup() {
-        baos = new ByteArrayOutputStream();
-        builder = new KStreamBuilder();
-        printWriter = new PrintWriter(new OutputStreamWriter(baos, StandardCharsets.UTF_8));
-        driver = null;
-    }
-
-    @After
-    public void cleanup() {
-        if (driver != null) {
-            driver.close();
-        }
-        driver = null;
-    }
-
-    @Test
-    public void testPrintKeyValueDefaultSerde() throws Exception {
-
-        KeyValuePrinter<String, String> keyValuePrinter = new KeyValuePrinter<>(printWriter,
null);
-        String[] suppliedKeys = {"foo", "bar", null};
-        String[] suppliedValues = {"value1", "value2", "value3"};
-        String[] expectedValues = {"[null]: foo , value1", "[null]: bar , value2", "[null]:
null , value3"};
-
-
-        KStream<String, String> stream = builder.stream(stringSerde, stringSerde, topicName);
-        stream.process(keyValuePrinter);
-
-        driver = new KStreamTestDriver(builder);
-        for (int i = 0; i < suppliedKeys.length; i++) {
-            driver.process(topicName, suppliedKeys[i], suppliedValues[i]);
-        }
-        printWriter.flush();
-        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 testPrintKeyValuesWithName() throws Exception {
-
-        KeyValuePrinter<String, String> keyValuePrinter = new KeyValuePrinter<>(printWriter,
"test-stream");
-        String[] suppliedKeys = {"foo", "bar", null};
-        String[] suppliedValues = {"value1", "value2", "value3"};
-        String[] expectedValues = {"[test-stream]: foo , value1", "[test-stream]: bar , value2",
"[test-stream]: null , value3"};
-
-
-        KStream<String, String> stream = builder.stream(stringSerde, stringSerde, topicName);
-        stream.process(keyValuePrinter);
-
-        driver = new KStreamTestDriver(builder);
-        for (int i = 0; i < suppliedKeys.length; i++) {
-            driver.process(topicName, suppliedKeys[i], suppliedValues[i]);
-        }
-        printWriter.flush();
-        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<>(printWriter,
stringSerde, mockObjectSerde, null);
-        KStream<String, MockObject> stream = builder.stream(stringSerde, mockObjectSerde,
topicName);
-
-        stream.process(keyValuePrinter);
-
-        driver = new KStreamTestDriver(builder);
-
-        String suppliedKey = null;
-        byte[] suppliedValue = "{\"name\":\"print\", \"label\":\"test\"}".getBytes(Charset.forName("UTF-8"));
-
-        driver.process(topicName, suppliedKey, suppliedValue);
-        printWriter.flush();
-        String capturedValue = new String(baos.toByteArray(), Charset.forName("UTF-8")).trim();
-
-        assertEquals("[null]: null , name:print label:test", capturedValue);
-
-    }
-
-    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() {
-
-        }
-    }
-
-
-}
\ No newline at end of file


Mime
View raw message