Now it can adapt to any schema.

zaxxio · zaxxio · commit 2df3b4d3c96f · 2024-03-18T02:52:16.000+06:00
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -42,6 +42,8 @@ services:
     environment:
       SCHEMA_REGISTRY_HOST_NAME: schema-registry
       SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "PLAINTEXT://kafka-1:19092"
+      # SCHEMA_COMPATIBILITY: BACKWARD
+      SCHEMA_COMPATIBILITY: FULL_TRANSITIVE
     depends_on:
       - zookeeper
       - kafka-1
diff --git a/src/main/avro/SensorEvent.avsc b/src/main/avro/SensorEvent.avsc
@@ -1,7 +1,7 @@
 {
   "type": "record",
   "name": "SensorEventAvro",
-  "namespace": "org.wsd.app.avro",
+  "namespace": "org.wsd.app.event",
   "fields": [
     {
       "name": "id",
@@ -12,8 +12,9 @@
       "type": "double"
     },
     {
-      "name": "y",
-      "type": "double"
+      "name": "z",
+      "type": "double",
+      "default": "-1"
     }
   ]
 }
diff --git a/src/main/java/org/wsd/app/bootloader/BootLoader.java b/src/main/java/org/wsd/app/bootloader/BootLoader.java
@@ -23,28 +23,16 @@
 package org.wsd.app.bootloader;
 
 import com.github.javafaker.Faker;
-import io.micrometer.observation.annotation.Observed;
 import lombok.RequiredArgsConstructor;
 import org.springframework.boot.CommandLineRunner;
-import org.springframework.data.mongodb.core.query.Query;
-import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
-import org.wsd.app.avro.SensorEventAvro;
-import org.wsd.app.config.TopicConfiguration;
 import org.wsd.app.domain.PhotoEntity;
 import org.wsd.app.mongo.Address;
 import org.wsd.app.mongo.Gender;
 import org.wsd.app.mongo.Person;
 import org.wsd.app.mongo.PersonRepository;
 import org.wsd.app.repository.PhotoRepository;
-import org.wsd.app.repository.UserRepository;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.stream.Stream;
 
 @Service
 @RequiredArgsConstructor
@@ -72,12 +60,6 @@ public void run(String... args) throws Exception {
         person.setAddress(address);
         person.setGender(Gender.MALE);
 
-        SensorEventAvro eventAvro = SensorEventAvro.newBuilder()
-                .setId(UUID.randomUUID().toString())
-                .setX(1)
-                .setY(Math.random())
-                .build();
-
 
         // personRepository.save(person);
 
diff --git a/src/main/java/org/wsd/app/config/KafkaConfig.java b/src/main/java/org/wsd/app/config/KafkaConfig.java
@@ -22,7 +22,10 @@
 
 package org.wsd.app.config;
 
+import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
+import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
+import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
 import io.confluent.kafka.serializers.KafkaAvroSerializer;
 import lombok.extern.log4j.Log4j2;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -54,6 +57,7 @@
 import org.springframework.retry.policy.SimpleRetryPolicy;
 import org.springframework.retry.support.RetryTemplate;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
@@ -117,16 +121,26 @@ public RetryTemplate retryTemplate() {
         return retryTemplate;
     }
 
+
     @Bean
     public ConsumerFactory<String, Object> consumerFactory() {
         final Map<String, Object> configProps = new HashMap<>();
         configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
 
         configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, UUIDDeserializer.class.getName());
-        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
-//        // Specify the actual deserializer classes as properties of ErrorHandlingDeserializer
-//        configProps.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, UUIDDeserializer.class.getName());
-//        configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, KafkaAvroDeserializer.class.getName());
+        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class.getName());
+        configProps.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, UUIDDeserializer.class.getName());
+        configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, KafkaAvroDeserializer.class.getName());
+
+        // Configure the trusted package for Avro deserialization
+        configProps.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
+        configProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
+
+        configProps.put("specific.avro.reader.schema.trusted", "true");
+        configProps.put("schema.registry.compatibility.level", "FULL_TRANSITIVE");
+
+        // Specify the schema registry URL
+
 
         // Trusted package
         configProps.put(JsonDeserializer.TRUSTED_PACKAGES, "org.wsd.app.event");
@@ -150,9 +164,6 @@ public ConsumerFactory<String, Object> consumerFactory() {
         configProps.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "300000");
         configProps.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "305000");
         configProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "60000");
-
-        configProps.put("schema.registry.url", "http://localhost:8081");
-        configProps.put("specific.avro.read", "true");
         return new DefaultKafkaConsumerFactory<>(configProps);
     }
 
diff --git a/src/main/java/org/wsd/app/messaging/pubs/ProducerService.java b/src/main/java/org/wsd/app/messaging/pubs/ProducerService.java
@@ -35,7 +35,7 @@
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.annotation.Transactional;
-import org.wsd.app.avro.SensorEventAvro;
+import org.wsd.app.event.SensorEventAvro;
 import org.wsd.app.config.TopicConfiguration;
 import org.wsd.app.repository.UserRepository;
 
