Native Hints for Avro SchemaRegistryClient

Author: garyrussell

docker-compose.yml

@@ -0,0 +1,31 @@
+version: '3.1'
+services:
+  avro:
+    image: spring-native-kafka-avro:0.0.1-SNAPSHOT
+    ports:
+      - 9999:9999
+    depends_on:
+      - schema-registry
+  kafka:
+    image: wurstmeister/kafka:latest
+    ports:
+      - "9092:9092"
+    environment:
+      - KAFKA_ADVERTISED_HOST_NAME=kafka
+      - KAFKA_ADVERTISED_PORT=9092
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+    depends_on:
+      - zookeeper
+  zookeeper:
+    image: wurstmeister/zookeeper:latest
+    ports:
+      - "2181:2181"
+    environment:
+      - KAFKA_ADVERTISED_HOST_NAME=zookeeper
+  schema-registry:
+    image: confluentinc/cp-schema-registry:latest
+    ports:
+      - "8081:8081"
+    environment:
+      - SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=kafka:9092
+      - SCHEMA_REGISTRY_HOST_NAME=schema-registry

gradlew

@@ -3,6 +3,9 @@
 import com.sample.avro.Address;
 import io.confluent.kafka.schemaregistry.client.rest.entities.ErrorMessage;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaString;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaTypeConverter;
+import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaRequest;
+import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaResponse;
 import io.confluent.kafka.serializers.KafkaAvroSerializer;
 import io.confluent.kafka.serializers.subject.TopicNameStrategy;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
@@ -35,6 +38,9 @@
                 SpecificAvroSerde.class,
                 KafkaAvroSerializer.class,
                 ErrorMessage.class,
+                RegisterSchemaRequest.class,
+                RegisterSchemaResponse.class,
+                SchemaTypeConverter.class,
                 SchemaString.class
         })
 )

src/main/java/com/sample/config/AotConfig.java

@@ -4,6 +4,7 @@
 import com.sample.model.AddrReq;
 import com.sample.topology.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.config.StreamsBuilderFactoryBean;
 import org.springframework.web.bind.annotation.*;
 
 import java.util.UUID;
@@ -14,13 +15,19 @@ public class KafkaController {
 
     private final Producer producer;
 
+    private final StreamsBuilderFactoryBean fb;
+
     @Autowired
-    KafkaController(Producer producer) {
+    KafkaController(Producer producer, StreamsBuilderFactoryBean fb) {
         this.producer = producer;
+        this.fb = fb;
     }
 
     @PostMapping
     public void sendMessageToKafkaTopic(@RequestBody AddrReq addr) {
+        if (!this.fb.isRunning()) {
+            fb.start();
+        }
         Address address = Address.newBuilder().setAddressId(UUID.randomUUID().toString()).setAddressPostcode(addr.getPostcode()).build();
         this.producer.sendMessage(address);
     }

src/main/java/com/sample/controllers/KafkaController.java

@@ -6,6 +6,12 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 @Configuration
 @Slf4j
@@ -21,7 +27,18 @@ public Producer(KafkaTemplate<String, Address> kafkaTemplate) {
     }
 
     public void sendMessage(Address address) {
-        this.kafkaTemplate.send(this.TOPIC, address.getAddressId(), address);
+        Future<SendResult<String, Address>> future = this.kafkaTemplate.send(this.TOPIC, address.getAddressId(), address);
+        try {
+            SendResult<String, Address> result = future.get(10, TimeUnit.SECONDS);
+            log.info(result.getRecordMetadata().toString());
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            e.printStackTrace();
+        } catch (ExecutionException e) {
+            e.printStackTrace();
+        } catch (TimeoutException e) {
+            e.printStackTrace();
+        }
         log.info(String.format("Produced address -> %s", address));
     }
 

src/main/java/com/sample/topology/Producer.java

@@ -1,32 +1,32 @@
 server:
-  port: 8080
+  port: 9999
 
 spring:
   application:
     name: "spring-native-kafka-streams"
 
   kafka:
-    bootstrap-servers: localhost:9092
+    bootstrap-servers: kafka:9092
     streams:
-      bootstrap-servers: localhost:9092
+      bootstrap-servers: kafka:9092
       application-id: ${spring.application.name}
       client-id: ${spring.application.name}-stream
       auto-offset-reset: earliest
       properties:
         default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
         default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
         default.deserialization.exception.handler: org.apache.kafka.streams.errors.LogAndFailExceptionHandler
+      auto-startup: false
     properties:
-      schema.registry.url: http://localhost:8081
+      schema.registry.url: http://schema-registry:8081
       bootstrap.servers: ${spring.kafka.bootstrap-servers}
     producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
 
-
 logging:
-  pattern:
-    console: "%d{yyyy-MM-dd HH:mm:ss} ${LOG_LEVEL_PATTERN:-%5p} %m%n"
+#  pattern:
+#    console: "%d{yyyy-MM-dd HH:mm:ss} ${LOG_LEVEL_PATTERN:-%5p} %m%n"
   level:
     org.springframework.kafka.config: DEBUG