diff --git a/multi-binder-samples/multibinder-kafka-streams/.mvn b/multi-binder-samples/multi-binder-kafka-streams/.mvn similarity index 100% rename from multi-binder-samples/multibinder-kafka-streams/.mvn rename to multi-binder-samples/multi-binder-kafka-streams/.mvn diff --git a/multi-binder-samples/multibinder-kafka-streams/README.adoc b/multi-binder-samples/multi-binder-kafka-streams/README.adoc similarity index 67% rename from multi-binder-samples/multibinder-kafka-streams/README.adoc rename to multi-binder-samples/multi-binder-kafka-streams/README.adoc index 2b3f076..7bdae3d 100644 --- a/multi-binder-samples/multibinder-kafka-streams/README.adoc +++ b/multi-binder-samples/multi-binder-kafka-streams/README.adoc @@ -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 diff --git a/multi-binder-samples/multibinder-kafka-streams/docker-compose.yml b/multi-binder-samples/multi-binder-kafka-streams/docker-compose.yml similarity index 90% rename from multi-binder-samples/multibinder-kafka-streams/docker-compose.yml rename to multi-binder-samples/multi-binder-kafka-streams/docker-compose.yml index e7082eb..0fa14da 100644 --- a/multi-binder-samples/multibinder-kafka-streams/docker-compose.yml +++ b/multi-binder-samples/multi-binder-kafka-streams/docker-compose.yml @@ -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: diff --git a/multi-binder-samples/multibinder-kafka-streams/mvnw b/multi-binder-samples/multi-binder-kafka-streams/mvnw similarity index 100% rename from multi-binder-samples/multibinder-kafka-streams/mvnw rename to multi-binder-samples/multi-binder-kafka-streams/mvnw diff --git a/multi-binder-samples/multibinder-kafka-streams/mvnw.cmd b/multi-binder-samples/multi-binder-kafka-streams/mvnw.cmd similarity index 100% rename from multi-binder-samples/multibinder-kafka-streams/mvnw.cmd rename to multi-binder-samples/multi-binder-kafka-streams/mvnw.cmd diff --git a/multi-binder-samples/multi-binder-kafka-streams/pom.xml b/multi-binder-samples/multi-binder-kafka-streams/pom.xml new file mode 100644 index 0000000..89fb830 --- /dev/null +++ b/multi-binder-samples/multi-binder-kafka-streams/pom.xml @@ -0,0 +1,127 @@ + + + 4.0.0 + + multi-binder-kafka-streams + 0.0.1-SNAPSHOT + jar + multi-binder-kafka-streams + Spring Cloud Stream Multibinder Two Kafka Clusters (Kafka Streams) Sample + + + org.springframework.boot + spring-boot-starter-parent + 2.2.0.RELEASE + + + + + Hoxton.BUILD-SNAPSHOT + + + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + + + + org.springframework.cloud + spring-cloud-stream-binder-kafka + + + org.springframework.cloud + spring-cloud-stream-binder-kafka-streams + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.kafka + spring-kafka-test + test + + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release-local + + false + + + + + diff --git a/multi-binder-samples/multi-binder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java b/multi-binder-samples/multi-binder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java new file mode 100644 index 0000000..639061e --- /dev/null +++ b/multi-binder-samples/multi-binder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java @@ -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 process() { + return payload -> payload.toUpperCase(); + } + + + static class TestProducer { + + private AtomicBoolean semaphore = new AtomicBoolean(true); + + @Bean + public Supplier sendTestData() { + return () -> this.semaphore.getAndSet(!this.semaphore.get()) ? "foo" : "bar"; + } + } + + static class ConcatConfig { + + @Bean + public Function, KStream> concatFn() { + return input -> + input.mapValues(v -> v.concat(v)); + } + + } + + static class TestConsumer { + + private final Log logger = LogFactory.getLog(getClass()); + + @Bean + public Consumer receive() { + return s -> logger.info("Data received..." + s); + } + } +} diff --git a/multi-binder-samples/multibinder-kafka-streams/src/main/resources/application.yml b/multi-binder-samples/multi-binder-kafka-streams/src/main/resources/application.yml similarity index 78% rename from multi-binder-samples/multibinder-kafka-streams/src/main/resources/application.yml rename to multi-binder-samples/multi-binder-kafka-streams/src/main/resources/application.yml index 219ae70..1860e7d 100644 --- a/multi-binder-samples/multibinder-kafka-streams/src/main/resources/application.yml +++ b/multi-binder-samples/multi-binder-kafka-streams/src/main/resources/application.yml @@ -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 diff --git a/multi-binder-samples/multibinder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java b/multi-binder-samples/multi-binder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java similarity index 97% rename from multi-binder-samples/multibinder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java rename to multi-binder-samples/multi-binder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java index 01332fc..968f95f 100644 --- a/multi-binder-samples/multibinder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java +++ b/multi-binder-samples/multi-binder-kafka-streams/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java @@ -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()))); } } diff --git a/multi-binder-samples/multi-binder-two-kafka-clusters/pom.xml b/multi-binder-samples/multi-binder-two-kafka-clusters/pom.xml index e45d6ea..e3b6899 100644 --- a/multi-binder-samples/multi-binder-two-kafka-clusters/pom.xml +++ b/multi-binder-samples/multi-binder-two-kafka-clusters/pom.xml @@ -36,10 +36,6 @@ org.springframework.cloud spring-cloud-stream-binder-kafka - - org.springframework.cloud - spring-cloud-stream-binder-rabbit - org.springframework.boot spring-boot-starter-test diff --git a/multi-binder-samples/multibinder-kafka-streams/pom.xml b/multi-binder-samples/multibinder-kafka-streams/pom.xml deleted file mode 100644 index a060a97..0000000 --- a/multi-binder-samples/multibinder-kafka-streams/pom.xml +++ /dev/null @@ -1,52 +0,0 @@ - - - 4.0.0 - - multibinder-kafka-streams - 0.0.1-SNAPSHOT - jar - multibinder-kafka-streams - Spring Cloud Stream Multibinder Two Kafka Clusters (Kafka Streams) Sample - - - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. - - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - - - org.springframework.cloud - spring-cloud-stream-binder-kafka-core - - - org.springframework.cloud - spring-cloud-stream-binder-kafka-streams - - - org.springframework.boot - spring-boot-starter-test - test - - - org.springframework.kafka - spring-kafka-test - test - - - - - - - org.springframework.boot - spring-boot-maven-plugin - - - - - diff --git a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/BridgeTransformer.java b/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/BridgeTransformer.java deleted file mode 100644 index a05c448..0000000 --- a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/BridgeTransformer.java +++ /dev/null @@ -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 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 input) { - ObjectMapper mapper = new ObjectMapper(); - Serde 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.>as("test-events-snapshots").withKeySerde(Serdes.String()). - withValueSerde(Serdes.String()) - ); - } - } - - @RestController - public class FooController { - - @RequestMapping("/events") - public String events() { - - final ReadOnlyKeyValueStore topFiveStore = - interactiveQueryService.getQueryableStore("test-events-snapshots", QueryableStoreTypes.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(); - } - -} diff --git a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/DomainEvent.java b/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/DomainEvent.java deleted file mode 100644 index 8e8de2d..0000000 --- a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/DomainEvent.java +++ /dev/null @@ -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; - } -} diff --git a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java b/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java deleted file mode 100644 index 6359938..0000000 --- a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/MultibinderApplication.java +++ /dev/null @@ -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); - } - -} diff --git a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/Producers.java b/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/Producers.java deleted file mode 100644 index 229a215..0000000 --- a/multi-binder-samples/multibinder-kafka-streams/src/main/java/multibinder/Producers.java +++ /dev/null @@ -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 domainEventSerde = new JsonSerde<>(DomainEvent.class, mapper); - - - Map 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 pf = new DefaultKafkaProducerFactory<>(props); - KafkaTemplate template = new KafkaTemplate<>(pf, true); - template.setDefaultTopic("foobar"); - - template.sendDefault("", ddEvent); - } -} diff --git a/multi-binder-samples/pom.xml b/multi-binder-samples/pom.xml index 74d1042..6241382 100644 --- a/multi-binder-samples/pom.xml +++ b/multi-binder-samples/pom.xml @@ -12,7 +12,7 @@ multi-binder-kafka-rabbit multi-binder-two-kafka-clusters kafka-multibinder-jaas - multibinder-kafka-streams + multi-binder-kafka-streams