Refactoring mult-binder-two-kafka-clusters sample

multi-binder-samples/multi-binder-two-kafka-clusters/README.adoc

@@ -20,7 +20,7 @@ After running the program, watch your console, every second some data is sent to
 
 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-differentsystems-0.0.1-SNAPSHOT.jar --kafkaBroker1=localhost:9092 --zk1=localhost:2181 --kafkaBroker2=localhost:9093 --zk2=localhost:2182
+java -jar target/multi-binder-differentsystems-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.
@@ -29,10 +29,10 @@ 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-multi-binder-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.

multi-binder-samples/multi-binder-two-kafka-clusters/docker-compose.yml

@@ -2,7 +2,7 @@ version: '3'
 services:
   kafka1:
     image: wurstmeister/kafka
-    container_name: kafka-multibinder-1
+    container_name: kafka-multi-binder-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/multi-binder-two-kafka-clusters/pom.xml

@@ -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-two-kafka-clusters</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+	<name>multi-binder-two-kafka-clusters</name>
+	<description>Spring Cloud Stream Multibinder Two Kafka Clusters 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-rabbit</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-two-kafka-clusters/src/main/java/multibinder/MultibinderApplication.java

@@ -16,8 +16,16 @@
 
 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.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
 
 @SpringBootApplication
 public class MultibinderApplication {
@@ -26,4 +34,29 @@ public class MultibinderApplication {
 		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 TestConsumer {
+
+		private final Log logger = LogFactory.getLog(getClass());
+
+		@Bean
+		public Consumer<String> receive() {
+			return s -> logger.info("Data received..." + s);
+		}
+	}
+
 }

multi-binder-samples/multi-binder-two-kafka-clusters/src/main/resources/application.yml

@@ -0,0 +1,41 @@
+spring:
+  cloud:
+    stream:
+      bindings:
+        process-in-0:
+          destination: dataIn
+          binder: kafka1
+        process-out-0:
+          destination: dataOut
+          binder: kafka2
+        #Test sink binding (used for testing)
+        sendTestData-out-0:
+          destination: dataIn
+          binder: kafka1
+        #Test sink binding (used for testing)
+        receive-in-0:
+          destination: dataOut
+          binder: kafka2
+      function:
+        definition: sendTestData;process;receive
+      binders:
+        kafka1:
+          type: kafka
+          environment:
+            spring:
+              cloud:
+                stream:
+                  kafka:
+                    binder:
+                      brokers: ${kafkaBroker1}
+                      zkNodes: ${zk1}
+        kafka2:
+          type: kafka
+          environment:
+            spring:
+              cloud:
+                stream:
+                  kafka:
+                    binder:
+                      brokers: ${kafkaBroker2}
+                      zkNodes: ${zk2}

multi-binder-samples/multi-binder-two-kafka-clusters/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java

@@ -95,7 +95,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/multibinder-two-kafka-clusters/pom.xml

@@ -1,48 +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-two-kafka-clusters</artifactId>
-	<version>0.0.1-SNAPSHOT</version>
-	<packaging>jar</packaging>
-	<name>multibinder-two-kafka-clusters</name>
-	<description>Spring Cloud Stream Multibinder Two Kafka Clusters 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.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-two-kafka-clusters/src/main/java/multibinder/BridgeTransformer.java

@@ -1,94 +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.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-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.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.messaging.MessageChannel;
-import org.springframework.messaging.SubscribableChannel;
-import org.springframework.messaging.handler.annotation.SendTo;
-import org.springframework.messaging.support.GenericMessage;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-
-/**
- * @author Marius Bogoevici
- * @author Soby Chacko
- */
-@EnableBinding(Processor.class)
-public class BridgeTransformer {
-
-	@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);
-		}
-	}
-
-	interface TestSink {
-
-		String INPUT = "input1";
-
-		@Input(INPUT)
-		SubscribableChannel input1();
-
-	}
-
-	interface TestSource {
-
-		String OUTPUT = "output1";
-
-		@Output(TestSource.OUTPUT)
-		MessageChannel output();
-
-	}
-}

multi-binder-samples/multibinder-two-kafka-clusters/src/main/resources/application.yml

@@ -1,40 +0,0 @@
-spring:
-  cloud:
-    stream:
-      bindings:
-        input:
-          destination: dataIn
-          binder: kafka1
-          group: testGroup
-        output:
-          destination: dataOut
-          binder: kafka2
-        #Test sink binding (used for testing)
-        output1:
-          destination: dataIn
-          binder: kafka1
-        #Test sink binding (used for testing)
-        input1:
-          destination: dataOut
-          binder: kafka2
-      binders:
-        kafka1:
-         type: kafka
-         environment:
-           spring:
-             cloud:
-               stream:
-                kafka:
-                  binder:
-                    brokers: ${kafkaBroker1}
-                    zkNodes: ${zk1}
-        kafka2:
-         type: kafka
-         environment:
-           spring:
-             cloud:
-               stream:
-                 kafka:
-                   binder:
-                     brokers: ${kafkaBroker2}
-                     zkNodes: ${zk2}

multi-binder-samples/pom.xml

@@ -10,7 +10,7 @@
 
     <modules>
         <module>multi-binder-kafka-rabbit</module>
-        <module>multibinder-two-kafka-clusters</module>
+        <module>multi-binder-two-kafka-clusters</module>
         <module>kafka-multibinder-jaas</module>
         <module>multibinder-kafka-streams</module>
     </modules>