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

Cleanup Confluent schema registry sample code

Browse Source 
* Use best practices
* Use SL4J
* Tidy formatting
This commit is contained in:
Chris Bono
2022-05-06 13:46:49 -05:00
parent 835d70b4a7
commit 807b71d726
7 changed files with 36 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
samples/avro-samples/confluent-schema-registry-integration/README.adoc
View File

@@ -76,11 +76,11 @@ java -jar target/confluent-schema-registry-integration-producer2-4.0.0-SNAPSHOT.
The producer apps each expose a REST endpoint which sends a sample Kafka event when invoked. Execute the following commands to send some sample data.
[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: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
----
===== View consumer output

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

@@ -3,24 +3,25 @@ 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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
@SpringBootApplication(proxyBeanMethods = false)
public class ConfluentAvroConsumerApplication {
private final Log logger = LogFactory.getLog(getClass());
private final Logger logger = LoggerFactory.getLogger(ConfluentAvroConsumerApplication.class);
public static void main(String[] args) {
SpringApplication.run(ConfluentAvroConsumerApplication.class, args);
}
@Bean
public Consumer<Sensor> process() {
return input -> logger.info("input: " + input);
Consumer<Sensor> process() {
return input -> logger.info("input: {}", input);
}
}

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

@@ -7,14 +7,12 @@ spring:
consumer:
useNativeDecoding: true
kafka:
# binder:
# consumerProperties:
# value:
# subject:
# name:
# strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy
bindings:
process-in-0:
consumer:

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

@@ -9,11 +9,10 @@ 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.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
@SpringBootApplication
@SpringBootApplication(proxyBeanMethods = false)
@RestController
public class ConfluentAvroProducer1Application {
@@ -26,6 +25,12 @@ public class ConfluentAvroProducer1Application {
SpringApplication.run(ConfluentAvroProducer1Application.class, args);
}
@PostMapping("/randomMessage")
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");
@@ -34,13 +39,6 @@ public class ConfluentAvroProducer1Application {
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!";
}
}

2
samples/avro-samples/confluent-schema-registry-integration/confluent-schema-registry-integration-producer1/src/main/resources/application.yml
View File

@@ -13,11 +13,11 @@ spring:
# subject:
# name:
# strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy
bindings:
supplier-out-0:
producer:
configuration:
value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url: http://localhost:8081
server.port: 9009

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

@@ -9,11 +9,10 @@ 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.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
@SpringBootApplication
@SpringBootApplication(proxyBeanMethods = false)
@RestController
public class ConfluentAvroProducer2Application {
@@ -26,6 +25,12 @@ public class ConfluentAvroProducer2Application {
SpringApplication.run(ConfluentAvroProducer2Application.class, args);
}
@PostMapping("/randomMessage")
public String sendRandomMessage() {
streamBridge.send("supplier-out-0", randomSensor());
return "ok, have fun with v2 payload!";
}
private Sensor randomSensor() {
Sensor sensor = new Sensor();
sensor.setId(UUID.randomUUID().toString() + "-v2");
@@ -36,11 +41,5 @@ public class ConfluentAvroProducer2Application {
sensor.setMagneticField(null);
return sensor;
}
@RequestMapping(value = "/messages", method = RequestMethod.POST)
public String sendMessage() {
streamBridge.send("supplier-out-0", randomSensor());
return "ok, have fun with v2 payload!";
}
}

14
samples/avro-samples/confluent-schema-registry-integration/confluent-schema-registry-integration-producer2/src/main/resources/application.yml
View File

@@ -7,17 +7,17 @@ spring:
producer:
useNativeEncoding: true
kafka:
# binder:
# producerProperties:
# value:
# subject:
# name:
# strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy
# binder:
# producerProperties:
# value:
# subject:
# name:
# strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy
bindings:
supplier-out-0:
producer:
configuration:
value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url: http://localhost:8081
server.port: 9010