mirror of
https://github.com/wahyd4/spring-cloud-stream-samples.git
synced 2026-08-09 05:16:18 +10:00
Add test-embedded-kafka Sample
Also some general README and pom cleanup. README Polishing Resolves #37 Polishing - Use Boot Config for Test Consumer Possible now that https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/pull/137 is merged.
This commit is contained in:
Gary Russell
committed by
Vinicius Carvalho
Vinicius Carvalho
parent
6fec783413
commit
b29fc5491a
+22
-8
@@ -10,17 +10,31 @@ To build the samples do:
|
||||
|
||||
NOTE: The main set of samples are "vanilla" in the sense that they are not deployable as XD modules by the current generation (1.x) of XD. You can still interact with an XD system using the appropriate naming convention for input and output channel names (`<stream>.<index>` format).
|
||||
|
||||
* `source` is a Java config version of the classic "timer" module from Spring XD. It has a "fixedDelay" option (in milliseconds) for the period between emitting messages.
|
||||
|
||||
* `sink` is a Java config version of the classic "log" module from Spring XD. It has no options (but some could easily be added), and just logs incoming messages at INFO level.
|
||||
|
||||
* `transform` is a simple pass through logging transformer (just logs the incoming message and passes it on).
|
||||
|
||||
* `double` is a combination of 2 modules defined locally (a source and a sink, so the whole app is self contained).
|
||||
|
||||
* `multibinder` is a sample application that shows how an application could use multiple binders. In this case, the processor's input/output channels connect to different brokers using their own binder configurations.
|
||||
* `dynamic-source` publishes messages to dynamically created destinations.
|
||||
|
||||
* `multi-io` shows how to use configure multiple input/output channels inside a single application.
|
||||
|
||||
* `multibinder-differentsystems` shows how an application could use same binder implementation but different configurations for its channels. In this case, a processor's input/output channels connect to same binder implementation but with two separate broker configurations.
|
||||
|
||||
If you run the source and the sink and point them at the same redis instance (e.g. do nothing to get the one on localhost, or the one they are both bound to as a service on Cloud Foundry) then they will form a "stream" and start talking to each other. All the samples have friendly JMX and Actuator endpoints for inspecting what is going on in the system.
|
||||
* `multibinder` shows how an application could use multiple binders. In this case, the processor's input/output channels connect to different brokers using their own binder configurations.
|
||||
|
||||
* `non-self-contained-aggregate-app` shows how to write a non self-contained aggregate application.
|
||||
|
||||
* `reactive-processor-kafka` shows how to create a reactive Apache Kafka processor application.
|
||||
|
||||
* `rxjava-processor` shows how to create an RxJava processor application.
|
||||
|
||||
* `sink` is a Java config version of the classic "log" module from Spring XD. It has no options (but some could easily be added), and just logs incoming messages at INFO level.
|
||||
|
||||
* `source` is a Java config version of the classic "time" module from Spring XD. It has a "fixedDelay" option (in milliseconds) for the period between emitting messages.
|
||||
|
||||
* `stream-listener` shows how to use StreamListener support to enable message mapping and automatic type conversion.
|
||||
|
||||
* test-embedded-kafka is a sample that shows how to test with an embedded Apache Kafka broker.
|
||||
We generally recommend testing with the http://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_testing[TestSupportBinder] but if you have a need for testing with an embedded broker, you can use the techniques in this sample.
|
||||
|
||||
* `transform` is a simple pass through logging transformer (just logs the incoming message and passes it on).
|
||||
|
||||
If you run the source and the sink and point them at the same redis instance (e.g. do nothing to get the one on localhost, or the one they are both bound to as a service on Cloud Foundry) then they will form a "stream" and start talking to each other. All the samples have friendly JMX and Actuator endpoints for inspecting what is going on in the system.
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-samples</artifactId>
|
||||
<version>1.1.0.BUILD-SNAPSHOT</version>
|
||||
<version>1.2.0.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<properties>
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-samples</artifactId>
|
||||
<version>1.1.0.BUILD-SNAPSHOT</version>
|
||||
<version>1.2.0.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<properties>
|
||||
@@ -28,7 +28,7 @@
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
|
||||
@@ -33,6 +33,7 @@
|
||||
<module>multi-io</module>
|
||||
<module>stream-listener</module>
|
||||
<module>reactive-processor-kafka</module>
|
||||
<module>test-embedded-kafka</module>
|
||||
</modules>
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
|
||||
@@ -9,7 +9,7 @@
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>reactive-processor-kafka</artifactId>
|
||||
<artifactId>spring-cloud-stream-sample-reactive-processor-kafka</artifactId>
|
||||
|
||||
<properties>
|
||||
<java.version>1.8</java.version>
|
||||
@@ -59,4 +59,4 @@
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
</project>
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
target/
|
||||
!.mvn/wrapper/maven-wrapper.jar
|
||||
|
||||
### STS ###
|
||||
.apt_generated
|
||||
.classpath
|
||||
.factorypath
|
||||
.project
|
||||
.settings
|
||||
.springBeans
|
||||
|
||||
### IntelliJ IDEA ###
|
||||
.idea
|
||||
*.iws
|
||||
*.iml
|
||||
*.ipr
|
||||
|
||||
### NetBeans ###
|
||||
nbproject/private/
|
||||
build/
|
||||
nbbuild/
|
||||
dist/
|
||||
nbdist/
|
||||
.nb-gradle/
|
||||
@@ -0,0 +1,37 @@
|
||||
Spring Cloud Stream Testing with Embedded Kafka Broker Sample
|
||||
=============================
|
||||
|
||||
In this *Spring Cloud Stream* sample, an embedded kafka broker is used for testing.
|
||||
|
||||
## Requirements
|
||||
|
||||
To run this sample, you will need to have installed:
|
||||
|
||||
* Java 8 or Above
|
||||
|
||||
## Code Tour
|
||||
|
||||
This sample is a Spring Boot application that uses Spring Cloud Stream to receive a `String` message and forward an upper case version of that String; it uses the Kafka binder.
|
||||
|
||||
* EmbeddedKafkaApplication - the Spring Boot Main Application
|
||||
* EmbeddedKafkaApplicationTests - the test case
|
||||
|
||||
The `spring-kafka-test` dependency added to the `pom.xml` puts the `KafkaEmbedded` JUnit `@Rule` on the class path.
|
||||
Refer to the [Spring for Apache Kafka Reference Manual](http://docs.spring.io/spring-kafka/reference/htmlsingle/#testing) for more information about this.
|
||||
Notice how the `@BeforeClass` method sets up the Boot and binder properties to locate the servers.
|
||||
|
||||
|
||||
The test method starts by using the Spring Boot auto-configured `KafkaTemplate` to send a message to the input destination.
|
||||
It then creates a consumer to consume from the output destination; gets the output message and asserts that it's an upper case version of the sent message.
|
||||
|
||||
See the `application.properties` file for the properties that are required for this test case and application.
|
||||
In particular, `raw` header mode is used.
|
||||
|
||||
## Building with Maven
|
||||
|
||||
Build the sample and run the test by executing:
|
||||
|
||||
source>$ mvn clean package
|
||||
|
||||
or run the test in your favorite IDE.
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>spring-cloud-stream-sample-test-embedded-kafka</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<name>spring-cloud-stream-sample-test-embedded-kafka</name>
|
||||
<description>Demo project for Testing with an embedded Kafka broker</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-samples</artifactId>
|
||||
<version>1.2.0.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<properties>
|
||||
<start-class>demo.EmbeddedKafkaApplication</start-class>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-kafka</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>
|
||||
<configuration>
|
||||
<classifier>exec</classifier>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,44 @@
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* http://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.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.annotation.StreamListener;
|
||||
import org.springframework.cloud.stream.messaging.Processor;
|
||||
import org.springframework.messaging.handler.annotation.SendTo;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
*
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableBinding(Processor.class)
|
||||
public class EmbeddedKafkaApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(EmbeddedKafkaApplication.class, args);
|
||||
}
|
||||
|
||||
@StreamListener(Processor.INPUT)
|
||||
@SendTo(Processor.OUTPUT)
|
||||
public byte[] handle(byte[] in){
|
||||
return new String(in).toUpperCase().getBytes();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
# Test consumer properties
|
||||
spring.kafka.consumer.auto-offset-reset=earliest
|
||||
spring.kafka.consumer.group-id=testEmbeddedKafkaApplication
|
||||
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
|
||||
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
|
||||
|
||||
# Binding properties
|
||||
spring.cloud.stream.bindings.output.destination=testEmbeddedOut
|
||||
spring.cloud.stream.bindings.input.destination=testEmbeddedIn
|
||||
spring.cloud.stream.bindings.output.producer.headerMode=raw
|
||||
spring.cloud.stream.bindings.input.consumer.headerMode=raw
|
||||
spring.cloud.stream.bindings.input.group=embeddedKafkaApplication
|
||||
@@ -0,0 +1,81 @@
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* http://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 static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import java.util.Collections;
|
||||
|
||||
import org.apache.kafka.clients.consumer.Consumer;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
|
||||
import org.springframework.kafka.core.KafkaTemplate;
|
||||
import org.springframework.kafka.test.rule.KafkaEmbedded;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
/**
|
||||
* Test class demonstrating how to use an embedded kafka service with the
|
||||
* kafka binder.
|
||||
*
|
||||
* @author Gary Russell
|
||||
*
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest
|
||||
public class EmbeddedKafkaApplicationTests {
|
||||
|
||||
@ClassRule
|
||||
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1);
|
||||
|
||||
@Autowired
|
||||
private KafkaTemplate<byte[], byte[]> template;
|
||||
|
||||
@Autowired
|
||||
private DefaultKafkaConsumerFactory<byte[], byte[]> consumerFactory;
|
||||
|
||||
@Value("${spring.cloud.stream.bindings.input.destination}")
|
||||
private String inputDestination;
|
||||
|
||||
@Value("${spring.cloud.stream.bindings.output.destination}")
|
||||
private String outputDestination;
|
||||
|
||||
@BeforeClass
|
||||
public static void setup() {
|
||||
System.setProperty("spring.kafka.bootstrap-servers", embeddedKafka.getBrokersAsString());
|
||||
System.setProperty("spring.cloud.stream.kafka.binder.zkNodes", embeddedKafka.getZookeeperConnectionString());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendReceive() {
|
||||
template.send(this.inputDestination, "foo".getBytes());
|
||||
Consumer<byte[], byte[]> consumer = this.consumerFactory.createConsumer();
|
||||
consumer.subscribe(Collections.singleton(this.outputDestination));
|
||||
ConsumerRecords<byte[], byte[]> records = consumer.poll(10_000);
|
||||
consumer.commitSync();
|
||||
assertThat(records.count()).isEqualTo(1);
|
||||
assertThat(new String(records.iterator().next().value())).isEqualTo("FOO");
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user