4.10.0

Author: krystianity

CHANGELOG.md

@@ -1,5 +1,16 @@
 # kafka-streams CHANGELOG
 
+## 2019-05-13, Version 4.10.0
+
+* fixed typo that broke window functionality
+* correctly handling rejection of .to() promise call
+* correctly handling produce errors (make sure to sub `kafkaStreams.on("error", (error) => {...});`)
+* `KStream.branch([...])` now returns stream clones that consist of deeply cloned events
+* `KStream.clone(cloneObjects = false, cloneDeep = false)` now offers optional parameters to clone the stream events,
+otherwise mutating the origin stream will alter the cloned stream's objects
+* hardened the StreamDSL JSON convenience methods
+* other small refactorings
+
 ## 2019-05-07, Version 4.9.0
 
 * upgraded dependencies

examples/consumeFromTopic.js

@@ -6,6 +6,10 @@ const { nativeConfig: config } = require("./../test/test-config.js");
 const kafkaStreams = new KafkaStreams(config);
 const stream = kafkaStreams.getKStream("my-input-topic");
 
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 //adding a side effect call to the stream via tap
 stream.forEach((message) => {
     console.log("key", message.key ? message.key.toString("utf8") : null);

examples/consumeOneProduceTwo.js

@@ -6,6 +6,10 @@ const { nativeConfig: config } = require("./../test/test-config.js");
 const kafkaStreams = new KafkaStreams(config);
 const stream = kafkaStreams.getKStream();
 
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 stream
     .from("input_topic")
     .mapJSONConvenience()

examples/consumeOneProduceTwoTopics.js

@@ -0,0 +1,38 @@
+"use strict";
+
+const { KafkaStreams } = require("./../index.js");
+const { nativeConfig: config } = require("./../test/test-config.js");
+
+const kafkaStreams = new KafkaStreams(config);
+const stream$ = kafkaStreams.getKStream("input-topic");
+
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
+const [one$, two$] = stream$
+    .branch([() => true, () => true]);
+
+const producerPromiseOne = one$
+    .mapJSONConvenience()
+    .mapWrapKafkaValue()
+    .tap((msg) => console.log("one", msg))
+    .wrapAsKafkaValue()
+    .to("output-topic-1", 1, "buffer");
+
+const producerPromiseTwo = two$
+    .mapJSONConvenience()
+    .mapWrapKafkaValue()
+    .tap((msg) => console.log("two", msg))
+    .wrapAsKafkaValue()
+    .to("output-topic-2", 1, "buffer");
+
+Promise.all([
+    producerPromiseOne,
+    producerPromiseTwo,
+    stream$.start(),
+]).then(() => {
+    console.log("Stream started, as kafka consumer and producers are ready.");
+}, (error) => {
+    console.log("Streaming operation failed to start: ", error);
+});

examples/fieldSum.js

@@ -6,6 +6,10 @@ const { nativeConfig: config } = require("./../test/test-config.js");
 const kafkaStreams = new KafkaStreams(config);
 const stream = kafkaStreams.getKStream("my-input-topic");
 
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 stream
     .mapStringToKV(" ", 0, 1) //string to key-value object; args: delimiter, key-index, value-index
     .sumByKey("key", "value", "sum")

examples/mergeTopics.js

@@ -5,6 +5,10 @@ const { nativeConfig: config } = require("./../test/test-config.js");
 
 const kafkaStreams = new KafkaStreams(config);
 
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 const stream1 = kafkaStreams.getKStream("my-input-topic-1");
 const stream2 = kafkaStreams.getKStream("my-input-topic-2");
 const stream3 = kafkaStreams.getKStream("my-input-topic-3");

examples/produceToTopic.js

@@ -4,6 +4,11 @@ const { KafkaStreams } = require("./../index.js");
 const { nativeConfig: config } = require("./../test/test-config.js");
 
 const kafkaStreams = new KafkaStreams(config);
+
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 const stream = kafkaStreams.getKStream(null);
 //creating a stream without topic is possible
 //no consumer will be created during stream.start()

examples/topicAsTable.js

@@ -24,6 +24,10 @@ const { nativeConfig: config } = require("./../test/test-config.js");
 
 const kafkaStreams = new KafkaStreams(config);
 
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 //creating a ktable requires a function that can be
 //used to turn the kafka messages into key-value objects
 //as tables can only be built on key-value pairs

examples/window.js

@@ -4,6 +4,11 @@ const { KafkaStreams } = require("./../index.js");
 const { nativeConfig: config } = require("./../test/test-config.js");
 
 const kafkaStreams = new KafkaStreams(config);
+
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 const consumeStream = kafkaStreams.getKStream("my-input-topic");
 
 const windowPeriod = 30 * 1000; // 30 seconds

examples/wordCount.js

@@ -32,6 +32,11 @@ const keyMapperEtl = (kafkaMessage) => {
 };
 
 const kafkaStreams = new KafkaStreams(config);
+
+kafkaStreams.on("error", (error) => {
+    console.log("Error occured:", error.message);
+});
+
 const stream = kafkaStreams.getKStream();
 
 stream