Add Java DSL sample for Kafka 0.9

Make `PromiseTest` more fast decreasing `sleep()` in the `MathService` Kafka-0.10.0 Compatibility Upgrade to Spring Boot 1.4.0.RELEASE Regenerate poms according to upgrades
2016-07-11 16:07:48 -04:00
parent b6cbf15e4c
commit a46bd8acb3
71 changed files with 584 additions and 227 deletions
--- a/dsl/cafe-dsl/pom.xml
+++ b/dsl/cafe-dsl/pom.xml
@@ -4,7 +4,7 @@
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
-    <version>1.4.0.RC1</version>
+    <version>1.4.0.RELEASE</version>
  </parent>
  <groupId>org.springframework.integration.samples</groupId>
  <artifactId>cafe-dsl</artifactId>
@@ -71,7 +71,7 @@
    <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-core</artifactId>
-      <version>4.3.0.RELEASE</version>
+      <version>4.3.1.RELEASE</version>
      <scope>compile</scope>
    </dependency>
    <dependency>
@@ -83,19 +83,7 @@
    <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-java-dsl</artifactId>
-      <version>1.1.2.RELEASE</version>
-      <scope>compile</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.projectreactor</groupId>
-      <artifactId>reactor-stream</artifactId>
-      <version>2.0.7.RELEASE</version>
-      <scope>compile</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.projectreactor.spring</groupId>
-      <artifactId>reactor-spring-context</artifactId>
-      <version>2.0.7.RELEASE</version>
+      <version>1.2.0.M1</version>
      <scope>compile</scope>
    </dependency>
    <dependency>
@@ -119,7 +107,7 @@
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-test</artifactId>
-      <version>4.3.0.RELEASE</version>
+      <version>4.3.2.RELEASE</version>
      <scope>test</scope>
    </dependency>
    <dependency>
--- a/dsl/cafe-dsl/src/main/java/org/springframework/integration/samples/dsl/cafe/nonlambda/Application.java
+++ b/dsl/cafe-dsl/src/main/java/org/springframework/integration/samples/dsl/cafe/nonlambda/Application.java
@@ -104,12 +104,12 @@ public class Application {
 				.split("payload.items")
 				.channel(MessageChannels.executor(Executors.newCachedThreadPool()))
 				.route("payload.iced",
-						new Consumer<RouterSpec<ExpressionEvaluatingRouter>>() {
+						new Consumer<RouterSpec<Object, ExpressionEvaluatingRouter>>() {

 							@Override
-							public void accept(RouterSpec<ExpressionEvaluatingRouter> spec) {
-								spec.channelMapping("true", "iced")
-										.channelMapping("false", "hot");
+							public void accept(RouterSpec<Object, ExpressionEvaluatingRouter> spec) {
+								spec.channelMapping(true, "iced")
+										.channelMapping(false, "hot");
 							}

 						})
--- a/dsl/kafka-dsl/README.md
+++ b/dsl/kafka-dsl/README.md
@@ -0,0 +1,36 @@
+Spring Integration Java DSL and Apache Kafka Sample
+==============
+
+This example demonstrates the use of `Kafka09` Namespace Factory from Spring Integration Java DSL.
+
+## Running the sample
+
+Start Apache Zookeeper and Apache Kafka according to the documentation for the Apache Kafka project.
+
+    $ gradlew :kafka:run
+
+This will package the application and run it using the [Gradle Application Plugin](http://www.gradle.org/docs/current/userguide/application_plugin.html)
+
+#### Using an IDE such as SpringSource Tool Suite™ (STS)
+
+In STS (Eclipse), go to package **org.springframework.integration.samples.kafka**, right-click **Application** and select **Run as** --> **Java Application** (or Spring Boot Application).
+
+### Output
+
+The application sends 10 messages (`foo0` ... `foo9`) to a kafka topic `si.topic` (which is created if necessary).
+
+The message-driven adapter receives the messages and places them in a `QueueChannel` which the application reads and
+writes to stdout:
+
+	GenericMessage [payload=foo0, headers={kafka_offset=21, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=22}]
+	GenericMessage [payload=foo1, headers={kafka_offset=22, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=23}]
+	GenericMessage [payload=foo2, headers={kafka_offset=23, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=24}]
+	GenericMessage [payload=foo3, headers={kafka_offset=24, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=25}]
+	GenericMessage [payload=foo4, headers={kafka_offset=25, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=26}]
+	GenericMessage [payload=foo5, headers={kafka_offset=26, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=27}]
+	GenericMessage [payload=foo6, headers={kafka_offset=27, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=28}]
+	GenericMessage [payload=foo7, headers={kafka_offset=28, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=29}]
+	GenericMessage [payload=foo8, headers={kafka_offset=29, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=30}]
+	GenericMessage [payload=foo9, headers={kafka_offset=30, kafka_messageKey=si.key, kafka_topic=si.topic, kafka_partitionId=0, kafka_nextOffset=31}]
+
+Notice that the offset header increases on each run (the topic is not removed, to demonstrate that the offset is retained between executions).
--- a/dsl/kafka-dsl/pom.xml
+++ b/dsl/kafka-dsl/pom.xml
@@ -0,0 +1,140 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>1.4.0.RELEASE</version>
+  </parent>
+  <groupId>org.springframework.integration.samples</groupId>
+  <artifactId>kafka-dsl</artifactId>
+  <version>4.3.0.BUILD-SNAPSHOT</version>
+  <name>Java DSL Kafka Sample</name>
+  <description>Java DSL Kafka Sample</description>
+  <url>http://projects.spring.io/spring-integration</url>
+  <organization>
+    <name>SpringIO</name>
+    <url>https://spring.io</url>
+  </organization>
+  <licenses>
+    <license>
+      <name>The Apache Software License, Version 2.0</name>
+      <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+      <distribution>repo</distribution>
+    </license>
+  </licenses>
+  <developers>
+    <developer>
+      <id>garyrussell</id>
+      <name>Gary Russell</name>
+      <email>grussell@pivotal.io</email>
+      <roles>
+        <role>project lead</role>
+      </roles>
+    </developer>
+    <developer>
+      <id>markfisher</id>
+      <name>Mark Fisher</name>
+      <email>mfisher@pivotal.io</email>
+      <roles>
+        <role>project founder and lead emeritus</role>
+      </roles>
+    </developer>
+    <developer>
+      <id>ghillert</id>
+      <name>Gunnar Hillert</name>
+      <email>ghillert@pivotal.io</email>
+    </developer>
+    <developer>
+      <id>abilan</id>
+      <name>Artem Bilan</name>
+      <email>abilan@pivotal.io</email>
+    </developer>
+  </developers>
+  <scm>
+    <connection>scm:git:scm:git:git://github.com/spring-projects/spring-integration-samples.git</connection>
+    <developerConnection>scm:git:scm:git:ssh://git@github.com:spring-projects/spring-integration-samples.git</developerConnection>
+    <url>https://github.com/spring-projects/spring-integration-samples</url>
+  </scm>
+  <dependencies>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-integration</artifactId>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.integration</groupId>
+      <artifactId>spring-integration-core</artifactId>
+      <version>4.3.1.RELEASE</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.integration</groupId>
+      <artifactId>spring-integration-java-dsl</artifactId>
+      <version>1.2.0.M1</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka</artifactId>
+      <version>1.1.0.M1</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka-test</artifactId>
+      <version>1.1.0.M1</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>log4j</groupId>
+      <artifactId>log4j</artifactId>
+      <version>1.2.17</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.12</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-all</artifactId>
+      <version>1.3</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>1.9.5</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework</groupId>
+      <artifactId>spring-test</artifactId>
+      <version>4.3.2.RELEASE</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+  <repositories>
+    <repository>
+      <id>repo.spring.io.milestone</id>
+      <name>Spring Framework Maven Milestone Repository</name>
+      <url>https://repo.spring.io/libs-milestone</url>
+    </repository>
+  </repositories>
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>
--- a/dsl/kafka-dsl/src/main/java/org/springframework/integration/samples/dsl/kafka/Application.java
+++ b/dsl/kafka-dsl/src/main/java/org/springframework/integration/samples/dsl/kafka/Application.java
@@ -0,0 +1,158 @@
+/*
+ * Copyright 2016 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 org.springframework.integration.samples.dsl.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import javax.annotation.PostConstruct;
+
+import org.I0Itec.zkclient.ZkClient;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.annotation.Gateway;
+import org.springframework.integration.annotation.IntegrationComponentScan;
+import org.springframework.integration.annotation.MessagingGateway;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.kafka.Kafka09;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.messaging.Message;
+
+import kafka.admin.AdminUtils;
+import kafka.common.TopicExistsException;
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils;
+
+/**
+ * @author Gary Russell
+ * @author Artem Bilan
+ * @since 4.3
+ */
+@SpringBootApplication
+@IntegrationComponentScan
+public class Application {
+
+	public static void main(String[] args) throws Exception {
+		ConfigurableApplicationContext context =
+				new SpringApplicationBuilder(Application.class)
+				.web(false)
+				.run(args);
+
+		KafkaGateway kafkaGateway = context.getBean(KafkaGateway.class);
+		for (int i = 0; i < 10; i++) {
+			String message = "foo" + i;
+			System.out.println("Send to Kafka: " + message);
+			kafkaGateway.sendToKafka(message);
+		}
+
+		Message<?> received = kafkaGateway.receiveFromKafka();
+		while (received != null) {
+			System.out.println(received);
+			received = kafkaGateway.receiveFromKafka();
+		}
+
+		context.close();
+		System.exit(0);
+	}
+
+	@Value("${kafka.topic}")
+	private String topic;
+
+	@Value("${kafka.messageKey}")
+	private String messageKey;
+
+	@Value("${kafka.broker.address}")
+	private String brokerAddress;
+
+	@Value("${kafka.zookeeper.connect}")
+	private String zookeeperConnect;
+
+	@PostConstruct
+	public void init() {
+		ZkClient zkClient = new ZkClient(this.zookeeperConnect, 6000, 6000, ZKStringSerializer$.MODULE$);
+		ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
+		try {
+			AdminUtils.createTopic(zkUtils, topic, 1, 1, new Properties(), null);
+		}
+		catch (TopicExistsException e) {
+			// no-op
+		}
+	}
+
+	@MessagingGateway
+	public interface KafkaGateway {
+
+		@Gateway(requestChannel = "toKafka.input")
+		void sendToKafka(String payload);
+
+		@Gateway(replyChannel = "fromKafka", replyTimeout = 10000)
+		Message<?> receiveFromKafka();
+
+	}
+
+
+	@Bean
+	public ProducerFactory<String, String> producerFactory() {
+		Map<String, Object> props = new HashMap<>();
+		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
+		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		return new DefaultKafkaProducerFactory<>(props);
+	}
+
+	@Bean
+	public IntegrationFlow toKafka() {
+		return f -> f
+				.handle(Kafka09.outboundChannelAdapter(producerFactory())
+						.topic(this.topic)
+						.messageKey(this.messageKey));
+	}
+
+	@Bean
+	public ConsumerFactory<String, String> consumerFactory() {
+		Map<String, Object> props = new HashMap<>();
+		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, "siTestGroup");
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
+		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+		return new DefaultKafkaConsumerFactory<>(props);
+	}
+
+	@Bean
+	public IntegrationFlow fromKafka() {
+		return IntegrationFlows
+				.from(Kafka09.messageDriverChannelAdapter(consumerFactory(), this.topic))
+				.channel(c -> c.queue("fromKafka"))
+				.get();
+	}
+
+}
--- a/dsl/kafka-dsl/src/main/resources/application.yml
+++ b/dsl/kafka-dsl/src/main/resources/application.yml
@@ -0,0 +1,7 @@
+kafka:
+  broker:
+    address: localhost:9092
+  zookeeper:
+    connect: localhost:2181
+  topic: si.topic
+  messageKey: si.key
--- a/dsl/kafka-dsl/src/main/resources/logback.xml
+++ b/dsl/kafka-dsl/src/main/resources/logback.xml
@@ -0,0 +1,14 @@
+<configuration>
+
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <!-- encoders are assigned the type
+         ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+    <encoder>
+      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+    </encoder>
+  </appender>
+
+  <root level="warn">
+    <appender-ref ref="STDOUT" />
+  </root>
+</configuration>
--- a/dsl/pom.xml
+++ b/dsl/pom.xml
@@ -72,7 +72,7 @@
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-test</artifactId>
-      <version>4.3.0.RELEASE</version>
+      <version>4.3.2.RELEASE</version>
      <scope>test</scope>
    </dependency>
  </dependencies>
--- a/dsl/si4demo/pom.xml
+++ b/dsl/si4demo/pom.xml
@@ -4,7 +4,7 @@
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
-    <version>1.4.0.RC1</version>
+    <version>1.4.0.RELEASE</version>
  </parent>
  <groupId>org.springframework.integration.samples</groupId>
  <artifactId>si4demo</artifactId>
@@ -90,7 +90,7 @@
    <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-java-dsl</artifactId>
-      <version>1.1.2.RELEASE</version>
+      <version>1.2.0.M1</version>
      <scope>compile</scope>
    </dependency>
    <dependency>
@@ -132,7 +132,7 @@
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-test</artifactId>
-      <version>4.3.0.RELEASE</version>
+      <version>4.3.2.RELEASE</version>
      <scope>test</scope>
    </dependency>
    <dependency>