@@ -48,15 +48,14 @@
 public class ProducerService {
     private final UserRepository userRepository;
 
-
     private final KafkaTemplate<UUID, SensorEventAvro> sensorEventAvroKafkaTemplate;
 
     @Bean
     public CommandLineRunner commandLineRunner(ProducerService producerService) {
         return args -> {
             for (int i = 0; i < 10; i++) {
                 try {
-                    producerService.process(i);
+                    // producerService.process(i);
                 } catch (Exception e) {
                     System.out.println(e.getMessage());
                 }
@@ -66,19 +65,18 @@ public CommandLineRunner commandLineRunner(ProducerService producerService) {
 
     @Transactional(value = "transactionManager", rollbackFor = Exception.class)
     public void process(int i) throws TransactionException {
-        final SensorEventAvro sensorEvent = new SensorEventAvro();
-        sensorEvent.setX(i);
-        sensorEvent.setY(Math.random());
+
 
         SensorEventAvro eventAvro = SensorEventAvro.newBuilder()
                 .setId(UUID.randomUUID().toString())
                 .setX(1)
-                .setY(Math.random())
+                .setZ(Math.random())
                 .build();
 
         final Message<SensorEventAvro> message = MessageBuilder
                 .withPayload(eventAvro)
                 .setHeader(KafkaHeaders.KEY, UUID.randomUUID())
+                .setHeader("schema.version","V2")
                 .setHeader(KafkaHeaders.TOPIC, TopicConfiguration.SENSOR)
                 .setHeader(KafkaHeaders.TIMESTAMP, System.currentTimeMillis())
                 .build();
diff --git a/src/main/java/org/wsd/app/messaging/subs/ConsumerService.java b/src/main/java/org/wsd/app/messaging/subs/ConsumerService.java
@@ -24,31 +24,19 @@
 
 import lombok.extern.log4j.Log4j2;
 import org.apache.avro.generic.GenericRecord;
-import org.apache.kafka.clients.NetworkClient;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.springframework.context.annotation.Lazy;
 import org.springframework.kafka.annotation.DltHandler;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.annotation.RetryableTopic;
-import org.springframework.kafka.annotation.TopicPartition;
 import org.springframework.kafka.support.Acknowledgment;
-import org.springframework.kafka.support.converter.ConversionException;
-import org.springframework.kafka.support.serializer.DeserializationException;
-import org.springframework.messaging.converter.MessageConversionException;
-import org.springframework.messaging.handler.annotation.Headers;
 import org.springframework.messaging.handler.annotation.Payload;
-import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.stereotype.Service;
-import org.wsd.app.avro.SensorEventAvro;
+import org.wsd.app.event.SensorEventAvro;
 import org.wsd.app.config.TopicConfiguration;
-import org.wsd.app.event.SensorEvent;
 
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
 
 @Log4j2
 @Service
@@ -59,8 +47,8 @@ public class ConsumerService {
             backoff = @Backoff(delay = 10, multiplier = 1.5, maxDelay = 2000)
     )
     @KafkaListener(topics = TopicConfiguration.SENSOR, groupId = "sensor-group", containerFactory = "kafkaListenerContainerFactory")
-    public void consumerGroup1(@Payload GenericRecord record) {
-        log.info("Consumed : " + record);
+    public void consumerGroup1(@Payload ConsumerRecord<UUID, SensorEventAvro> record) {
+        log.info("Consumed : " + record.value());
     }
 
 
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
@@ -1,3 +1,4 @@
+# Hibernate Level Cache Configuration
 spring.jpa.properties.javax.persistence.sharedCache.mode=ENABLE_SELECTIVE
 spring.jpa.properties.hibernate.cache.use_second_level_cache=true
 spring.jpa.properties.hibernate.cache.use_query_cache=true
@@ -13,17 +14,26 @@ spring.quartz.properties.org.quartz.jobStore.driverDelegateClass=org.quartz.impl
 spring.quartz.properties.org.quartz.jobStore.isClustered=true
 spring.quartz.properties.org.quartz.scheduler.instanceId=AUTO
 
-# Log
+# Log Configuration
 logging.file.name=logs/app.log
 
 #The filename pattern used to create log archives.
 logging.logback.rollingpolicy.file-name-pattern=logs/%d{yyyy-MM, aux}/app.%d{yyyy-MM-dd}.%i.log
 
 #The maximum size of log file before it is archived.
-logging.logback.rollingpolicy.max-file-size=1MB
+logging.logback.rollingpolicy.max-file-size=10MB
 
 #The maximum amount of size log archives can take before being deleted.
 logging.logback.rollingpolicy.total-size-cap=10GB
 
 #The maximum number of archive log files to keep (defaults to 7).
 logging.logback.rollingpolicy.max-history=10
+
+## root level
+#logging.level.root=error
+#
+## package level logging
+#logging.level.org.springframework.web=debug
+#logging.level.org.hibernate=error
+#logging.level.com.mkyong=error
+

Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"type": "record",`
`3`	`3`	`"name": "SensorEventAvro",`
`4`		`- "namespace": "org.wsd.app.avro",`
	`4`	`+ "namespace": "org.wsd.app.event",`
`5`	`5`	`"fields": [`
`6`	`6`	`{`
`7`	`7`	`"name": "id",`
`@@ -12,8 +12,9 @@`
`12`	`12`	`"type": "double"`
`13`	`13`	`},`
`14`	`14`	`{`
`15`		`- "name": "y",`
`16`		`- "type": "double"`
	`15`	`+ "name": "z",`
	`16`	`+ "type": "double",`
	`17`	`+ "default": "-1"`
`17`	`18`	`}`
`18`	`19`	`]`
`19`	`20`	`}`