SpringCloud/spring-cloud-stream-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleanup kafka streams samples yaml configuration

Browse Source
This commit is contained in:
Soby Chacko
2019-01-07 16:55:37 -05:00
parent 92803928ab
commit c518e38aef
5 changed files with 21 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
kafka-streams-samples/kafka-streams-dlq-sample/src/main/java/kafka/streams/dlq/sample/KafkaStreamsDlqSample.java
View File

@@ -43,9 +43,6 @@ public class KafkaStreamsDlqSample {
@EnableBinding(KafkaStreamsProcessor.class)
public static class WordCountProcessorApplication {
@Autowired
private TimeWindows timeWindows;
@StreamListener("input")
@SendTo("output")
public KStream<?, WordCount> process(KStream<Object, String> input) {
@@ -54,7 +51,7 @@ public class KafkaStreamsDlqSample {
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Serialized.with(Serdes.String(), Serdes.String()))
.windowedBy(timeWindows)
.windowedBy(TimeWindows.of(5000))
.count(Materialized.as("WordCounts-1"))
.toStream()
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));

18
kafka-streams-samples/kafka-streams-dlq-sample/src/main/resources/application.yml
View File

@@ -1,22 +1,20 @@
spring.cloud.stream.bindings.output.contentType: application/json
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms: 1000
spring.cloud.stream.kafka.streams.binder.configuration:
default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde: org.apache.kafka.common.serialization.Serdes$IntegerSerde
application.id: default
spring.cloud.stream.kafka.streams.binder:
configuration:
commit.interval.ms: 1000
default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde: org.apache.kafka.common.serialization.Serdes$IntegerSerde
application.id: default
brokers: localhost #192.168.99.100
serdeError: sendToDlq
spring.cloud.stream.bindings.output:
destination: counts
producer:
#useNativeEncoding: true
spring.cloud.stream.bindings.input:
destination: words
group: group1
consumer:
useNativeDecoding: true
spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqName: words-count-dlq
spring.cloud.stream.kafka.streams.binder:
brokers: localhost #192.168.99.100
serdeError: sendToDlq

21
kafka-streams-samples/kafka-streams-dlq-sample/src/main/resources/by-framework-decoding.yml
View File

@@ -1,9 +1,12 @@
spring.cloud.stream.bindings.output.contentType: application/json
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms: 1000
spring.cloud.stream.kafka.streams.binder.configuration:
default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
application.id: default
spring.cloud.stream.kafka.streams.binder:
brokers: localhost #192.168.99.100
serdeError: sendToDlq
configuration:
commit.interval.ms: 1000
default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
application.id: default
spring.cloud.stream.bindings.output:
destination: counts
producer:
@@ -17,9 +20,9 @@ spring.cloud.stream.bindings.input:
headerMode: raw
useNativeDecoding: false
spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqName: words-count-dlq
spring.cloud.stream.kafka.streams.binder:
brokers: localhost #192.168.99.100
zkNodes: localhost #192.168.99.100
serdeError: sendToDlq

2
kafka-streams-samples/kafka-streams-message-channel/src/main/resources/application.yml
View File

@@ -5,8 +5,6 @@ spring.cloud.stream.kafka.streams.binder.configuration:
default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.bindings.singleOutput:
destination: counts
producer:
#useNativeEncoding: true
spring.cloud.stream.bindings.binding2:
destination: words
consumer:

2
kafka-streams-samples/kafka-streams-product-tracker/src/main/resources/application.yml
View File

@@ -8,8 +8,6 @@ spring.cloud.stream.kafka.streams:
keySerde: org.apache.kafka.common.serialization.Serdes$IntegerSerde
spring.cloud.stream.bindings.output:
destination: product-counts
producer:
#useNativeEncoding: true
spring.cloud.stream.bindings.input:
destination: products
consumer: