SpringCloud/spring-cloud-stream-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Schema registry samples update - Iteration #3

Browse Source 
Update the samples that use Confluent Schema Registry and Confluent avro serializer/deserializer.
This commit is contained in:
Soby Chacko
2019-09-23 12:07:26 -04:00
parent a123a621ec
commit a93ac7d702
7 changed files with 290 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml
View File

@@ -9,17 +9,30 @@
<description>Schema Registry Consumer</description>
<parent>
<groupId>io.spring.cloud.stream.sample</groupId>
<artifactId>spring-cloud-stream-samples-parent</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../../..</relativePath>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<avro.version>1.8.2</avro.version>
<confluent.version>4.0.0</confluent.version>
<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
@@ -73,12 +86,66 @@
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>confluent</id>
<url>https://packages.confluent.io/maven/</url>
</repository>
</repositories>
</project>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</project>

10
schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java
View File

@@ -1,5 +1,7 @@
package sample.consumer;
import java.util.function.Consumer;
import com.example.Sensor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -8,9 +10,9 @@ 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.Sink;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
@EnableBinding(Sink.class)
public class ConfluentAvroConsumerApplication {
private final Log logger = LogFactory.getLog(getClass());
@@ -19,9 +21,9 @@ public class ConfluentAvroConsumerApplication {
SpringApplication.run(ConfluentAvroConsumerApplication.class, args);
}
@StreamListener(Sink.INPUT)
public void process(Sensor data) {
logger.info(data);
@Bean
public Consumer<Sensor> process() {
return input -> logger.info("input: " + input);
}
}

1
schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml
View File

@@ -13,4 +13,5 @@ spring:
configuration:
value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
schema.registry.url: http://localhost:8081
specific.avro.reader: true
server.port: 9999

92
schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml
View File

@@ -9,17 +9,31 @@
<description>Schema Registry Producer1</description>
<parent>
<groupId>io.spring.cloud.stream.sample</groupId>
<artifactId>spring-cloud-stream-samples-parent</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../../..</relativePath>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<avro.version>1.8.2</avro.version>
<confluent.version>4.0.0</confluent.version>
<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
@@ -45,6 +59,18 @@
<artifactId>kafka-schema-registry-client</artifactId>
<version>${confluent.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
<build>
@@ -74,9 +100,65 @@
</plugins>
</build>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>confluent</id>
<url>https://packages.confluent.io/maven/</url>
</repository>
</repositories>
</project>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</project>

53
schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/java/sample/producer1/ConfluentAvroProducer1Application.java
View File

@@ -1,40 +1,31 @@
package sample.producer1;
import com.example.Sensor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Supplier;
import com.example.Sensor;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@SpringBootApplication
@EnableBinding(Source.class)
@RestController
public class ConfluentAvroProducer1Application {
@Autowired
private Source source;
private Random random = new Random();
BlockingQueue<Sensor> unbounded = new LinkedBlockingQueue<>();
public static void main(String[] args) {
SpringApplication.run(ConfluentAvroProducer1Application.class, args);
}
@RequestMapping(value = "/messages", method = RequestMethod.POST)
public String sendMessage() {
source.output().send(MessageBuilder.withPayload(randomSensor()).build());
return "ok, have fun with v1 payload!";
}
private Sensor randomSensor() {
Sensor sensor = new Sensor();
sensor.setId(UUID.randomUUID().toString() + "-v1");
@@ -44,19 +35,17 @@ public class ConfluentAvroProducer1Application {
return sensor;
}
//Another convenience POST method for testing deterministic values
@RequestMapping(value = "/messagesX", method = RequestMethod.POST)
public String sendMessageX(@RequestParam(value="id") String id, @RequestParam(value="acceleration") float acceleartion,
@RequestParam(value="velocity") float velocity, @RequestParam(value="temperature") float temperature) {
Sensor sensor = new Sensor();
sensor.setId(id + "-v1");
sensor.setAcceleration(acceleartion);
sensor.setVelocity(velocity);
sensor.setTemperature(temperature);
source.output().send(MessageBuilder.withPayload(sensor).build());
@RequestMapping(value = "/messages", method = RequestMethod.POST)
public String sendMessage() {
unbounded.offer(randomSensor());
return "ok, have fun with v1 payload!";
}
@Bean
public Supplier<Sensor> supplier() {
return () -> unbounded.poll();
}
}

93
schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml
View File

@@ -6,20 +6,33 @@
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>schema-registry-confluent-avro-serializer-producer2</name>
<description>Schema Registry Producer2</description>
<description>Schema Registry Producer1</description>
<parent>
<groupId>io.spring.cloud.stream.sample</groupId>
<artifactId>spring-cloud-stream-samples-parent</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../../..</relativePath>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<avro.version>1.8.2</avro.version>
<confluent.version>4.0.0</confluent.version>
<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
@@ -45,6 +58,18 @@
<artifactId>kafka-schema-registry-client</artifactId>
<version>${confluent.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
<build>
@@ -74,9 +99,65 @@
</plugins>
</build>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>confluent</id>
<url>https://packages.confluent.io/maven/</url>
</repository>
</repositories>
</project>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</project>

39
schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/java/sample/producer2/ConfluentAvroProducer2Application.java
View File

@@ -2,39 +2,30 @@ package sample.producer2;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Supplier;
import com.example.Sensor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@SpringBootApplication
@EnableBinding(Source.class)
@RestController
public class ConfluentAvroProducer2Application {
@Autowired
private Source source;
private Random random = new Random();
BlockingQueue<Sensor> unbounded = new LinkedBlockingQueue<>();
public static void main(String[] args) {
SpringApplication.run(ConfluentAvroProducer2Application.class, args);
}
@RequestMapping(value = "/messages", method = RequestMethod.POST)
public String sendMessage() {
source.output().send(MessageBuilder.withPayload(randomSensor()).build());
return "ok, have fun with v2 payload!";
}
private Sensor randomSensor() {
Sensor sensor = new Sensor();
sensor.setId(UUID.randomUUID().toString() + "-v2");
@@ -46,17 +37,15 @@ public class ConfluentAvroProducer2Application {
return sensor;
}
//Another convenience POST method for testing deterministic values
@RequestMapping(value = "/messagesX", method = RequestMethod.POST)
public String sendMessageX(@RequestParam(value="id") String id, @RequestParam(value="acceleration") float acceleartion,
@RequestParam(value="velocity") float velocity, @RequestParam(value="temperature") float temperature) {
Sensor sensor = new Sensor();
sensor.setId(id + "-v2");
sensor.setAcceleration(acceleartion);
sensor.setVelocity(velocity);
sensor.setInternalTemperature(temperature);
source.output().send(MessageBuilder.withPayload(sensor).build());
@RequestMapping(value = "/messages", method = RequestMethod.POST)
public String sendMessage() {
unbounded.offer(randomSensor());
return "ok, have fun with v2 payload!";
}
@Bean
public Supplier<Sensor> supplier() {
return () -> unbounded.poll();
}
}