diff --git a/samples/pom.xml b/samples/pom.xml index bc0f85d27..dfbaa1cbe 100644 --- a/samples/pom.xml +++ b/samples/pom.xml @@ -22,7 +22,7 @@ - schema-registry-integration + spring-cloud-stream-schema-registry-integration confluent-schema-registry-integration kafka-streams-interactive-query diff --git a/samples/scst-schema-registry-integration/README.adoc b/samples/scst-schema-registry-integration/README.adoc deleted file mode 100644 index 428712db0..000000000 --- a/samples/scst-schema-registry-integration/README.adoc +++ /dev/null @@ -1,134 +0,0 @@ -== Spring Cloud Stream and Schema Evolution in Action with Kafka Binder. - -These are a set of Spring Boot applications to demonstrate Schema Evolution using Spring Cloud Stream with Kafka binder. -Producer V1 (`producer1`), Producer V2 (`producer2`), and Consumer (`consumer`) are included in this project. - -=== Requirement -As a developer, I'd like to design my consumer to be resilient to differing payload schemas. - -=== Assumptions -For this demonstration, we will simply assume there are two producers producing events with different payload schemas. -A consumer that consumes both the payload versions will be designed to adapt to evolving schemas. - -Both producers and consumers interact with the SCSt schema registry to register and evolve the schema. - -[[build-apps]] -=== Building the applications -To build the applications simply execute the following command from the `scst-schema-registry-integration` directory: -[source,bash] ----- -./mvnw clean install ----- -NOTE: The apps can be built and run from w/in an IDE (such as IntelliJ) but you will need to invoke the Maven `package` goal and then `refresh` the project as the Avro Maven plugin needs to execute so that it generates the required model classes - otherwise you will see compile failures around missing `Sensor` class. - -[[run-apps]] -=== Running the applications - -==== Pre-requisites -**** -* The components have all been built by following the <> steps. -* Apache Kafka broker available at `localhost:9092` - -By default, the schema registry is backed by an H2 database. If you want to instead use Postgres, it must be available at `localhost:5432`. Similarly, if you want to use MySQL it must be available at `localhost:3306`. - -TIP: The included link:../../../tools/kafka/docker-compose/README.adoc#_all_the_things[Kafka tools] can be used to easily start a broker locally on the required coordinates -**** - -==== Steps -Make sure the above pre-requisites are satisfied and that you are in the `scst-schema-registry-integration` directory and follow the steps below. - -- Start the Schema Registry server (Change these commands accordingly if you are not on a Unix like platform) -[source,bash] ----- -java -jar /schema-registry/spring-cloud-stream-schema-registry-server-.jar ----- - -This starts the schema registry with a local H2 database. - -To use Postgres instead of H2, additional properties must be specified when starting the server: - -[source,bash] ----- -java -jar /schema-registry/spring-cloud-stream-schema-registry-server-.jar \ - --spring.datasource.url=jdbc:postgresql://localhost:5432/registry \ - --spring.datasource.username=root \ - --spring.datasource.password=rootpw \ - --spring.datasource.driver-class-name=org.postgresql.Driver \ - --spring.jpa.database-platform=org.hibernate.dialect.PostgreSQLDialect \ - --spring.jpa.hibernate.ddl-auto-create=true \ - --spring.jpa.hibernate.ddl-auto=update \ - --spring.jpa.generate-ddl=true ----- - -To use MySQL database instead of H2, additional properties must be specified when starting the server: - -[source,bash] ----- -java -jar /schema-registry/spring-cloud-stream-schema-registry-server-.jar \ - --spring.datasource.url=jdbc:mariadb://localhost:3306/registry \ - --spring.datasource.username=root \ - --spring.datasource.password=rootpw \ - --spring.datasource.driver-class-name=org.mariadb.jdbc.Driver \ - --spring.jpa.database-platform=org.hibernate.dialect.MariaDB53Dialect \ - --spring.jpa.hibernate.ddl-auto-create=true \ - --spring.jpa.hibernate.ddl-auto=update \ - --spring.jpa.generate-ddl=true ----- - -- Start `consumer` on another terminal session (or run it from an IDE) -[source,bash] ----- -java -jar scst-schema-registry-consumer-kafka/target/scst-schema-registry-consumer-kafka--SNAPSHOT.jar ----- - -- Start `producer1` on another terminal session (or run it from an IDE) -[source,bash] ----- -java -jar scst-schema-registry-producer1-kafka/target/scst-schema-registry-producer1-kafka--SNAPSHOT.jar ----- -- Start `producer2` on another terminal session (or run it from an IDE) -[source,bash] ----- -java -jar scst-schema-registry-producer2-kafka/target/scst-schema-registry-producer2-kafka--SNAPSHOT.jar ----- - -=== 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. - -_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 ----- - -=== Output -The consumer should log the results. - -[source,bash,options=nowrap,subs=attributes] ----- -{"id": "d135efc3-72f8-4612-9497-184cae508e31-v1", "internalTemperature": 34.36362, "externalTemperature": 0.0, "acceleration": 9.656547, "velocity": 33.29733} -{"id": "3ecaae18-3144-4570-800a-223ca3198001-v1", "internalTemperature": 28.410656, "externalTemperature": 0.0, "acceleration": 1.752817, "velocity": 69.82016} -{"id": "81262d85-2d60-40e8-9fcc-7f797fd6dcd4-v2", "internalTemperature": 11.2908125, "externalTemperature": 26.260101, "acceleration": 3.268205, "velocity": 3.331542} ----- - -NOTE: Refer to the payload suffix in the `id` field. Each of them are appended with `-v1` or `-v2` indicating they are from -`producer1` and `producer2` respectively. - -=== What just happened? -The schema evolved on the `temperature` field. That field is now split into `internalTemperature` and `externalTemperature`, -as two separate fields. The `producer1` produces payload only with `temperature` and on the other hand, `producer2` produces -payload with `internalTemperature` and `externalTemperature` fields in it. - -The `consumer` is coded against a base schema that include the split fields. - -The `consumer` app can happily deserialize the payload with `internalTemperature` and `externalTemperature` fields. However, when -a `producer1` payload arrives (which includes `temperature` field), the schema evolution and compatibility check are automatically -applied. - -Because each payload also includes the payload version in the header, Spring Cloud Stream with the help of Schema Registry server and Avro, the schema evolution occurs behind the scenes. -The automatic mapping of `temperature` to `internalTemperature` field is applied, since that's the field where the `aliases` is defined. diff --git a/samples/spring-cloud-stream-schema-registry-integration/README.adoc b/samples/spring-cloud-stream-schema-registry-integration/README.adoc new file mode 100644 index 000000000..8abba24c2 --- /dev/null +++ b/samples/spring-cloud-stream-schema-registry-integration/README.adoc @@ -0,0 +1,174 @@ +== Spring Cloud Stream and Schema Evolution in Action with Kafka Binder. + +:project-version: 4.0.0-SNAPSHOT + +These are a set of Spring Boot applications to demonstrate Schema Evolution using Spring Cloud Stream with Kafka binder. +Producer V1 (`producer1`), Producer V2 (`producer2`), and Consumer (`consumer`) are included in this project. + +=== Requirement +As a developer, I'd like to design my consumer to be resilient to differing payload schemas. + +=== Assumptions +For this demonstration, we will simply assume there are two producers producing events with different payload schemas. +A consumer that consumes both the payload versions will be designed to adapt to evolving schemas. + +Both producers and consumers interact with the SCSt schema registry to register and evolve the schema. + +[[build-apps]] +=== Building + +==== +NOTE: It is expected that this repo has been checked out locally and All commands are executed from this sample's directory `spring-cloud-stream-schema-registry-integration` unless otherwise noted. +==== + +==== Build the schema registry +The schema registry must be built by executing the following command: +[source,bash] +---- +pushd ../../schema-registry +../mvnw clean install +popd +---- + +==== Build the apps +To build the applications simply execute the following command: +[source,bash] +---- +./mvnw clean install +---- + +NOTE: The apps can be built and run from w/in an IDE (such as IntelliJ) but you will need to invoke the Maven `package` goal and then `refresh` the project as the Avro Maven plugin needs to execute so that it generates the required model classes - otherwise you will see compile failures around missing `Sensor` class. + +[[run-apps]] +=== Running + +==== +NOTE: It is expected that this repo has been checked out locally and All commands are executed from this sample's directory `spring-cloud-stream-schema-registry-integration` unless otherwise noted. +==== + +==== Pre-requisites +**** +* The components have all been built by following the <> steps. +* Apache Kafka broker available at `localhost:9092` + +TIP: The included link:../../../tools/kafka/docker-compose/README.adoc#_all_the_things[Kafka tools] can be used to easily start a broker locally on the required coordinates + +* By default, the schema registry is backed by an `H2` database. +** To instead use `Postgres` it must be available at `localhost:5432` +** To instead use `MySQL` it must be available at `localhost:3306`. + +TIP: Docker compose files are provided for both link:./postgres.yml[Postgres] and link:./mysql.yml[MySQL]. You can simply run `docker-compose -f ` to start/stop the database server +**** + +==== Steps +Make sure the above pre-requisites are satisfied and follow the steps below. + +===== Start Schema Registry +Start the Schema Registry server (_adjust commands accordingly if you are not on a Unix like platform_) +[source,bash,subs="attributes"] +---- +java -jar ../../schema-registry/spring-cloud-stream-schema-registry-server/target/spring-cloud-stream-schema-registry-server-4.0.0-SNAPSHOT.jar +---- + +By default the schema registry starts with a local `H2` database. + +.To use `Postgres` database instead of `H2`... +[%collapsible] +==== +additional properties must be specified when starting the server: +[source,bash,subs="attributes"] +---- +java -jar ../../schema-registry/spring-cloud-stream-schema-registry-server/target/spring-cloud-stream-schema-registry-server-{project-version}.jar \ + --spring.datasource.url=jdbc:postgresql://localhost:5432/registry \ + --spring.datasource.username=root \ + --spring.datasource.password=rootpw \ + --spring.datasource.driver-class-name=org.postgresql.Driver \ + --spring.jpa.database-platform=org.hibernate.dialect.PostgreSQLDialect \ + --spring.jpa.hibernate.ddl-auto-create=true \ + --spring.jpa.hibernate.ddl-auto=update \ + --spring.jpa.generate-ddl=true +---- +==== + + +.To use `MySQL` database instead of `H2`... +[%collapsible] +==== +additional properties must be specified when starting the server: +[source,bash,subs="attributes"] +---- +java -jar ../../schema-registry/spring-cloud-stream-schema-registry-server/target/spring-cloud-stream-schema-registry-server-{project-version}.jar \ + --spring.datasource.url=jdbc:mariadb://localhost:3306/registry \ + --spring.datasource.username=root \ + --spring.datasource.password=rootpw \ + --spring.datasource.driver-class-name=org.mariadb.jdbc.Driver \ + --spring.jpa.database-platform=org.hibernate.dialect.MariaDB53Dialect \ + --spring.jpa.hibernate.ddl-auto-create=true \ + --spring.jpa.hibernate.ddl-auto=update \ + --spring.jpa.generate-ddl=true +---- +==== + +===== Start consumer +Start `consumer` on another terminal session (or run it from an IDE) +[source,bash,subs="attributes"] +---- +java -jar schema-registry-consumer-kafka/target/schema-registry-consumer-kafka-{project-version}.jar +---- + +===== Start V1 producer +Start `producer1` on another terminal session (or run it from an IDE) +[source,bash,subs="attributes"] +---- +java -jar schema-registry-producer1-kafka/target/schema-registry-producer1-kafka-{project-version}.jar +---- + +===== Start V2 producer +Start `producer2` on another terminal session (or run it from an IDE) +[source,bash,subs="attributes"] +---- +java -jar schema-registry-producer2-kafka/target/schema-registry-producer2-kafka-{project-version}.jar +---- + +=== Sample Data +Both the producers in the demonstration are _also_ REST controllers. We will hit the `/randomMessage` endpoint on each producer +to POST sample data. + +_Example:_ +[source,bash] +---- +curl -X POST http://localhost:9009/randomMessage +curl -X POST http://localhost:9010/randomMessage +curl -X POST http://localhost:9009/randomMessage +curl -X POST http://localhost:9009/randomMessage +curl -X POST http://localhost:9010/randomMessage +---- + +=== Output +The consumer should log the results. + +[source,bash,options=nowrap,subs=attributes] +---- +{"id": "d5657e55-c2cd-48f0-a22e-d28d1ef10873-v1", "internalTemperature": 19.534815, "externalTemperature": 0.0, "acceleration": 5.286502, "velocity": 25.349945} +{"id": "6a6de265-997c-4bf9-8eae-97accccb78e9-v2", "internalTemperature": 39.443855, "externalTemperature": 40.365253, "acceleration": 1.8879288, "velocity": 2.5296867} +{"id": "f09defad-828f-43ae-93a4-e777754cf57a-v1", "internalTemperature": 15.895501, "externalTemperature": 0.0, "acceleration": 1.9341749, "velocity": 52.868507} +{"id": "b39b8c73-eec3-4abd-b8d2-cc405eb39bd7-v1", "internalTemperature": 44.90698, "externalTemperature": 0.0, "acceleration": 1.5393275, "velocity": 87.0358} +{"id": "19d5c20e-ec18-4b35-a82a-c8322d7fea27-v2", "internalTemperature": 19.203693, "externalTemperature": 47.290142, "acceleration": 1.125809, "velocity": 11.153614} +---- + +NOTE: Refer to the payload suffix in the `id` field. Each of them are appended with `-v1` or `-v2` indicating they are from +`producer1` and `producer2` respectively. + +=== What just happened? +The schema evolved on the `temperature` field. That field is now split into `internalTemperature` and `externalTemperature`, +as two separate fields. The `producer1` produces payload only with `temperature` and on the other hand, `producer2` produces +payload with `internalTemperature` and `externalTemperature` fields in it. + +The `consumer` is coded against a base schema that include the split fields. + +The `consumer` app can happily deserialize the payload with `internalTemperature` and `externalTemperature` fields. However, when +a `producer1` payload arrives (which includes `temperature` field), the schema evolution and compatibility check are automatically +applied. + +Because each payload also includes the payload version in the header, Spring Cloud Stream with the help of Schema Registry server and Avro, the schema evolution occurs behind the scenes. +The automatic mapping of `temperature` to `internalTemperature` field is applied, since that's the field where the `aliases` is defined. diff --git a/samples/scst-schema-registry-integration/mvnw b/samples/spring-cloud-stream-schema-registry-integration/mvnw similarity index 100% rename from samples/scst-schema-registry-integration/mvnw rename to samples/spring-cloud-stream-schema-registry-integration/mvnw diff --git a/samples/scst-schema-registry-integration/mvnw.cmd b/samples/spring-cloud-stream-schema-registry-integration/mvnw.cmd similarity index 100% rename from samples/scst-schema-registry-integration/mvnw.cmd rename to samples/spring-cloud-stream-schema-registry-integration/mvnw.cmd diff --git a/samples/scst-schema-registry-integration/mysql.yml b/samples/spring-cloud-stream-schema-registry-integration/mysql.yml similarity index 100% rename from samples/scst-schema-registry-integration/mysql.yml rename to samples/spring-cloud-stream-schema-registry-integration/mysql.yml diff --git a/samples/scst-schema-registry-integration/pom.xml b/samples/spring-cloud-stream-schema-registry-integration/pom.xml similarity index 96% rename from samples/scst-schema-registry-integration/pom.xml rename to samples/spring-cloud-stream-schema-registry-integration/pom.xml index 689a6ef30..d1656b02a 100644 --- a/samples/scst-schema-registry-integration/pom.xml +++ b/samples/spring-cloud-stream-schema-registry-integration/pom.xml @@ -1,10 +1,10 @@ 4.0.0 - scst-schema-registry-integration + spring-cloud-stream-schema-registry-integration 4.0.0-SNAPSHOT pom - scst-schema-registry-integration + spring-cloud-stream-schema-registry-integration SCSt Schema Registry Integration Sample App diff --git a/samples/scst-schema-registry-integration/postgres.yml b/samples/spring-cloud-stream-schema-registry-integration/postgres.yml similarity index 100% rename from samples/scst-schema-registry-integration/postgres.yml rename to samples/spring-cloud-stream-schema-registry-integration/postgres.yml diff --git a/samples/scst-schema-registry-integration/schema-registry-consumer-kafka/pom.xml b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/pom.xml similarity index 78% rename from samples/scst-schema-registry-integration/schema-registry-consumer-kafka/pom.xml rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/pom.xml index 421e5ce29..38fd84e62 100644 --- a/samples/scst-schema-registry-integration/schema-registry-consumer-kafka/pom.xml +++ b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/pom.xml @@ -2,15 +2,15 @@ 4.0.0 - scst-schema-registry-consumer-kafka + schema-registry-consumer-kafka 4.0.0-SNAPSHOT jar - scst-schema-registry-consumer-kafka + schema-registry-consumer-kafka SCSt Schema Registry Consumer org.springframework.cloud - scst-schema-registry-integration + spring-cloud-stream-schema-registry-integration 4.0.0-SNAPSHOT diff --git a/samples/scst-schema-registry-integration/schema-registry-consumer-kafka/src/main/java/sample/consumer/ConsumerApplication.java b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/src/main/java/sample/consumer/ConsumerApplication.java similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-consumer-kafka/src/main/java/sample/consumer/ConsumerApplication.java rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/src/main/java/sample/consumer/ConsumerApplication.java diff --git a/samples/scst-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/application.yml b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/application.yml similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/application.yml rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/application.yml diff --git a/samples/scst-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/avro/sensor.avsc b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/avro/sensor.avsc similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/avro/sensor.avsc rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-consumer-kafka/src/main/resources/avro/sensor.avsc diff --git a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/mvnw b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/mvnw similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-producer1-kafka/mvnw rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/mvnw diff --git a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/mvnw.cmd b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/mvnw.cmd similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-producer1-kafka/mvnw.cmd rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/mvnw.cmd diff --git a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/pom.xml b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/pom.xml similarity index 79% rename from samples/scst-schema-registry-integration/schema-registry-producer1-kafka/pom.xml rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/pom.xml index f3e457dc6..057dcf205 100644 --- a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/pom.xml +++ b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/pom.xml @@ -2,15 +2,15 @@ 4.0.0 - scst-schema-registry-producer1-kafka + schema-registry-producer1-kafka 4.0.0-SNAPSHOT jar - scst-schema-registry-producer1-kafka + schema-registry-producer1-kafka SCSt Schema Registry Producer 1 org.springframework.cloud - scst-schema-registry-integration + spring-cloud-stream-schema-registry-integration 4.0.0-SNAPSHOT diff --git a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/src/main/java/sample/producer1/Producer1ApplicationKafka.java b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/src/main/java/sample/producer1/Producer1ApplicationKafka.java similarity index 90% rename from samples/scst-schema-registry-integration/schema-registry-producer1-kafka/src/main/java/sample/producer1/Producer1ApplicationKafka.java rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/src/main/java/sample/producer1/Producer1ApplicationKafka.java index 150204ffc..fd667b9aa 100644 --- a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/src/main/java/sample/producer1/Producer1ApplicationKafka.java +++ b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/src/main/java/sample/producer1/Producer1ApplicationKafka.java @@ -22,12 +22,18 @@ public class Producer1ApplicationKafka { private Random random = new Random(); @Autowired - StreamBridge streamBridge; + private StreamBridge streamBridge; public static void main(String[] args) { SpringApplication.run(Producer1ApplicationKafka.class, args); } + @RequestMapping(value = "/randomMessage", method = RequestMethod.POST) + public String sendRandomMessage() { + streamBridge.send("supplier-out-0", randomSensor()); + return "ok, have fun with v1 payload!"; + } + private Sensor randomSensor() { Sensor sensor = new Sensor(); sensor.setId(UUID.randomUUID() + "-v1"); @@ -36,12 +42,4 @@ public class Producer1ApplicationKafka { sensor.setTemperature(random.nextFloat() * 50); return sensor; } - - @RequestMapping(value = "/messages", method = RequestMethod.POST) - public String sendMessage() { - streamBridge.send("supplier-out-0", randomSensor()); - return "ok, have fun with v1 payload!"; - } - } - diff --git a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/application.yml b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/application.yml similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/application.yml rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/application.yml diff --git a/samples/scst-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/avro/sensor.avsc b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/avro/sensor.avsc similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/avro/sensor.avsc rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer1-kafka/src/main/resources/avro/sensor.avsc diff --git a/samples/scst-schema-registry-integration/schema-registry-producer2-kafka/pom.xml b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/pom.xml similarity index 79% rename from samples/scst-schema-registry-integration/schema-registry-producer2-kafka/pom.xml rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/pom.xml index 35dc7c28c..ad6f2f644 100644 --- a/samples/scst-schema-registry-integration/schema-registry-producer2-kafka/pom.xml +++ b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/pom.xml @@ -2,15 +2,15 @@ 4.0.0 - scst-schema-registry-producer2-kafka + schema-registry-producer2-kafka 4.0.0-SNAPSHOT jar - scst-schema-registry-producer2-kafka + schema-registry-producer2-kafka SCSt Schema Registry Producer 2 org.springframework.cloud - scst-schema-registry-integration + spring-cloud-stream-schema-registry-integration 4.0.0-SNAPSHOT diff --git a/samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/java/sample/producer2/Producer2ApplicationKafka.java b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/java/sample/producer2/Producer2ApplicationKafka.java similarity index 77% rename from samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/java/sample/producer2/Producer2ApplicationKafka.java rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/java/sample/producer2/Producer2ApplicationKafka.java index 1599122b7..32669e93e 100644 --- a/samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/java/sample/producer2/Producer2ApplicationKafka.java +++ b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/java/sample/producer2/Producer2ApplicationKafka.java @@ -1,13 +1,17 @@ package sample.producer2; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.cloud.stream.function.StreamBridge; import org.springframework.cloud.stream.schema.registry.client.EnableSchemaRegistryClient; import org.springframework.context.annotation.Bean; +import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController; +import java.util.Collections; import java.util.Random; import java.util.UUID; import java.util.concurrent.BlockingQueue; @@ -23,23 +27,19 @@ public class Producer2ApplicationKafka { private Random random = new Random(); - BlockingQueue unbounded = new LinkedBlockingQueue<>(); + @Autowired + private StreamBridge streamBridge; public static void main(String[] args) { SpringApplication.run(Producer2ApplicationKafka.class, args); } - @RequestMapping(value = "/messages", method = RequestMethod.POST) - public String sendMessage() { - unbounded.offer(randomSensor()); + @PostMapping("/randomMessage") + public String sendRandomMessage() { + streamBridge.send("supplier-out-0", randomSensor()); return "ok, have fun with v2 payload!"; } - @Bean - public Supplier supplier() { - return () -> unbounded.poll(); - } - private Sensor randomSensor() { Sensor sensor = new Sensor(); sensor.setId(UUID.randomUUID().toString() + "-v2"); @@ -47,9 +47,6 @@ public class Producer2ApplicationKafka { sensor.setVelocity(random.nextFloat() * 100); sensor.setInternalTemperature(random.nextFloat() * 50); sensor.setExternalTemperature(random.nextFloat() * 50); - sensor.setAccelerometer(null); - sensor.setMagneticField(null); return sensor; } } - diff --git a/samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/application.yml b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/application.yml similarity index 100% rename from samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/application.yml rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/application.yml diff --git a/samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/avro/sensor.avsc b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/avro/sensor.avsc similarity index 54% rename from samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/avro/sensor.avsc rename to samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/avro/sensor.avsc index 8d2e60535..5da668b3b 100644 --- a/samples/scst-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/avro/sensor.avsc +++ b/samples/spring-cloud-stream-schema-registry-integration/schema-registry-producer2-kafka/src/main/resources/avro/sensor.avsc @@ -7,19 +7,6 @@ {"name":"internalTemperature", "type":"float", "default":0.0, "aliases":["temperature"]}, {"name":"externalTemperature", "type":"float", "default":0.0}, {"name":"acceleration", "type":"float","default":0.0}, - {"name":"velocity","type":"float","default":0.0}, - {"name":"accelerometer","type":[ - "null",{ - "type":"array", - "items":"float" - } - ]}, - {"name":"magneticField","type":[ - "null",{ - "type":"array", - "items":"float" - } - ]} + {"name":"velocity","type":"float","default":0.0} ] - -} \ No newline at end of file +}