Merge branch 'refs/heads/feature/interceptor' into feature/transactio…

…nal-outbox

statistics-service-provider/src/main/java/de/sample/schulung/statistics/kafka/CustomJsonSerializer.java

@@ -0,0 +1,23 @@
+package de.sample.schulung.statistics.kafka;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.springframework.kafka.support.JacksonUtils;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+public class CustomJsonSerializer extends JsonSerializer<Object> {
+
+  private static ObjectMapper createCustomObjectMapper() {
+    final var result = JacksonUtils.enhancedObjectMapper();
+    result.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
+    result.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+    result.registerModule(new JavaTimeModule());
+    return result;
+  }
+
+  public CustomJsonSerializer() {
+    super(createCustomObjectMapper());
+  }
+}

statistics-service-provider/src/main/java/de/sample/schulung/statistics/kafka/CustomerEventRecord.java

@@ -1,10 +1,18 @@
 package de.sample.schulung.statistics.kafka;
 
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+
 import java.util.UUID;
 
 public record CustomerEventRecord(
+  @NotNull
+  @Pattern(regexp = "created|replaced|deleted")
   String eventType,
+  @NotNull
   UUID uuid,
+  @Valid
   CustomerRecord customer
 ) {
 }

statistics-service-provider/src/main/java/de/sample/schulung/statistics/kafka/CustomerRecord.java

@@ -1,9 +1,15 @@
 package de.sample.schulung.statistics.kafka;
 
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+
 import java.time.LocalDate;
 
 public record CustomerRecord(
+  @NotNull
   LocalDate birthdate,
+  @NotNull
+  @Pattern(regexp = "active|locked|disabled")
   String state
 ) {
 }

statistics-service-provider/src/main/resources/application.yml

@@ -18,6 +18,14 @@ spring:
       enabled: true
   kafka:
     bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
+    producer: #DTL
+      # key-serializer: org.apache.kafka.common.serialization.UUIDSerializer
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      # https://docs.spring.io/spring-kafka/reference/kafka/serdes.html#json-serde (Jackson)
+      value-serializer: de.sample.schulung.statistics.kafka.CustomJsonSerializer
+      properties:
+        # do not serialize the class name into the message
+        "[spring.json.add.type.headers]": false
     consumer:
       value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
       #de.sample.schulung.statistics.kafka.CustomJsonDeserializer