diff --git a/partitioning-samples/README.adoc b/partitioning-samples/README.adoc deleted file mode 100644 index 80dfc22..0000000 --- a/partitioning-samples/README.adoc +++ /dev/null @@ -1,87 +0,0 @@ -Spring Cloud Stream Partitioning Sample -======================================== - -This is a collection of applications that demonstrates how partitioning works in Spring Cloud Stream. - -## Quick introduction - -The producer used in the sample produces messages with text that has a length of 1, 2, 3 or 4. -There is a configuration in the producer's application.yml file for `partition-key-expression` that uses the length of the payload minus 1 as the partition key expression to use. -This value will be used by the binder for selecting the correct partition based on the total number of partitions configured on the destination at the broker. -We use 4 partitions for this demo. - -There is a common producer module called partitioning-producer and then there is a consumer for kafka and rabbit - partitioning-consumer-kafka and partitioning-consumer-rabbit respectively. -Follow the instructions below to run the demo for Kafka or RabbitMQ. - -## Running the sample for Kafka - -The following instructions assume that you are running Kafka as a Docker image. - -* `docker-compose up -d` - -* cd partitioning-consumer-kafka - -* `./mvnw clean package` - -* `java -jar target/partitioning-consumer-kafka-0.0.1-SNAPSHOT.jar --server.port=9008` - -On another termimal start another instance of the consumer. - -* `java -jar target/partitioning-consumer-kafka-0.0.1-SNAPSHOT.jar --server.port=9009` - -* cd ../partitioning-producer - -* `./mvnw clean package` - -* `java -jar target/partitioning-producer-0.0.1-SNAPSHOT.jar --server.port=9010` - -Producer sends messages randomly that has string length of 1, 2, 3, or 4. -Watch the consumer console logs and verify that the correct partitions are receiving the messages. -The log message has the payload and partition information in it. - -Once you are done testing, stop all the instances. - -* `docker-compose down` - -## Running the sample for Rabbit - -The following instructions assume that you are running Rabbit as a Docker image. - -Make sure that you are at the root directory of partitioning samples (partitioning-samples) - -Rabbit partitioning demo is slightly different from Kafka. -We need to spin up 4 consumers for each of the four partitions. - -* `docker-compose -f docker-compose-rabbit.yml up -d` - -* cd partitioning-consumer-rabbit - -* `./mvnw clean package` - -* `java -jar target/partitioning-consumer-rabbit-0.0.1-SNAPSHOT.jar --server.port=9005` - -On another terminal start another instance of the consumer. - -* `java -jar target/partitioning-consumer-rabbit-0.0.1-SNAPSHOT.jar --server.port=9006 --spring.cloud.stream.bindings.input.consumer.instanceIndex=1` - -On another terminal start another instance of the consumer. - -* `java -jar target/partitioning-consumer-rabbit-0.0.1-SNAPSHOT.jar --server.port=9007 --spring.cloud.stream.bindings.input.consumer.instanceIndex=2` - -On another terminal start yet another instance of the consumer. - -* `java -jar target/partitioning-consumer-rabbit-0.0.1-SNAPSHOT.jar --server.port=9008 --spring.cloud.stream.bindings.input.consumer.instanceIndex=3` - -* cd ../partitioning-producer - -* `./mvnw clean package -P rabbit-binder` - -* `java -jar target/partitioning-producer-0.0.1-SNAPSHOT.jar --server.port=9010` - -Producer sends messages randomly that has string length of 1, 2, 3, or 4. -Watch the consumer console logs and verify that the correct instances are receiving the messages. -The first consumer we started should receive messages with a string length of 1 (partition-0), second consumer with `instanceIndex` set to 1 should receive messages with string length of 2 (partittion-1) so on and so forth. - -Once you are done testing, stop all the instances. - -* `docker-compose -f docker-compose-rabbit.yml down` \ No newline at end of file diff --git a/partitioning-samples/partitioning-producer-sample/.mvn/jvm.config b/partitioning-samples/kafka-partitioning/.mvn/jvm.config similarity index 100% rename from partitioning-samples/partitioning-producer-sample/.mvn/jvm.config rename to partitioning-samples/kafka-partitioning/.mvn/jvm.config diff --git a/partitioning-samples/partitioning-producer-sample/.mvn/maven.config b/partitioning-samples/kafka-partitioning/.mvn/maven.config similarity index 100% rename from partitioning-samples/partitioning-producer-sample/.mvn/maven.config rename to partitioning-samples/kafka-partitioning/.mvn/maven.config diff --git a/partitioning-samples/partitioning-producer-sample/.mvn/wrapper/maven-wrapper.jar b/partitioning-samples/kafka-partitioning/.mvn/wrapper/maven-wrapper.jar similarity index 100% rename from partitioning-samples/partitioning-producer-sample/.mvn/wrapper/maven-wrapper.jar rename to partitioning-samples/kafka-partitioning/.mvn/wrapper/maven-wrapper.jar diff --git a/partitioning-samples/partitioning-producer-sample/.mvn/wrapper/maven-wrapper.properties b/partitioning-samples/kafka-partitioning/.mvn/wrapper/maven-wrapper.properties similarity index 100% rename from partitioning-samples/partitioning-producer-sample/.mvn/wrapper/maven-wrapper.properties rename to partitioning-samples/kafka-partitioning/.mvn/wrapper/maven-wrapper.properties diff --git a/partitioning-samples/kafka-partitioning/README.adoc b/partitioning-samples/kafka-partitioning/README.adoc new file mode 100644 index 0000000..b8a86a1 --- /dev/null +++ b/partitioning-samples/kafka-partitioning/README.adoc @@ -0,0 +1,40 @@ +Spring Cloud Stream Partitioning Sample +======================================== + +These sample apps demonstrate how partitioning works in Spring Cloud Stream with Apache Kafka. +Sample producer and consumer are provided. + +## Quick introduction + +The producer used in the sample produces messages with text that has a length of 1, 2, 3 or 4. +There is a configuration in the producer's application.yml file for `partition-key-expression` that uses the length of the payload minus 1 as the partition key expression to use. +This value will be used by the binder for selecting the correct partition based on the total number of partitions configured on the destination at the broker. +We use 4 partitions for this demo. + +## Running the samples + +The following instructions assume that you are running Kafka as a Docker image. + +* `docker-compose up -d` + +* Ensure that you are in the directory `kafka-partitioning` + +* `./mvnw clean package` + +* `java -jar partitioning-consumer-sample-kafka/target/partitioning-consumer-sample-kafka-0.0.1-SNAPSHOT.jar --server.port=9008` + +On another termimal start another instance of the consumer. + +* `java -jar partitioning-consumer-sample-kafka/target/partitioning-consumer-sample-kafka-0.0.1-SNAPSHOT.jar --server.port=9009` + +On another terminal start the producer. + +* `java -jar partitioning-producer-sample-kafka/target/partitioning-producer-sample-kafka-0.0.1-SNAPSHOT.jar --server.port=9010` + +Producer sends messages randomly that has string length of 1, 2, 3, or 4. +Watch the consumer console logs and verify that the correct partitions are receiving the messages. +The log message has the payload and partition information in it. + +Once you are done testing, stop all the instances. + +* `docker-compose down` diff --git a/partitioning-samples/docker-compose.yml b/partitioning-samples/kafka-partitioning/docker-compose.yml similarity index 100% rename from partitioning-samples/docker-compose.yml rename to partitioning-samples/kafka-partitioning/docker-compose.yml diff --git a/partitioning-samples/mvnw b/partitioning-samples/kafka-partitioning/mvnw similarity index 100% rename from partitioning-samples/mvnw rename to partitioning-samples/kafka-partitioning/mvnw diff --git a/partitioning-samples/mvnw.cmd b/partitioning-samples/kafka-partitioning/mvnw.cmd similarity index 100% rename from partitioning-samples/mvnw.cmd rename to partitioning-samples/kafka-partitioning/mvnw.cmd diff --git a/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/pom.xml b/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/pom.xml new file mode 100644 index 0000000..9e5edae --- /dev/null +++ b/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/pom.xml @@ -0,0 +1,122 @@ + + + 4.0.0 + + partitioning-consumer-sample-kafka + 0.0.1-SNAPSHOT + jar + partitioning-consumer-sample-kafka + Spring Cloud Stream Partitioning producer Kafka + + + 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.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/partitioning-samples/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemoApplication.java b/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemoApplication.java similarity index 62% rename from partitioning-samples/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemoApplication.java rename to partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemoApplication.java index 485734f..a6805b6 100644 --- a/partitioning-samples/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemoApplication.java +++ b/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemoApplication.java @@ -16,14 +16,28 @@ package demo; +import java.util.function.Consumer; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.messaging.Message; @SpringBootApplication public class PartitioningKafkaDemoApplication { + private static final Logger logger = LoggerFactory.getLogger(PartitioningKafkaDemoApplication.class); + public static void main(String[] args) { SpringApplication.run(PartitioningKafkaDemoApplication.class, args); } + @Bean + public Consumer> listen() { + return message -> logger.info(message.getPayload() + " received from partition " + + message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID)); + } } diff --git a/partitioning-samples/partitioning-consumer-sample-kafka/src/main/resources/application.yml b/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/main/resources/application.yml similarity index 90% rename from partitioning-samples/partitioning-consumer-sample-kafka/src/main/resources/application.yml rename to partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/main/resources/application.yml index 4811154..17a5ec8 100644 --- a/partitioning-samples/partitioning-consumer-sample-kafka/src/main/resources/application.yml +++ b/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/main/resources/application.yml @@ -6,6 +6,6 @@ spring: autoAddPartitions: true minPartitionCount: 4 bindings: - input: + listen-in-0: destination: partitioned.destination group: myGroup \ No newline at end of file diff --git a/partitioning-samples/partitioning-consumer-sample-kafka/src/test/java/demo/ModuleApplicationTests.java b/partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/test/java/demo/ModuleApplicationTests.java similarity index 100% rename from partitioning-samples/partitioning-consumer-sample-kafka/src/test/java/demo/ModuleApplicationTests.java rename to partitioning-samples/kafka-partitioning/partitioning-consumer-sample-kafka/src/test/java/demo/ModuleApplicationTests.java diff --git a/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/pom.xml b/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/pom.xml new file mode 100644 index 0000000..e0a0136 --- /dev/null +++ b/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/pom.xml @@ -0,0 +1,122 @@ + + + 4.0.0 + + partitioning-producer-sample-kafka + 0.0.1-SNAPSHOT + jar + partitioning-producer-sample-kafka + Spring Cloud Stream Partitioning Kafka + + + 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.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/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/main/java/demo/producer/PartitioningProducerApplication.java b/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/main/java/demo/producer/PartitioningProducerApplication.java new file mode 100644 index 0000000..32ff448 --- /dev/null +++ b/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/main/java/demo/producer/PartitioningProducerApplication.java @@ -0,0 +1,48 @@ +package demo.producer; + +import java.util.Random; +import java.util.function.Supplier; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Soby Chacko + */ +@SpringBootApplication +public class PartitioningProducerApplication { + + private static final Random RANDOM = new Random(System.currentTimeMillis()); + + private static final Logger logger = LoggerFactory.getLogger(PartitioningProducerApplication.class); + + + public static void main(String[] args) { + SpringApplication.run(PartitioningProducerApplication.class, args); + } + + // We use a strategy so that this data will end up in a partition, + // P = L(x) - 1 where L is a length function on the payload. + private static final String[] data = new String[]{ + "f", "g", "h", //making them go to partition-0 by making a single char string + "fo", "go", "ho", + "foo", "goo", "hoo", + "fooz", "gooz", "hooz" + }; + + @Bean + public Supplier> generate() { + return () -> { + String value = data[RANDOM.nextInt(data.length)]; + logger.info("Sending: " + value); + return MessageBuilder.withPayload(value) + .setHeader("partitionKey", value.length()) + .build(); + }; + } +} diff --git a/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/main/resources/application.yml b/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/main/resources/application.yml new file mode 100644 index 0000000..21d7a72 --- /dev/null +++ b/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/main/resources/application.yml @@ -0,0 +1,10 @@ +spring: + cloud: + stream: + bindings: + generate-out-0: + destination: partitioned.destination + producer: + #payload string length - 1 is the partition where it will get stored + partition-key-expression: headers['partitionKey'] - 1 + partition-count: 4 \ No newline at end of file diff --git a/partitioning-samples/partitioning-producer-sample/src/test/java/demo/producer/ModuleApplicationTests.java b/partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/test/java/demo/producer/ModuleApplicationTests.java similarity index 100% rename from partitioning-samples/partitioning-producer-sample/src/test/java/demo/producer/ModuleApplicationTests.java rename to partitioning-samples/kafka-partitioning/partitioning-producer-sample-kafka/src/test/java/demo/producer/ModuleApplicationTests.java diff --git a/partitioning-samples/kafka-partitioning/pom.xml b/partitioning-samples/kafka-partitioning/pom.xml new file mode 100644 index 0000000..ddbf31a --- /dev/null +++ b/partitioning-samples/kafka-partitioning/pom.xml @@ -0,0 +1,16 @@ + + + 4.0.0 + io.spring.cloud.stream.sample + kafka-partitioning + 0.0.1-SNAPSHOT + pom + kafka-partitioning + Collection of Spring Cloud Stream partitioning Samples using Kafka binder + + + partitioning-producer-sample-kafka + partitioning-consumer-sample-kafka + + + diff --git a/partitioning-samples/partitioning-consumer-sample-kafka/pom.xml b/partitioning-samples/partitioning-consumer-sample-kafka/pom.xml deleted file mode 100644 index 088e781..0000000 --- a/partitioning-samples/partitioning-consumer-sample-kafka/pom.xml +++ /dev/null @@ -1,54 +0,0 @@ - - - 4.0.0 - - partitioning-consumer-sample-kafka - 0.0.1-SNAPSHOT - jar - partitioning-consumer-sample-kafka - Spring Cloud Stream Partitioning Kafka - - - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. - - - - - org.springframework.boot - spring-boot-starter-test - test - - - org.springframework.cloud - spring-cloud-stream-test-support - test - - - - - - kafka-binder - - true - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - - - - - - - - - org.springframework.boot - spring-boot-maven-plugin - - - - diff --git a/partitioning-samples/partitioning-consumer-sample-kafka/src/main/docker/assembly-kafka.xml b/partitioning-samples/partitioning-consumer-sample-kafka/src/main/docker/assembly-kafka.xml deleted file mode 100644 index 044f57a..0000000 --- a/partitioning-samples/partitioning-consumer-sample-kafka/src/main/docker/assembly-kafka.xml +++ /dev/null @@ -1,16 +0,0 @@ - - uppercase-transformer-kafka - - - - io.spring.cloud.stream.sample:partitioning-consumer-sample-kafka - - . - partitioning-consumer-sample-kafka.jar - - - - diff --git a/partitioning-samples/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemo.java b/partitioning-samples/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemo.java deleted file mode 100644 index ec13f14..0000000 --- a/partitioning-samples/partitioning-consumer-sample-kafka/src/main/java/demo/PartitioningKafkaDemo.java +++ /dev/null @@ -1,40 +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 demo; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.kafka.support.KafkaHeaders; -import org.springframework.messaging.handler.annotation.Header; -import org.springframework.messaging.handler.annotation.Payload; - -/** - * @author Soby Chacko - */ -@EnableBinding(Sink.class) -public class PartitioningKafkaDemo { - - private static final Logger logger = LoggerFactory.getLogger(PartitioningKafkaDemo.class); - - @StreamListener(Sink.INPUT) - public void listen(@Payload String in, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) { - logger.info(in + " received from partition " + partition); - } -} diff --git a/partitioning-samples/partitioning-consumer-sample-rabbit/mvnw b/partitioning-samples/partitioning-consumer-sample-rabbit/mvnw deleted file mode 100755 index 0ce08e9..0000000 --- a/partitioning-samples/partitioning-consumer-sample-rabbit/mvnw +++ /dev/null @@ -1,226 +0,0 @@ -#!/bin/sh -# ---------------------------------------------------------------------------- -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you 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. -# ---------------------------------------------------------------------------- - -# ---------------------------------------------------------------------------- -# Maven2 Start Up Batch script -# -# Required ENV vars: -# ------------------ -# JAVA_HOME - location of a JDK home dir -# -# Optional ENV vars -# ----------------- -# M2_HOME - location of maven2's installed home dir -# MAVEN_OPTS - parameters passed to the Java VM when running Maven -# e.g. to debug Maven itself, use -# set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000 -# MAVEN_SKIP_RC - flag to disable loading of mavenrc files -# ---------------------------------------------------------------------------- - -if [ -z "$MAVEN_SKIP_RC" ] ; then - - if [ -f /etc/mavenrc ] ; then - . /etc/mavenrc - fi - - if [ -f "$HOME/.mavenrc" ] ; then - . "$HOME/.mavenrc" - fi - -fi - -# OS specific support. $var _must_ be set to either true or false. -cygwin=false; -darwin=false; -mingw=false -case "`uname`" in - CYGWIN*) cygwin=true ;; - MINGW*) mingw=true;; - Darwin*) darwin=true - # Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home - # See https://developer.apple.com/library/mac/qa/qa1170/_index.html - if [ -z "$JAVA_HOME" ]; then - if [ -x "/usr/libexec/java_home" ]; then - export JAVA_HOME="`/usr/libexec/java_home`" - else - export JAVA_HOME="/Library/Java/Home" - fi - fi - ;; -esac - -if [ -z "$JAVA_HOME" ] ; then - if [ -r /etc/gentoo-release ] ; then - JAVA_HOME=`java-config --jre-home` - fi -fi - -if [ -z "$M2_HOME" ] ; then - ## resolve links - $0 may be a link to maven's home - PRG="$0" - - # need this for relative symlinks - while [ -h "$PRG" ] ; do - ls=`ls -ld "$PRG"` - link=`expr "$ls" : '.*-> \(.*\)$'` - if expr "$link" : '/.*' > /dev/null; then - PRG="$link" - else - PRG="`dirname "$PRG"`/$link" - fi - done - - saveddir=`pwd` - - M2_HOME=`dirname "$PRG"`/.. - - # make it fully qualified - M2_HOME=`cd "$M2_HOME" && pwd` - - cd "$saveddir" - # echo Using m2 at $M2_HOME -fi - -# For Cygwin, ensure paths are in UNIX format before anything is touched -if $cygwin ; then - [ -n "$M2_HOME" ] && - M2_HOME=`cygpath --unix "$M2_HOME"` - [ -n "$JAVA_HOME" ] && - JAVA_HOME=`cygpath --unix "$JAVA_HOME"` - [ -n "$CLASSPATH" ] && - CLASSPATH=`cygpath --path --unix "$CLASSPATH"` -fi - -# For Migwn, ensure paths are in UNIX format before anything is touched -if $mingw ; then - [ -n "$M2_HOME" ] && - M2_HOME="`(cd "$M2_HOME"; pwd)`" - [ -n "$JAVA_HOME" ] && - JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`" - # TODO classpath? -fi - -if [ -z "$JAVA_HOME" ]; then - javaExecutable="`which javac`" - if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then - # readlink(1) is not available as standard on Solaris 10. - readLink=`which readlink` - if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then - if $darwin ; then - javaHome="`dirname \"$javaExecutable\"`" - javaExecutable="`cd \"$javaHome\" && pwd -P`/javac" - else - javaExecutable="`readlink -f \"$javaExecutable\"`" - fi - javaHome="`dirname \"$javaExecutable\"`" - javaHome=`expr "$javaHome" : '\(.*\)/bin'` - JAVA_HOME="$javaHome" - export JAVA_HOME - fi - fi -fi - -if [ -z "$JAVACMD" ] ; then - if [ -n "$JAVA_HOME" ] ; then - if [ -x "$JAVA_HOME/jre/sh/java" ] ; then - # IBM's JDK on AIX uses strange locations for the executables - JAVACMD="$JAVA_HOME/jre/sh/java" - else - JAVACMD="$JAVA_HOME/bin/java" - fi - else - JAVACMD="`which java`" - fi -fi - -if [ ! -x "$JAVACMD" ] ; then - echo "Error: JAVA_HOME is not defined correctly." >&2 - echo " We cannot execute $JAVACMD" >&2 - exit 1 -fi - -if [ -z "$JAVA_HOME" ] ; then - echo "Warning: JAVA_HOME environment variable is not set." -fi - -CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher - -# traverses directory structure from process work directory to filesystem root -# first directory with .mvn subdirectory is considered project base directory -find_maven_basedir() { - - if [ -z "$1" ] - then - echo "Path not specified to find_maven_basedir" - return 1 - fi - - basedir="$1" - wdir="$1" - while [ "$wdir" != '/' ] ; do - if [ -d "$wdir"/.mvn ] ; then - basedir=$wdir - break - fi - # workaround for JBEAP-8937 (on Solaris 10/Sparc) - if [ -d "${wdir}" ]; then - wdir=`cd "$wdir/.."; pwd` - fi - # end of workaround - done - echo "${basedir}" -} - -# concatenates all lines of a file -concat_lines() { - if [ -f "$1" ]; then - echo "$(tr -s '\n' ' ' < "$1")" - fi -} - -BASE_DIR=`find_maven_basedir "$(pwd)"` -if [ -z "$BASE_DIR" ]; then - exit 1; -fi - -export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"} -echo $MAVEN_PROJECTBASEDIR -MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS" - -# For Cygwin, switch paths to Windows format before running java -if $cygwin; then - [ -n "$M2_HOME" ] && - M2_HOME=`cygpath --path --windows "$M2_HOME"` - [ -n "$JAVA_HOME" ] && - JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"` - [ -n "$CLASSPATH" ] && - CLASSPATH=`cygpath --path --windows "$CLASSPATH"` - [ -n "$MAVEN_PROJECTBASEDIR" ] && - MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"` -fi - -WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain - -"$JAVACMD" \ - $MAVEN_OPTS \ - -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \ - "-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \ - ${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@" - diff --git a/partitioning-samples/partitioning-consumer-sample-rabbit/mvnw.cmd b/partitioning-samples/partitioning-consumer-sample-rabbit/mvnw.cmd deleted file mode 100644 index 7ecd01d..0000000 --- a/partitioning-samples/partitioning-consumer-sample-rabbit/mvnw.cmd +++ /dev/null @@ -1,145 +0,0 @@ -@REM ---------------------------------------------------------------------------- -@REM Licensed to the Apache Software Foundation (ASF) under one -@REM or more contributor license agreements. See the NOTICE file -@REM distributed with this work for additional information -@REM regarding copyright ownership. The ASF licenses this file -@REM to you under the Apache License, Version 2.0 (the -@REM "License"); you may not use this file except in compliance -@REM with the License. You may obtain a copy of the License at -@REM -@REM https://www.apache.org/licenses/LICENSE-2.0 -@REM -@REM Unless required by applicable law or agreed to in writing, -@REM software distributed under the License is distributed on an -@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -@REM KIND, either express or implied. See the License for the -@REM specific language governing permissions and limitations -@REM under the License. -@REM ---------------------------------------------------------------------------- - -@REM ---------------------------------------------------------------------------- -@REM Maven2 Start Up Batch script -@REM -@REM Required ENV vars: -@REM JAVA_HOME - location of a JDK home dir -@REM -@REM Optional ENV vars -@REM M2_HOME - location of maven2's installed home dir -@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands -@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a key stroke before ending -@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven -@REM e.g. to debug Maven itself, use -@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000 -@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files -@REM ---------------------------------------------------------------------------- - -@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on' -@echo off -@REM enable echoing my setting MAVEN_BATCH_ECHO to 'on' -@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO% - -@REM set %HOME% to equivalent of $HOME -if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%") - -@REM Execute a user defined script before this one -if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre -@REM check for pre script, once with legacy .bat ending and once with .cmd ending -if exist "%HOME%\mavenrc_pre.bat" call "%HOME%\mavenrc_pre.bat" -if exist "%HOME%\mavenrc_pre.cmd" call "%HOME%\mavenrc_pre.cmd" -:skipRcPre - -@setlocal - -set ERROR_CODE=0 - -@REM To isolate internal variables from possible post scripts, we use another setlocal -@setlocal - -@REM ==== START VALIDATION ==== -if not "%JAVA_HOME%" == "" goto OkJHome - -echo. -echo Error: JAVA_HOME not found in your environment. >&2 -echo Please set the JAVA_HOME variable in your environment to match the >&2 -echo location of your Java installation. >&2 -echo. -goto error - -:OkJHome -if exist "%JAVA_HOME%\bin\java.exe" goto init - -echo. -echo Error: JAVA_HOME is set to an invalid directory. >&2 -echo JAVA_HOME = "%JAVA_HOME%" >&2 -echo Please set the JAVA_HOME variable in your environment to match the >&2 -echo location of your Java installation. >&2 -echo. -goto error - -@REM ==== END VALIDATION ==== - -:init - -set MAVEN_CMD_LINE_ARGS=%* - -@REM Find the project base dir, i.e. the directory that contains the folder ".mvn". -@REM Fallback to current working directory if not found. - -set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR% -IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir - -set EXEC_DIR=%CD% -set WDIR=%EXEC_DIR% -:findBaseDir -IF EXIST "%WDIR%"\.mvn goto baseDirFound -cd .. -IF "%WDIR%"=="%CD%" goto baseDirNotFound -set WDIR=%CD% -goto findBaseDir - -:baseDirFound -set MAVEN_PROJECTBASEDIR=%WDIR% -cd "%EXEC_DIR%" -goto endDetectBaseDir - -:baseDirNotFound -set MAVEN_PROJECTBASEDIR=%EXEC_DIR% -cd "%EXEC_DIR%" - -:endDetectBaseDir - -IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig - -@setlocal EnableExtensions EnableDelayedExpansion -for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a -@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS% - -:endReadAdditionalConfig - -SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe" - -set WRAPPER_JAR="".\.mvn\wrapper\maven-wrapper.jar"" -set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain - -%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CMD_LINE_ARGS% -if ERRORLEVEL 1 goto error -goto end - -:error -set ERROR_CODE=1 - -:end -@endlocal & set ERROR_CODE=%ERROR_CODE% - -if not "%MAVEN_SKIP_RC%" == "" goto skipRcPost -@REM check for post script, once with legacy .bat ending and once with .cmd ending -if exist "%HOME%\mavenrc_post.bat" call "%HOME%\mavenrc_post.bat" -if exist "%HOME%\mavenrc_post.cmd" call "%HOME%\mavenrc_post.cmd" -:skipRcPost - -@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on' -if "%MAVEN_BATCH_PAUSE%" == "on" pause - -if "%MAVEN_TERMINATE_CMD%" == "on" exit %ERROR_CODE% - -exit /B %ERROR_CODE% diff --git a/partitioning-samples/partitioning-consumer-sample-rabbit/pom.xml b/partitioning-samples/partitioning-consumer-sample-rabbit/pom.xml deleted file mode 100644 index 42f39c5..0000000 --- a/partitioning-samples/partitioning-consumer-sample-rabbit/pom.xml +++ /dev/null @@ -1,54 +0,0 @@ - - - 4.0.0 - - partitioning-consumer-sample-rabbit - 0.0.1-SNAPSHOT - jar - partitioning-consumer-sample-rabbit - Spring Cloud Stream Partitioning Rabbit - - - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. - - - - - org.springframework.boot - spring-boot-starter-test - test - - - org.springframework.cloud - spring-cloud-stream-test-support - test - - - - - - rabbit-binder - - true - - - - org.springframework.cloud - spring-cloud-stream-binder-rabbit - - - - - - - - - org.springframework.boot - spring-boot-maven-plugin - - - - diff --git a/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/docker/assembly-rabbit.xml b/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/docker/assembly-rabbit.xml deleted file mode 100644 index 4a6e788..0000000 --- a/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/docker/assembly-rabbit.xml +++ /dev/null @@ -1,16 +0,0 @@ - - uppercase-transformer-kafka - - - - io.spring.cloud.stream.sample:partitioning-consumer-sample-rabbit - - . - partitioning-consumer-sample-rabbit.jar - - - - diff --git a/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemo.java b/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemo.java deleted file mode 100644 index 9cee7c0..0000000 --- a/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemo.java +++ /dev/null @@ -1,40 +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 demo; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.amqp.support.AmqpHeaders; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.messaging.handler.annotation.Header; -import org.springframework.messaging.handler.annotation.Payload; - -/** - * @author Soby Chacko - */ -@EnableBinding(Sink.class) -public class PartitioningRabbitDemo { - - private static final Logger logger = LoggerFactory.getLogger(PartitioningRabbitDemo.class); - - @StreamListener(Sink.INPUT) - public void listen(@Payload String in, @Header(AmqpHeaders.CONSUMER_QUEUE) String partition) { - logger.info(in + " received from partition " + partition); - } -} diff --git a/partitioning-samples/partitioning-producer-sample/mvnw b/partitioning-samples/partitioning-producer-sample/mvnw deleted file mode 100755 index 0ce08e9..0000000 --- a/partitioning-samples/partitioning-producer-sample/mvnw +++ /dev/null @@ -1,226 +0,0 @@ -#!/bin/sh -# ---------------------------------------------------------------------------- -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you 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. -# ---------------------------------------------------------------------------- - -# ---------------------------------------------------------------------------- -# Maven2 Start Up Batch script -# -# Required ENV vars: -# ------------------ -# JAVA_HOME - location of a JDK home dir -# -# Optional ENV vars -# ----------------- -# M2_HOME - location of maven2's installed home dir -# MAVEN_OPTS - parameters passed to the Java VM when running Maven -# e.g. to debug Maven itself, use -# set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000 -# MAVEN_SKIP_RC - flag to disable loading of mavenrc files -# ---------------------------------------------------------------------------- - -if [ -z "$MAVEN_SKIP_RC" ] ; then - - if [ -f /etc/mavenrc ] ; then - . /etc/mavenrc - fi - - if [ -f "$HOME/.mavenrc" ] ; then - . "$HOME/.mavenrc" - fi - -fi - -# OS specific support. $var _must_ be set to either true or false. -cygwin=false; -darwin=false; -mingw=false -case "`uname`" in - CYGWIN*) cygwin=true ;; - MINGW*) mingw=true;; - Darwin*) darwin=true - # Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home - # See https://developer.apple.com/library/mac/qa/qa1170/_index.html - if [ -z "$JAVA_HOME" ]; then - if [ -x "/usr/libexec/java_home" ]; then - export JAVA_HOME="`/usr/libexec/java_home`" - else - export JAVA_HOME="/Library/Java/Home" - fi - fi - ;; -esac - -if [ -z "$JAVA_HOME" ] ; then - if [ -r /etc/gentoo-release ] ; then - JAVA_HOME=`java-config --jre-home` - fi -fi - -if [ -z "$M2_HOME" ] ; then - ## resolve links - $0 may be a link to maven's home - PRG="$0" - - # need this for relative symlinks - while [ -h "$PRG" ] ; do - ls=`ls -ld "$PRG"` - link=`expr "$ls" : '.*-> \(.*\)$'` - if expr "$link" : '/.*' > /dev/null; then - PRG="$link" - else - PRG="`dirname "$PRG"`/$link" - fi - done - - saveddir=`pwd` - - M2_HOME=`dirname "$PRG"`/.. - - # make it fully qualified - M2_HOME=`cd "$M2_HOME" && pwd` - - cd "$saveddir" - # echo Using m2 at $M2_HOME -fi - -# For Cygwin, ensure paths are in UNIX format before anything is touched -if $cygwin ; then - [ -n "$M2_HOME" ] && - M2_HOME=`cygpath --unix "$M2_HOME"` - [ -n "$JAVA_HOME" ] && - JAVA_HOME=`cygpath --unix "$JAVA_HOME"` - [ -n "$CLASSPATH" ] && - CLASSPATH=`cygpath --path --unix "$CLASSPATH"` -fi - -# For Migwn, ensure paths are in UNIX format before anything is touched -if $mingw ; then - [ -n "$M2_HOME" ] && - M2_HOME="`(cd "$M2_HOME"; pwd)`" - [ -n "$JAVA_HOME" ] && - JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`" - # TODO classpath? -fi - -if [ -z "$JAVA_HOME" ]; then - javaExecutable="`which javac`" - if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then - # readlink(1) is not available as standard on Solaris 10. - readLink=`which readlink` - if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then - if $darwin ; then - javaHome="`dirname \"$javaExecutable\"`" - javaExecutable="`cd \"$javaHome\" && pwd -P`/javac" - else - javaExecutable="`readlink -f \"$javaExecutable\"`" - fi - javaHome="`dirname \"$javaExecutable\"`" - javaHome=`expr "$javaHome" : '\(.*\)/bin'` - JAVA_HOME="$javaHome" - export JAVA_HOME - fi - fi -fi - -if [ -z "$JAVACMD" ] ; then - if [ -n "$JAVA_HOME" ] ; then - if [ -x "$JAVA_HOME/jre/sh/java" ] ; then - # IBM's JDK on AIX uses strange locations for the executables - JAVACMD="$JAVA_HOME/jre/sh/java" - else - JAVACMD="$JAVA_HOME/bin/java" - fi - else - JAVACMD="`which java`" - fi -fi - -if [ ! -x "$JAVACMD" ] ; then - echo "Error: JAVA_HOME is not defined correctly." >&2 - echo " We cannot execute $JAVACMD" >&2 - exit 1 -fi - -if [ -z "$JAVA_HOME" ] ; then - echo "Warning: JAVA_HOME environment variable is not set." -fi - -CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher - -# traverses directory structure from process work directory to filesystem root -# first directory with .mvn subdirectory is considered project base directory -find_maven_basedir() { - - if [ -z "$1" ] - then - echo "Path not specified to find_maven_basedir" - return 1 - fi - - basedir="$1" - wdir="$1" - while [ "$wdir" != '/' ] ; do - if [ -d "$wdir"/.mvn ] ; then - basedir=$wdir - break - fi - # workaround for JBEAP-8937 (on Solaris 10/Sparc) - if [ -d "${wdir}" ]; then - wdir=`cd "$wdir/.."; pwd` - fi - # end of workaround - done - echo "${basedir}" -} - -# concatenates all lines of a file -concat_lines() { - if [ -f "$1" ]; then - echo "$(tr -s '\n' ' ' < "$1")" - fi -} - -BASE_DIR=`find_maven_basedir "$(pwd)"` -if [ -z "$BASE_DIR" ]; then - exit 1; -fi - -export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"} -echo $MAVEN_PROJECTBASEDIR -MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS" - -# For Cygwin, switch paths to Windows format before running java -if $cygwin; then - [ -n "$M2_HOME" ] && - M2_HOME=`cygpath --path --windows "$M2_HOME"` - [ -n "$JAVA_HOME" ] && - JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"` - [ -n "$CLASSPATH" ] && - CLASSPATH=`cygpath --path --windows "$CLASSPATH"` - [ -n "$MAVEN_PROJECTBASEDIR" ] && - MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"` -fi - -WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain - -"$JAVACMD" \ - $MAVEN_OPTS \ - -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \ - "-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \ - ${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@" - diff --git a/partitioning-samples/partitioning-producer-sample/mvnw.cmd b/partitioning-samples/partitioning-producer-sample/mvnw.cmd deleted file mode 100644 index 7ecd01d..0000000 --- a/partitioning-samples/partitioning-producer-sample/mvnw.cmd +++ /dev/null @@ -1,145 +0,0 @@ -@REM ---------------------------------------------------------------------------- -@REM Licensed to the Apache Software Foundation (ASF) under one -@REM or more contributor license agreements. See the NOTICE file -@REM distributed with this work for additional information -@REM regarding copyright ownership. The ASF licenses this file -@REM to you under the Apache License, Version 2.0 (the -@REM "License"); you may not use this file except in compliance -@REM with the License. You may obtain a copy of the License at -@REM -@REM https://www.apache.org/licenses/LICENSE-2.0 -@REM -@REM Unless required by applicable law or agreed to in writing, -@REM software distributed under the License is distributed on an -@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -@REM KIND, either express or implied. See the License for the -@REM specific language governing permissions and limitations -@REM under the License. -@REM ---------------------------------------------------------------------------- - -@REM ---------------------------------------------------------------------------- -@REM Maven2 Start Up Batch script -@REM -@REM Required ENV vars: -@REM JAVA_HOME - location of a JDK home dir -@REM -@REM Optional ENV vars -@REM M2_HOME - location of maven2's installed home dir -@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands -@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a key stroke before ending -@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven -@REM e.g. to debug Maven itself, use -@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000 -@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files -@REM ---------------------------------------------------------------------------- - -@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on' -@echo off -@REM enable echoing my setting MAVEN_BATCH_ECHO to 'on' -@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO% - -@REM set %HOME% to equivalent of $HOME -if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%") - -@REM Execute a user defined script before this one -if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre -@REM check for pre script, once with legacy .bat ending and once with .cmd ending -if exist "%HOME%\mavenrc_pre.bat" call "%HOME%\mavenrc_pre.bat" -if exist "%HOME%\mavenrc_pre.cmd" call "%HOME%\mavenrc_pre.cmd" -:skipRcPre - -@setlocal - -set ERROR_CODE=0 - -@REM To isolate internal variables from possible post scripts, we use another setlocal -@setlocal - -@REM ==== START VALIDATION ==== -if not "%JAVA_HOME%" == "" goto OkJHome - -echo. -echo Error: JAVA_HOME not found in your environment. >&2 -echo Please set the JAVA_HOME variable in your environment to match the >&2 -echo location of your Java installation. >&2 -echo. -goto error - -:OkJHome -if exist "%JAVA_HOME%\bin\java.exe" goto init - -echo. -echo Error: JAVA_HOME is set to an invalid directory. >&2 -echo JAVA_HOME = "%JAVA_HOME%" >&2 -echo Please set the JAVA_HOME variable in your environment to match the >&2 -echo location of your Java installation. >&2 -echo. -goto error - -@REM ==== END VALIDATION ==== - -:init - -set MAVEN_CMD_LINE_ARGS=%* - -@REM Find the project base dir, i.e. the directory that contains the folder ".mvn". -@REM Fallback to current working directory if not found. - -set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR% -IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir - -set EXEC_DIR=%CD% -set WDIR=%EXEC_DIR% -:findBaseDir -IF EXIST "%WDIR%"\.mvn goto baseDirFound -cd .. -IF "%WDIR%"=="%CD%" goto baseDirNotFound -set WDIR=%CD% -goto findBaseDir - -:baseDirFound -set MAVEN_PROJECTBASEDIR=%WDIR% -cd "%EXEC_DIR%" -goto endDetectBaseDir - -:baseDirNotFound -set MAVEN_PROJECTBASEDIR=%EXEC_DIR% -cd "%EXEC_DIR%" - -:endDetectBaseDir - -IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig - -@setlocal EnableExtensions EnableDelayedExpansion -for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a -@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS% - -:endReadAdditionalConfig - -SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe" - -set WRAPPER_JAR="".\.mvn\wrapper\maven-wrapper.jar"" -set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain - -%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CMD_LINE_ARGS% -if ERRORLEVEL 1 goto error -goto end - -:error -set ERROR_CODE=1 - -:end -@endlocal & set ERROR_CODE=%ERROR_CODE% - -if not "%MAVEN_SKIP_RC%" == "" goto skipRcPost -@REM check for post script, once with legacy .bat ending and once with .cmd ending -if exist "%HOME%\mavenrc_post.bat" call "%HOME%\mavenrc_post.bat" -if exist "%HOME%\mavenrc_post.cmd" call "%HOME%\mavenrc_post.cmd" -:skipRcPost - -@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on' -if "%MAVEN_BATCH_PAUSE%" == "on" pause - -if "%MAVEN_TERMINATE_CMD%" == "on" exit %ERROR_CODE% - -exit /B %ERROR_CODE% diff --git a/partitioning-samples/partitioning-producer-sample/pom.xml b/partitioning-samples/partitioning-producer-sample/pom.xml deleted file mode 100644 index 7cf5d51..0000000 --- a/partitioning-samples/partitioning-producer-sample/pom.xml +++ /dev/null @@ -1,85 +0,0 @@ - - - 4.0.0 - - partitioning-producer-sample - 0.0.1-SNAPSHOT - jar - partitioning-producer-sample - Spring Cloud Stream Partitioning Kafka - - - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. - - - - - org.springframework.boot - spring-boot-starter-test - test - - - org.springframework.cloud - spring-cloud-stream-test-support - test - - - - - - kafka-binder - - true - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - - - - - - org.springframework.boot - spring-boot-maven-plugin - - kafka - - - - - - - rabbit-binder - - - org.springframework.cloud - spring-cloud-stream-binder-rabbit - - - - - - org.springframework.boot - spring-boot-maven-plugin - - rabbit - - - - - - - - - - - org.springframework.boot - spring-boot-maven-plugin - - - - diff --git a/partitioning-samples/partitioning-producer-sample/src/main/docker/assembly-kafka.xml b/partitioning-samples/partitioning-producer-sample/src/main/docker/assembly-kafka.xml deleted file mode 100644 index 18d8320..0000000 --- a/partitioning-samples/partitioning-producer-sample/src/main/docker/assembly-kafka.xml +++ /dev/null @@ -1,16 +0,0 @@ - - uppercase-transformer-kafka - - - - io.spring.cloud.stream.sample:partitioning-producer-sample - - . - partitioning-producer-sample-kafka.jar - - - - diff --git a/partitioning-samples/partitioning-producer-sample/src/main/docker/assembly-rabbit.xml b/partitioning-samples/partitioning-producer-sample/src/main/docker/assembly-rabbit.xml deleted file mode 100644 index 8f3da73..0000000 --- a/partitioning-samples/partitioning-producer-sample/src/main/docker/assembly-rabbit.xml +++ /dev/null @@ -1,16 +0,0 @@ - - uppercase-transformer-kafka - - - - io.spring.cloud.stream.sample:partitioning-producer-sample - - . - partitioning-producer-sample-rabbit.jar - - - - diff --git a/partitioning-samples/partitioning-producer-sample/src/main/java/demo/producer/PartProducerApplication.java b/partitioning-samples/partitioning-producer-sample/src/main/java/demo/producer/PartProducerApplication.java deleted file mode 100644 index bb0b9d9..0000000 --- a/partitioning-samples/partitioning-producer-sample/src/main/java/demo/producer/PartProducerApplication.java +++ /dev/null @@ -1,16 +0,0 @@ -package demo.producer; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; - -/** - * @author Soby Chacko - */ -@SpringBootApplication -public class PartProducerApplication { - - public static void main(String[] args) { - SpringApplication.run(PartProducerApplication.class, args); - } - -} diff --git a/partitioning-samples/partitioning-producer-sample/src/main/java/demo/producer/Producer.java b/partitioning-samples/partitioning-producer-sample/src/main/java/demo/producer/Producer.java deleted file mode 100644 index a2b7afb..0000000 --- a/partitioning-samples/partitioning-producer-sample/src/main/java/demo/producer/Producer.java +++ /dev/null @@ -1,44 +0,0 @@ -package demo.producer; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.integration.annotation.InboundChannelAdapter; -import org.springframework.integration.annotation.Poller; -import org.springframework.messaging.Message; -import org.springframework.messaging.support.MessageBuilder; - -import java.util.Random; - -/** - * @author Soby Chacko - */ -public class Producer { - - private static final Logger logger = LoggerFactory.getLogger(Producer.class); - - @EnableBinding(Source.class) - static class KafkaPartitionProducerApplication { - - private static final Random RANDOM = new Random(System.currentTimeMillis()); - - // We use a strategy so that this data will end up in a partition, - // P = L(x) - 1 where L is a length function on the payload. - private static final String[] data = new String[]{ - "f", "g", "h", //making them go to partition-0 by making a single char string - "fo", "go", "ho", - "foo", "goo", "hoo", - "fooz", "gooz", "hooz" - }; - - @InboundChannelAdapter(channel = Source.OUTPUT, poller = @Poller(fixedRate = "1000")) - public Message generate() { - String value = data[RANDOM.nextInt(data.length)]; - logger.info("Sending: " + value); - return MessageBuilder.withPayload(value) - .setHeader("partitionKey", value.length()) - .build(); - } - } -} diff --git a/partitioning-samples/pom.xml b/partitioning-samples/pom.xml index c307020..3e15c9a 100644 --- a/partitioning-samples/pom.xml +++ b/partitioning-samples/pom.xml @@ -9,9 +9,8 @@ Collection of Spring Cloud Stream partitioning Samples - partitioning-producer-sample - partitioning-consumer-sample-kafka - partitioning-consumer-sample-rabbit + kafka-partitioning + rabbit-partitioning diff --git a/partitioning-samples/rabbit-partitioning/.mvn/jvm.config b/partitioning-samples/rabbit-partitioning/.mvn/jvm.config new file mode 100644 index 0000000..0e7dabe --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/.mvn/jvm.config @@ -0,0 +1 @@ +-Xmx1024m -XX:CICompilerCount=1 -XX:TieredStopAtLevel=1 -Djava.security.egd=file:/dev/./urandom \ No newline at end of file diff --git a/partitioning-samples/rabbit-partitioning/.mvn/maven.config b/partitioning-samples/rabbit-partitioning/.mvn/maven.config new file mode 100644 index 0000000..3b8cf46 --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/.mvn/maven.config @@ -0,0 +1 @@ +-DaltSnapshotDeploymentRepository=repo.spring.io::default::https://repo.spring.io/libs-snapshot-local -P spring diff --git a/partitioning-samples/rabbit-partitioning/.mvn/wrapper/maven-wrapper.jar b/partitioning-samples/rabbit-partitioning/.mvn/wrapper/maven-wrapper.jar new file mode 100644 index 0000000..5fd4d50 Binary files /dev/null and b/partitioning-samples/rabbit-partitioning/.mvn/wrapper/maven-wrapper.jar differ diff --git a/partitioning-samples/rabbit-partitioning/.mvn/wrapper/maven-wrapper.properties b/partitioning-samples/rabbit-partitioning/.mvn/wrapper/maven-wrapper.properties new file mode 100644 index 0000000..eb91947 --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/.mvn/wrapper/maven-wrapper.properties @@ -0,0 +1 @@ +distributionUrl=https://repo1.maven.org/maven2/org/apache/maven/apache-maven/3.3.3/apache-maven-3.3.3-bin.zip \ No newline at end of file diff --git a/partitioning-samples/rabbit-partitioning/README.adoc b/partitioning-samples/rabbit-partitioning/README.adoc new file mode 100644 index 0000000..2c3403a --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/README.adoc @@ -0,0 +1,51 @@ +Spring Cloud Stream Partitioning Sample +======================================== + +These sample apps demonstrate how partitioning works in Spring Cloud Stream with RabbitMQ. +Sample producer and consumer are provided. + +## Quick introduction + +The producer used in the sample produces messages with text that has a length of 1, 2, 3 or 4. +There is a configuration in the producer's application.yml file for `partition-key-expression` that uses the length of the payload minus 1 as the partition key expression to use. +This value will be used by the binder for selecting the correct partition based on the total number of partitions configured on the destination at the broker. +We use 4 partitions for this demo. + +## Running the samples + +The following instructions assume that you are running Rabbit as a Docker image. + +* `docker-compose up -d` + +* Ensure that you are in the directory `rabbit-partitioning` + +* `./mvnw clean package` + +Rabbit partitioning demo is slightly different from Kafka. +We need to spin up 4 consumers for each of the four partitions. + +* `java -jar partitioning-consumer-sample-rabbit/target/partitioning-consumer-sample-rabbit-0.0.1-SNAPSHOT.jar --server.port=9005` + +On another terminal start another instance of the consumer. + +* `java -jar partitioning-consumer-sample-rabbit/target/partitioning-consumer-sample-rabbit-0.0.1-SNAPSHOT.jar --server.port=9006 --spring.cloud.stream.bindings.listen-in-0.consumer.instanceIndex=1` + +On another terminal start another instance of the consumer. + +* `java -jar partitioning-consumer-sample-rabbit/target/partitioning-consumer-sample-rabbit-0.0.1-SNAPSHOT.jar --server.port=9007 --spring.cloud.stream.bindings.listen-in-0.consumer.instanceIndex=2` + +On another terminal start yet another instance of the consumer. + +* `java -jar partitioning-consumer-sample-rabbit/target/partitioning-consumer-sample-rabbit-0.0.1-SNAPSHOT.jar --server.port=9008 --spring.cloud.stream.bindings.listen-in-0.consumer.instanceIndex=3` + +On another terminal start the producer. + +* `java -jar partitioning-producer-sample-rabbit/target/partitioning-producer-sample-rabbit-0.0.1-SNAPSHOT.jar --server.port=9010` + +Producer sends messages randomly that has string length of 1, 2, 3, or 4. +Watch the consumer console logs and verify that the correct instances are receiving the messages. +The first consumer we started should receive messages with a string length of 1 (partition-0), second consumer with `instanceIndex` set to 1 should receive messages with string length of 2 (partition-1) so on and so forth. + +Once you are done testing, stop all the instances. + +* `docker-compose down` \ No newline at end of file diff --git a/partitioning-samples/docker-compose-rabbit.yml b/partitioning-samples/rabbit-partitioning/docker-compose.yml similarity index 100% rename from partitioning-samples/docker-compose-rabbit.yml rename to partitioning-samples/rabbit-partitioning/docker-compose.yml diff --git a/partitioning-samples/partitioning-consumer-sample-kafka/mvnw b/partitioning-samples/rabbit-partitioning/mvnw similarity index 100% rename from partitioning-samples/partitioning-consumer-sample-kafka/mvnw rename to partitioning-samples/rabbit-partitioning/mvnw diff --git a/partitioning-samples/partitioning-consumer-sample-kafka/mvnw.cmd b/partitioning-samples/rabbit-partitioning/mvnw.cmd similarity index 100% rename from partitioning-samples/partitioning-consumer-sample-kafka/mvnw.cmd rename to partitioning-samples/rabbit-partitioning/mvnw.cmd diff --git a/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/pom.xml b/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/pom.xml new file mode 100644 index 0000000..fab12a5 --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/pom.xml @@ -0,0 +1,117 @@ + + + 4.0.0 + + partitioning-consumer-sample-rabbit + 0.0.1-SNAPSHOT + jar + partitioning-consumer-sample-rabbit + Spring Cloud Stream Partitioning Rabbit + + + 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-rabbit + + + org.springframework.boot + spring-boot-starter-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/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemoApplication.java b/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemoApplication.java similarity index 63% rename from partitioning-samples/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemoApplication.java rename to partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemoApplication.java index ea4119b..c202dfd 100644 --- a/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemoApplication.java +++ b/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/main/java/demo/PartitioningRabbitDemoApplication.java @@ -16,14 +16,29 @@ package demo; +import java.util.function.Consumer; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.support.AmqpHeaders; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.Message; @SpringBootApplication public class PartitioningRabbitDemoApplication { + private static final Logger logger = LoggerFactory.getLogger(PartitioningRabbitDemoApplication.class); + public static void main(String[] args) { SpringApplication.run(PartitioningRabbitDemoApplication.class, args); } + @Bean + public Consumer> listen() { + return message -> logger.info(message.getPayload() + " received from partition " + + message.getHeaders().get(AmqpHeaders.CONSUMER_QUEUE)); + } + } diff --git a/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/resources/application.yml b/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/main/resources/application.yml similarity index 90% rename from partitioning-samples/partitioning-consumer-sample-rabbit/src/main/resources/application.yml rename to partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/main/resources/application.yml index f63a754..7641812 100644 --- a/partitioning-samples/partitioning-consumer-sample-rabbit/src/main/resources/application.yml +++ b/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/main/resources/application.yml @@ -2,7 +2,7 @@ spring: cloud: stream: bindings: - input: + listen-in-0: destination: partitioned.destination group: myGroup consumer: diff --git a/partitioning-samples/partitioning-consumer-sample-rabbit/src/test/java/demo/ModuleApplicationTests.java b/partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/test/java/demo/ModuleApplicationTests.java similarity index 100% rename from partitioning-samples/partitioning-consumer-sample-rabbit/src/test/java/demo/ModuleApplicationTests.java rename to partitioning-samples/rabbit-partitioning/partitioning-consumer-sample-rabbit/src/test/java/demo/ModuleApplicationTests.java diff --git a/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/pom.xml b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/pom.xml new file mode 100644 index 0000000..3c656ea --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/pom.xml @@ -0,0 +1,117 @@ + + + 4.0.0 + + partitioning-producer-sample-rabbit + 0.0.1-SNAPSHOT + jar + partitioning-producer-sample-rabbit + Spring Cloud Stream Partitioning producer rabbit + + + 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-rabbit + + + org.springframework.boot + spring-boot-starter-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/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/main/java/demo/producer/PartitioningProducerApplication.java b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/main/java/demo/producer/PartitioningProducerApplication.java new file mode 100644 index 0000000..32ff448 --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/main/java/demo/producer/PartitioningProducerApplication.java @@ -0,0 +1,48 @@ +package demo.producer; + +import java.util.Random; +import java.util.function.Supplier; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Soby Chacko + */ +@SpringBootApplication +public class PartitioningProducerApplication { + + private static final Random RANDOM = new Random(System.currentTimeMillis()); + + private static final Logger logger = LoggerFactory.getLogger(PartitioningProducerApplication.class); + + + public static void main(String[] args) { + SpringApplication.run(PartitioningProducerApplication.class, args); + } + + // We use a strategy so that this data will end up in a partition, + // P = L(x) - 1 where L is a length function on the payload. + private static final String[] data = new String[]{ + "f", "g", "h", //making them go to partition-0 by making a single char string + "fo", "go", "ho", + "foo", "goo", "hoo", + "fooz", "gooz", "hooz" + }; + + @Bean + public Supplier> generate() { + return () -> { + String value = data[RANDOM.nextInt(data.length)]; + logger.info("Sending: " + value); + return MessageBuilder.withPayload(value) + .setHeader("partitionKey", value.length()) + .build(); + }; + } +} diff --git a/partitioning-samples/partitioning-producer-sample/src/main/resources/application.yml b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/main/resources/application.yml similarity index 93% rename from partitioning-samples/partitioning-producer-sample/src/main/resources/application.yml rename to partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/main/resources/application.yml index 0284449..5f59691 100644 --- a/partitioning-samples/partitioning-producer-sample/src/main/resources/application.yml +++ b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/main/resources/application.yml @@ -2,10 +2,10 @@ spring: cloud: stream: bindings: - output: + generate-out-0: destination: partitioned.destination producer: #payload string length - 1 is the partition where it will get stored partition-key-expression: headers['partitionKey'] - 1 partition-count: 4 - required-groups: myGroup #only applicable for rabbit + required-groups: myGroup #only applicable for rabbit \ No newline at end of file diff --git a/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/test/java/demo/producer/ModuleApplicationTests.java b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/test/java/demo/producer/ModuleApplicationTests.java new file mode 100644 index 0000000..1498cd6 --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/partitioning-producer-sample-rabbit/src/test/java/demo/producer/ModuleApplicationTests.java @@ -0,0 +1,32 @@ +/* + * 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 demo.producer; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.junit4.SpringRunner; + +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE) +public class ModuleApplicationTests { + + @Test + public void contextLoads() { + } + +} diff --git a/partitioning-samples/rabbit-partitioning/pom.xml b/partitioning-samples/rabbit-partitioning/pom.xml new file mode 100644 index 0000000..b43c246 --- /dev/null +++ b/partitioning-samples/rabbit-partitioning/pom.xml @@ -0,0 +1,16 @@ + + + 4.0.0 + io.spring.cloud.stream.sample + rabbit-partitioning + 0.0.1-SNAPSHOT + pom + rabbit-partitioning + Collection of Spring Cloud Stream partitioning Samples using rabbit binder + + + partitioning-producer-sample-rabbit + partitioning-consumer-sample-rabbit + + +