Bael 1421 (#3312)

* BAEL-1421 live test of kafka streams * BAEL-1421 Removed not-needed dependency * BAEL-1421 rearannge * BAEL-1421 rearannge * fix pom * move to libraries-data * move to libraries-data
2017-12-28 11:21:11 +01:00
parent e87114be6e
commit 0a76081ac3
2 changed files with 112 additions and 19 deletions
@@ -0,0 +1,62 @@
+package com.baeldung.kafkastreams;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+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.StreamsConfig;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.test.TestUtils;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Properties;
+import java.util.regex.Pattern;
+
+public class KafkaStreamsLiveTest {
+    private String bootstrapServers = "localhost:9092";
+
+    @Test
+    @Ignore("it needs to have kafka broker running on local")
+    public void shouldTestKafkaStreams() throws InterruptedException {
+        //given
+        String inputTopic = "inputTopic";
+
+        Properties streamsConfiguration = new Properties();
+        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-live-test");
+        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
+        streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        // Use a temporary directory for storing state, which will be automatically removed after the test.
+        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
+
+        //when
+        KStreamBuilder builder = new KStreamBuilder();
+        KStream<String, String> textLines = builder.stream(inputTopic);
+        Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS);
+
+        KTable<String, Long> wordCounts = textLines
+                .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase())))
+                .groupBy((key, word) -> word)
+                .count();
+
+        wordCounts.foreach((word, count) -> System.out.println("word: " + word + " -> " + count));
+
+        String outputTopic = "outputTopic";
+        final Serde<String> stringSerde = Serdes.String();
+        final Serde<Long> longSerde = Serdes.Long();
+        wordCounts.to(stringSerde, longSerde, outputTopic);
+
+        KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
+        streams.start();
+
+        //then
+        Thread.sleep(30000);
+        streams.close();
+    }
+}