home/spring-cloud-stream-samples
2
0
Fork 0
mirror of https://github.com/wahyd4/spring-cloud-stream-samples.git synced 2026-08-08 21:06:34 +10:00
Code Issues Packages Projects Releases Wiki Activity

Schema registry samples update - interation 2

Browse Source 
Updating kafka-streams-schema-evolution with Confluent Scheam registry samples
This commit is contained in:
Soby Chacko
2019-09-18 16:30:25 -04:00
parent ff136d56c2
commit b1025b0c6c
10 changed files with 205 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer1/pom.xml
+76 -6
View File
@@ -9,17 +9,31 @@
<description>Custom Confluent 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>
<confluent.version>4.0.0</confluent.version>
<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>
@@ -97,9 +111,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>
schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer1/src/main/java/sample/producer1/FooSerde.java → schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer1/src/main/java/sample/producer1/FooSerializer.java
+1 -1
View File
@@ -10,7 +10,7 @@ import java.util.Map;
/**
* @author Soby Chacko
*/
public class FooSerde extends SpecificAvroSerializer<Sensor> {
public class FooSerializer extends SpecificAvroSerializer<Sensor> {
@Override
public void configure(Map<String, ?> serializerConfig, boolean isSerializerForRecordKeys) {
schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer1/src/main/java/sample/producer1/Producer1Application.java
+19 -11
View File
@@ -6,6 +6,7 @@ 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.context.annotation.Bean;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
@@ -13,27 +14,22 @@ 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;
@SpringBootApplication
@EnableBinding(Source.class)
@RestController
public class Producer1Application {
@Autowired
private Source source;
public class Producer1Application{
private Random random = new Random();
BlockingQueue<Sensor> unbounded = new LinkedBlockingQueue<>();
public static void main(String[] args) {
SpringApplication.run(Producer1Application.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");
@@ -42,7 +38,19 @@ public class Producer1Application {
sensor.setTemperature(random.nextFloat() * 50);
return sensor;
}
@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();
}
}
schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer1/src/main/resources/application.yml
+1 -1
View File
@@ -10,4 +10,4 @@ server.port: 9009
spring.cloud.stream.kafka.binder.configuration:
schema.registry.url: http://localhost:8081
key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer
value.serializer: sample.producer1.FooSerde
value.serializer: sample.producer1.FooSerializer
schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer2/src/main/java/sample/producer2/FooSerde.java → schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer2/src/main/java/sample/producer2/FooSerializer.java
+1 -1
View File
@@ -10,7 +10,7 @@ import java.util.Map;
/**
* @author Soby Chacko
*/
public class FooSerde extends SpecificAvroSerializer<Sensor> {
public class FooSerializer extends SpecificAvroSerializer<Sensor> {
@Override
public void configure(Map<String, ?> serializerConfig, boolean isSerializerForRecordKeys) {
schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer2/src/main/java/sample/producer2/Producer2Application.java
+19 -11
View File
@@ -6,6 +6,7 @@ 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.context.annotation.Bean;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
@@ -13,27 +14,22 @@ 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;
@SpringBootApplication
@EnableBinding(Source.class)
@RestController
public class Producer2Application {
@Autowired
private Source source;
public class Producer2Application{
private Random random = new Random();
BlockingQueue<Sensor> unbounded = new LinkedBlockingQueue<>();
public static void main(String[] args) {
SpringApplication.run(Producer2Application.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");
@@ -44,5 +40,17 @@ public class Producer2Application {
sensor.setMagneticField(null);
return sensor;
}
@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();
}
}
schema-registry-samples/kafka-streams-schema-evolution/custom-confluent-producer2/src/main/resources/application.yml
+1 -1
View File
@@ -10,4 +10,4 @@ server.port: 9010
spring.cloud.stream.kafka.binder.configuration:
schema.registry.url: http://localhost:8081
key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer
value.serializer: sample.producer2.FooSerde
value.serializer: sample.producer2.FooSerializer
schema-registry-samples/kafka-streams-schema-evolution/kafka-streams-confluent-consumer/pom.xml
+76 -13
View File
@@ -1,6 +1,5 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>kafka-streams-confluent-consumer</artifactId>
@@ -10,22 +9,31 @@
<description>Kafka Streams Confluent 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>
<confluent.version>4.0.0</confluent.version>
<avro.version>1.8.2</avro.version>
<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
<confluent.version>4.0.0</confluent.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-schema</artifactId>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
@@ -35,7 +43,6 @@
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-streams-avro-serde</artifactId>
@@ -87,9 +94,65 @@
</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>
schema-registry-samples/kafka-streams-schema-evolution/kafka-streams-confluent-consumer/src/main/java/sample/consumer/CountVersionApplication.java
+9 -5
View File
@@ -20,15 +20,16 @@ import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.Collections;
import java.util.Map;
import java.util.function.Function;
@SpringBootApplication
@EnableBinding(KafkaStreamsProcessor.class)
@EnableScheduling
public class CountVersionApplication {
@@ -45,9 +46,12 @@ public class CountVersionApplication {
SpringApplication.run(CountVersionApplication.class, args);
}
@StreamListener("input")
@SendTo("output")
public KStream<String, Long> process(KStream<Object, Sensor> input) {
@Bean
public Function<KStream<Object, Sensor>, KStream<String, Long>> process() {
//The following Serde definitions are not needed in the topoloyy below
//as we are not using it. However, if your topoloyg explicitly uses this
//Serde, you need to configure this with the schema registry url as below.
final Map<String, String> serdeConfig = Collections.singletonMap(
AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
@@ -55,7 +59,7 @@ public class CountVersionApplication {
final SpecificAvroSerde<Sensor> sensorSerde = new SpecificAvroSerde<>();
sensorSerde.configure(serdeConfig, false);
return input
return input -> input
.map((k, value) -> {
String newKey = "v1";
if (value.getId().toString().endsWith("v2")) {
schema-registry-samples/kafka-streams-schema-evolution/kafka-streams-confluent-consumer/src/main/resources/application.yml
+2 -4
View File
@@ -1,11 +1,9 @@
server.port: 9998
spring.application.name: kafka-streams-confluent-consumer
spring.cloud.stream.bindings.output:
spring.cloud.stream.bindings.process_out:
destination: sensor-versions
spring.cloud.stream.bindings.input:
spring.cloud.stream.bindings.process_in:
destination: sensors
consumer:
useNativeDecoding: true
spring.cloud.stream.kafka.streams.binder:
brokers: localhost
configuration: