home/spring-cloud-stream-samples
2
0
Fork 0
mirror of https://github.com/wahyd4/spring-cloud-stream-samples.git synced 2026-08-09 05:16:18 +10:00
Code Issues Packages Projects Releases Wiki Activity

Refactor multibinder Kafka Streams application

Browse Source
This commit is contained in:
Soby Chacko
2019-10-30 13:53:10 -04:00
parent dad51e3d40
commit 3eed560762
16 changed files with 228 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
multi-binder-samples/multibinder-kafka-streams/.mvn → multi-binder-samples/multi-binder-kafka-streams/.mvn
View File
multi-binder-samples/multibinder-kafka-streams/README.adoc → multi-binder-samples/multi-binder-kafka-streams/README.adoc
+5 -10
View File
@@ -1,7 +1,7 @@
== Spring Cloud Stream Multibinder Application with Different Systems
This example shows how to run a Spring Cloud Stream application with the same binder type configured for two separate Kafka clusters.
One cluster is also made use of processing data through a Kafka Streams application.
## Running the application
@@ -16,11 +16,11 @@ Local ports mapped for kafka are 9092 and 9093 (Zookeeper local parts mapped are
* `./mvnw clean package`
The sample comes with a convenient test producer and consumer to see the processor in action.
After running the program, watch your console, every second some data is sent to Kafka cluster 1 and it is received through Kafka cluster 2.
After running the program, watch your console, every second some data is sent to Kafka cluster 1 and it is received through Kafka cluster 2 which is concatenated with the same string and then consumed through another sink.
To run the example, command line parameters for the Zookeeper ensembles and Kafka clusters must be provided, as in the following example:
```
java -jar target/multibinder-kafka-streams-0.0.1-SNAPSHOT.jar --kafkaBroker1=localhost:9092 --zk1=localhost:2181 --kafkaBroker2=localhost:9093 --zk2=localhost:2182```
java -jar target/multi-binder-kafka-streams-0.0.1-SNAPSHOT.jar --kafkaBroker1=localhost:9092 --zk1=localhost:2181 --kafkaBroker2=localhost:9093 --zk2=localhost:2182```
```
Alternatively, the default values of `localhost:9092` and `localhost:2181` can be provided for both clusters.
@@ -28,16 +28,11 @@ Assuming you are running two dockerized Kafka clusters as above.
Issue the following commands:
`docker exec -it kafka-multibinder-1 /opt/kafka/bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic dataIn`
`docker exec -it kafka-multib-inder-1 /opt/kafka/bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic dataIn`
On another terminal:
`docker exec -it kafka-multibinder-2 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --topic dataOut`
`docker exec -it kafka-multi-binder-2 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --topic dataOut`
Enter some text on the first one and the same text appears on the second one.
## Running the Kafka Streams processor:
Run the stand-alone Producers application a few times to to generate some data.
Then go to the URL: http://localhost:8080/events
multi-binder-samples/multibinder-kafka-streams/docker-compose.yml → multi-binder-samples/multi-binder-kafka-streams/docker-compose.yml
+2 -2
View File
@@ -2,7 +2,7 @@ version: '3'
services:
kafka1:
image: wurstmeister/kafka
container_name: kafka-multibinder-1
container_name: kafka-multib-inder-1
ports:
- "9092:9092"
environment:
@@ -19,7 +19,7 @@ services:
- KAFKA_ADVERTISED_HOST_NAME=zookeeper1
kafka2:
image: wurstmeister/kafka
container_name: kafka-multibinder-2
container_name: kafka-multi-binder-2
ports:
- "9093:9092"
environment:
multi-binder-samples/multibinder-kafka-streams/mvnw → multi-binder-samples/multi-binder-kafka-streams/mvnw
Vendored
View File
multi-binder-samples/multibinder-kafka-streams/mvnw.cmd → multi-binder-samples/multi-binder-kafka-streams/mvnw.cmd
Vendored
View File
multi-binder-samples/multi-binder-kafka-streams/pom.xml
+127
View File
@@ -0,0 +1,127 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>multi-binder-kafka-streams</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>multi-binder-kafka-streams</name>
<description>Spring Cloud Stream Multibinder Two Kafka Clusters (Kafka Streams) Sample</description>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</project>
multi-binder-samples/multi-binder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java
+78
View File
@@ -0,0 +1,78 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed 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
*
* https://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 multibinder;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class MultibinderApplication {
@Autowired
private InteractiveQueryService interactiveQueryService;
public static void main(String[] args) {
SpringApplication.run(MultibinderApplication.class, args);
}
@Bean
public Function<String, String> process() {
return payload -> payload.toUpperCase();
}
static class TestProducer {
private AtomicBoolean semaphore = new AtomicBoolean(true);
@Bean
public Supplier<String> sendTestData() {
return () -> this.semaphore.getAndSet(!this.semaphore.get()) ? "foo" : "bar";
}
}
static class ConcatConfig {
@Bean
public Function<KStream<?, String>, KStream<?, String>> concatFn() {
return input ->
input.mapValues(v -> v.concat(v));
}
}
static class TestConsumer {
private final Log logger = LogFactory.getLog(getClass());
@Bean
public Consumer<String> receive() {
return s -> logger.info("Data received..." + s);
}
}
}
multi-binder-samples/multibinder-kafka-streams/src/main/resources/application.yml → multi-binder-samples/multi-binder-kafka-streams/src/main/resources/application.yml
+14 -10
View File
@@ -1,6 +1,8 @@
spring:
cloud:
stream:
function:
definition: sendTestData;process;receive;concatFn
kafka:
streams:
binder:
@@ -11,24 +13,26 @@ spring:
default.value.serde: org.apache.kafka.common.serialization.Serdes$BytesSerde
commit.interval.ms: 1000
bindings:
input:
process-in-0:
destination: dataIn
binder: kafka1
group: testGroup
output:
process-out-0:
destination: dataOut
binder: kafka2
#Test source binding (used for testing)
output1:
sendTestData-out-0:
destination: dataIn
binder: kafka1
#Test sink binding (used for testing)
input1:
destination: dataOut
binder: kafka2
input2:
destination: foobar
concatFn-in-0:
destination: dataOut
binder: kafka3
concatFn-out-0:
destination: concatOut
binder: kafka3
#Test sink binding (used for testing)
receive-in-0:
destination: concatOut
binder: kafka2
binders:
kafka1:
type: kafka
multi-binder-samples/multibinder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java → multi-binder-samples/multi-binder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java
+1 -1
View File
@@ -96,7 +96,7 @@ public class TwoKafkaBindersApplicationTest {
//receiving test message sent by the test producer in the application
Message<?> receive = dataConsumer.receive(60_000);
Assert.assertThat(receive, Matchers.notNullValue());
Assert.assertThat(receive.getPayload(), CoreMatchers.equalTo("foo".getBytes()));
Assert.assertThat(receive.getPayload(), CoreMatchers.anyOf(equalTo("FOO".getBytes()), equalTo("BAR".getBytes())));
}
}
multi-binder-samples/multi-binder-two-kafka-clusters/pom.xml
-4
View File
@@ -36,10 +36,6 @@
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
multi-binder-samples/multibinder-kafka-streams/pom.xml
-52
View File
@@ -1,52 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>multibinder-kafka-streams</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>multibinder-kafka-streams</name>
<description>Spring Cloud Stream Multibinder Two Kafka Clusters (Kafka Streams) Sample</description>
<parent>
<groupId>io.spring.cloud.stream.sample</groupId>
<artifactId>spring-cloud-stream-samples-parent</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../..</relativePath>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/BridgeTransformer.java
-153
View File
@@ -1,153 +0,0 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed 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
*
* https://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 multibinder;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.kstream.ForeachAction;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Serialized;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.core.MessageSource;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* @author Marius Bogoevici
* @author Soby Chacko
*/
@EnableBinding(Processor.class)
public class BridgeTransformer {
@Autowired
private InteractiveQueryService interactiveQueryService;
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public Object transform(Object payload) {
return payload;
}
//Following source is used as test producer.
@EnableBinding(TestSource.class)
static class TestProducer {
private AtomicBoolean semaphore = new AtomicBoolean(true);
@Bean
@InboundChannelAdapter(channel = TestSource.OUTPUT, poller = @Poller(fixedDelay = "1000"))
public MessageSource<String> sendTestData() {
return () ->
new GenericMessage<>(this.semaphore.getAndSet(!this.semaphore.get()) ? "foo" : "bar");
}
}
//Following sink is used as test consumer for the above processor. It logs the data received through the processor.
@EnableBinding(TestSink.class)
static class TestConsumer {
private final Log logger = LogFactory.getLog(getClass());
@StreamListener(TestSink.INPUT)
public void receive(String data) {
logger.info("Data received..." + data);
}
}
@EnableBinding(KafkaStreamsProcessorX.class)
static class KafkaStreamsAggregateSampleApplication {
@StreamListener("input2")
public void process(KStream<Object, DomainEvent> input) {
ObjectMapper mapper = new ObjectMapper();
Serde<DomainEvent> domainEventSerde = new JsonSerde<>( DomainEvent.class, mapper );
input
.groupBy(
(s, domainEvent) -> domainEvent.boardUuid,
Serialized.with(null, domainEventSerde))
.aggregate(
String::new,
(s, domainEvent, board) -> board.concat(domainEvent.eventType),
Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("test-events-snapshots").withKeySerde(Serdes.String()).
withValueSerde(Serdes.String())
);
}
}
@RestController
public class FooController {
@RequestMapping("/events")
public String events() {
final ReadOnlyKeyValueStore<String, String> topFiveStore =
interactiveQueryService.getQueryableStore("test-events-snapshots", QueryableStoreTypes.<String, String>keyValueStore());
return topFiveStore.get("12345");
}
}
interface TestSink {
String INPUT = "input1";
@Input(INPUT)
SubscribableChannel input1();
}
interface TestSource {
String OUTPUT = "output1";
@Output(TestSource.OUTPUT)
MessageChannel output();
}
interface KafkaStreamsProcessorX {
@Input("input2")
KStream<?, ?> input2();
}
}
multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/DomainEvent.java
-43
View File
@@ -1,43 +0,0 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed 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
*
* https://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 multibinder;
/**
* @author Soby Chacko
*/
public class DomainEvent {
String eventType;
String boardUuid;
public String getEventType() {
return eventType;
}
public void setEventType(String eventType) {
this.eventType = eventType;
}
public String getBoardUuid() {
return boardUuid;
}
public void setBoardUuid(String boardUuid) {
this.boardUuid = boardUuid;
}
}
multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java
-29
View File
@@ -1,29 +0,0 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed 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
*
* https://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 multibinder;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class MultibinderApplication {
public static void main(String[] args) {
SpringApplication.run(MultibinderApplication.class, args);
}
}
multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/Producers.java
-60
View File
@@ -1,60 +0,0 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed 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
*
* https://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 multibinder;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonSerde;
import java.util.HashMap;
import java.util.Map;
/**
* @author Soby Chacko
*/
public class Producers {
public static void main(String... args) {
ObjectMapper mapper = new ObjectMapper();
Serde<DomainEvent> domainEventSerde = new JsonSerde<>(DomainEvent.class, mapper);
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093");
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, domainEventSerde.serializer().getClass());
DomainEvent ddEvent = new DomainEvent();
ddEvent.setBoardUuid("12345");
ddEvent.setEventType("thisisanevent");
DefaultKafkaProducerFactory<String, DomainEvent> pf = new DefaultKafkaProducerFactory<>(props);
KafkaTemplate<String, DomainEvent> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("foobar");
template.sendDefault("", ddEvent);
}
}
multi-binder-samples/pom.xml
+1 -1
View File
@@ -12,7 +12,7 @@
<module>multi-binder-kafka-rabbit</module>
<module>multi-binder-two-kafka-clusters</module>
<module>kafka-multibinder-jaas</module>
<module>multibinder-kafka-streams</module>
<module>multi-binder-kafka-streams</module>
</modules>
</project>