diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml index 678b5f8..e864ebe 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/pom.xml @@ -9,17 +9,30 @@ 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 @@ -73,12 +86,66 @@ - + + 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-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java index 640abf7..bee0aab 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/java/sample/consumer/ConfluentAvroConsumerApplication.java @@ -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 process() { + return input -> logger.info("input: " + input); } } \ No newline at end of file diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml index 64987c8..37b9e66 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-consumer/src/main/resources/application.yml @@ -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 \ No newline at end of file diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml index 20b3068..8cf4d41 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/pom.xml @@ -9,17 +9,31 @@ 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 @@ -45,6 +59,18 @@ kafka-schema-registry-client ${confluent.version} + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + @@ -74,9 +100,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-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/java/sample/producer1/ConfluentAvroProducer1Application.java b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/java/sample/producer1/ConfluentAvroProducer1Application.java index 8c317e4..1755842 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/java/sample/producer1/ConfluentAvroProducer1Application.java +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer1/src/main/java/sample/producer1/ConfluentAvroProducer1Application.java @@ -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 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 supplier() { + return () -> unbounded.poll(); + } + } diff --git a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml index 5ff44a0..5c57d5e 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/pom.xml @@ -6,20 +6,33 @@ 0.0.1-SNAPSHOT jar schema-registry-confluent-avro-serializer-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 @@ -45,6 +58,18 @@ kafka-schema-registry-client ${confluent.version} + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + @@ -74,9 +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-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/java/sample/producer2/ConfluentAvroProducer2Application.java b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/java/sample/producer2/ConfluentAvroProducer2Application.java index c1f1084..9278db0 100644 --- a/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/java/sample/producer2/ConfluentAvroProducer2Application.java +++ b/schema-registry-samples/schema-registry-confluent-avro-serializer/schema-registry-confluent-avro-serializer-producer2/src/main/java/sample/producer2/ConfluentAvroProducer2Application.java @@ -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 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 supplier() { + return () -> unbounded.poll(); + } }