diff --git a/schema-registry-samples/schema-registry-confluent/README.adoc b/schema-registry-samples/schema-registry-confluent/README.adoc index 75437f6..6b70c44 100644 --- a/schema-registry-samples/schema-registry-confluent/README.adoc +++ b/schema-registry-samples/schema-registry-confluent/README.adoc @@ -30,6 +30,10 @@ In order to run this sample, you need to set compatibility to `NONE` on Confluen `curl -X PUT http://127.0.0.1:8081/config -d '{"compatibility": "NONE"}' -H "Content-Type:application/json"` +If you have sensor schema already registered, you might want to consider deleting the registered schema for that to run these apps with a clean slate. + +For e.g., `curl -X DELETE http://localhost:8081/subjects/sensor` + - Start `consumer` on another terminal session [source,bash] ---- @@ -40,6 +44,19 @@ java -jar schema-registry-confluent-consumer/target/schema-registry-confluent-co ---- java -jar schema-registry-confluent-producer1/target/schema-registry-confluent-producer1-0.0.1-SNAPSHOT.jar ---- + +=== Sample Data +The producers are _also_ REST controllers. We will hit the `/messages` endpoint of the first producer to POST sample data. + +_Example:_ +[source,bash] +---- +curl -X POST http://localhost:9009/messages +curl -X POST http://localhost:9009/messages +curl -X POST http://localhost:9009/messages +---- + + - Start `producer2` on another terminal session [source,bash] ---- @@ -47,16 +64,13 @@ java -jar schema-registry-confluent-producer2/target/schema-registry-confluent-p ---- === Sample Data -Both the producers in the demonstration are _also_ REST controllers. We will hit the `/messages` endpoint on each producer -to POST sample data. +We will hit the `/messages` endpoint on the second producer to POST sample data. _Example:_ [source,bash] ---- -curl -X POST http://localhost:9009/messages curl -X POST http://localhost:9010/messages -curl -X POST http://localhost:9009/messages -curl -X POST http://localhost:9009/messages +curl -X POST http://localhost:9010/messages curl -X POST http://localhost:9010/messages ---- diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml index 15412a3..219afc2 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/pom.xml @@ -9,22 +9,46 @@ Schema Registry Consumer - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../../.. + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + 1.8.2 4.0.0 + Hoxton.BUILD-SNAPSHOT + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + + org.springframework.cloud - spring-cloud-stream-schema + spring-cloud-stream-binder-kafka + + org.springframework.cloud + spring-cloud-schema-registry-client + + + org.apache.avro + avro + ${avro.version} + + io.confluent kafka-avro-serializer @@ -47,30 +71,6 @@ - - - kafka-binder - - true - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - - - - - rabbit-binder - - - org.springframework.cloud - spring-cloud-stream-binder-rabbit - - - - - @@ -99,10 +99,65 @@ + + 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/release + + false + + confluent https://packages.confluent.io/maven/ - - + + + 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 + + + + \ No newline at end of file diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/java/sample/consumer/ConsumerApplication.java b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/java/sample/consumer/ConsumerApplication.java index cbc3cbe..c1a57bf 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/java/sample/consumer/ConsumerApplication.java +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-consumer/src/main/java/sample/consumer/ConsumerApplication.java @@ -1,22 +1,20 @@ package sample.consumer; +import java.util.function.Consumer; + import com.example.Sensor; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.annotation.Value; 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.Sink; -import org.springframework.cloud.stream.schema.client.ConfluentSchemaRegistryClient; -import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient; -import org.springframework.cloud.stream.schema.client.SchemaRegistryClient; +import org.springframework.cloud.schema.registry.client.ConfluentSchemaRegistryClient; +import org.springframework.cloud.schema.registry.client.EnableSchemaRegistryClient; +import org.springframework.cloud.schema.registry.client.SchemaRegistryClient; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @SpringBootApplication -@EnableBinding(Sink.class) @EnableSchemaRegistryClient public class ConsumerApplication { @@ -26,9 +24,9 @@ public class ConsumerApplication { SpringApplication.run(ConsumerApplication.class, args); } - @StreamListener(Sink.INPUT) - public void process(Sensor data) { - logger.info(data); + @Bean + public Consumer process() { + return input -> logger.info("input: " + input); } @Configuration diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml index 3475b2e..bb541c2 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/pom.xml @@ -9,21 +9,43 @@ Schema Registry Producer1 - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../../.. + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + 1.8.2 4.0.0 + Hoxton.BUILD-SNAPSHOT + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + org.springframework.cloud - spring-cloud-stream-schema + spring-cloud-stream-binder-kafka + + + org.springframework.cloud + spring-cloud-schema-registry-client + + + org.apache.avro + avro + ${avro.version} io.confluent @@ -47,52 +69,6 @@ - - - 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 - - - - - - - @@ -119,10 +95,67 @@ + + + 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/release + + false + + confluent https://packages.confluent.io/maven/ - + + + 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 + + + + \ No newline at end of file diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/java/sample/producer1/Producer1Application.java b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/java/sample/producer1/Producer1Application.java index 23d3e2f..bb8492d 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/java/sample/producer1/Producer1Application.java +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer1/src/main/java/sample/producer1/Producer1Application.java @@ -1,47 +1,38 @@ package sample.producer1; -import com.example.Sensor; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -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.cloud.stream.schema.client.ConfluentSchemaRegistryClient; -import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient; -import org.springframework.cloud.stream.schema.client.SchemaRegistryClient; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -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.beans.factory.annotation.Value; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.cloud.schema.registry.client.ConfluentSchemaRegistryClient; +import org.springframework.cloud.schema.registry.client.EnableSchemaRegistryClient; +import org.springframework.cloud.schema.registry.client.SchemaRegistryClient; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +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) @EnableSchemaRegistryClient @RestController public class Producer1Application { - @Autowired - private Source source; - private Random random = new Random(); + BlockingQueue 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"); @@ -51,19 +42,17 @@ public class Producer1Application { 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 supplier() { + return () -> unbounded.poll(); + } + @Configuration static class ConfluentSchemaRegistryConfiguration { @Bean @@ -73,7 +62,6 @@ public class Producer1Application { return client; } } - } diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml index c6797eb..22cfb26 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/pom.xml @@ -6,24 +6,46 @@ 0.0.1-SNAPSHOT jar schema-registry-confluent-producer2 - Schema Registry Producer2 + Schema Registry Producer1 - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../../.. + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + 1.8.2 4.0.0 + Hoxton.BUILD-SNAPSHOT + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + org.springframework.cloud - spring-cloud-stream-schema + spring-cloud-stream-binder-kafka + + + org.springframework.cloud + spring-cloud-schema-registry-client + + + org.apache.avro + avro + ${avro.version} io.confluent @@ -47,52 +69,6 @@ - - - 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 - - - - - - - @@ -119,10 +95,67 @@ + + + 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/release + + false + + confluent https://packages.confluent.io/maven/ - + + + 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 + + + + \ No newline at end of file diff --git a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/src/main/java/sample/producer2/Producer2Application.java b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/src/main/java/sample/producer2/Producer2Application.java index 8cb1960..c5601fd 100644 --- a/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/src/main/java/sample/producer2/Producer2Application.java +++ b/schema-registry-samples/schema-registry-confluent/schema-registry-confluent-producer2/src/main/java/sample/producer2/Producer2Application.java @@ -1,47 +1,36 @@ package sample.producer2; -import com.example.Sensor; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -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.cloud.stream.schema.client.ConfluentSchemaRegistryClient; -import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient; -import org.springframework.cloud.stream.schema.client.SchemaRegistryClient; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -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.beans.factory.annotation.Value; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.cloud.schema.registry.client.ConfluentSchemaRegistryClient; +import org.springframework.cloud.schema.registry.client.SchemaRegistryClient; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RestController; @SpringBootApplication -@EnableSchemaRegistryClient -@EnableBinding(Source.class) @RestController public class Producer2Application { - @Autowired - private Source source; - private Random random = new Random(); + BlockingQueue 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"); @@ -53,19 +42,17 @@ public class Producer2Application { 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 supplier() { + return () -> unbounded.poll(); + } + @Configuration static class ConfluentSchemaRegistryConfiguration { @Bean