Refactoring and implementing one kafka stream example

jefersonm · Jun 27, 2016 · 2e11002 · 2e11002
1 parent 4c2041a
commit 2e11002
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 41 deletions.
diff --git a/...efersonm/KafkaStreamsPlayground/Main.java → ...ain/java/jefersonm/kafka/stream/Main.java b/...efersonm/KafkaStreamsPlayground/Main.java → ...ain/java/jefersonm/kafka/stream/Main.java
@@ -1,33 +1,30 @@
-package jefersonm.KafkaStreamsPlayground;
+package jefersonm.kafka.stream;
 
-import jefersonm.model.Tweet;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.kstream.KStreamBuilder;
 import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.processor.WallclockTimestampExtractor;
 
 import java.util.Properties;
 
 /**
  * Created by jefersonm on 6/15/16.
  */
-//http://docs.confluent.io/2.1.0-alpha1/streams/developer-guide.html#writing-a-kafka-streams-application
+//http://docs.confluent.io/3.0.0/streams/developer-guide.html
 public class Main {
 
-
     public static void main(String[] args){
 
+        // When you want to override serdes explicitly/selectively
         final Serde<String> stringSerde = Serdes.String();
-        final Serde<byte[]> byteArraySerde = Serdes.ByteArray();
 
         KStreamBuilder builder = new KStreamBuilder();
 
-        // Method 1: Read the input Kafka topic into a KStream instance.
+        // Method 1: Read the input Kafka topic `TextLinesTopic` into a KStream instance.
         KStream<String, String> textLines = builder.stream(stringSerde, stringSerde, "TextLinesTopic");
 
         // Write (i.e. persist) the results to a new Kafka topic called "UppercasedTextLinesTopic".
@@ -40,7 +37,7 @@ public static void main(String[] args){
         // Method 3: using `map`, modify both key and value
         KStream<String, String> originalAndUppercased = textLines.map((key, value) -> KeyValue.pair(value, value.toUpperCase()));
 
-        // Write the results to a new Kafka topic "OriginalAndUppercased".
+        // Write the results to a new Kafka topic `OriginalAndUppercased`.
         originalAndUppercased.to(stringSerde, stringSerde, "OriginalAndUppercased");
 
         StreamsConfig config = new StreamsConfig(getProperties());
@@ -57,6 +54,11 @@ private static Properties getProperties() {
         props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
         props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 1);
         props.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class);
+
+        // When configuring the default serdes of StreamConfig
+        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+
         return props;
     }
 

diff --git a/techs/reactive/kafka-streams/src/main/java/jefersonm/model/Tweet.java b/techs/reactive/kafka-streams/src/main/java/jefersonm/model/Tweet.java
diff --git a/techs/reactive/kafka-streams/src/main/java/jefersonm/twitter/TwitterClient.java b/techs/reactive/kafka-streams/src/main/java/jefersonm/twitter/TwitterClient.java