Merge pull request #3 from ralf-ueberfuhr-ars/feature/consumer

Add Kafka producer.

README.md

@@ -29,3 +29,30 @@ java -jar target/<file>.jar
 # oder
 mvn spring-boot:run
 ```
+
+# Customer Events über Kafka
+
+Producer und Consumer benötigen eine wohldefinierte Schnittstelle. Diese wird hier dokumentiert.
+
+- Name des Topics: `customer-events`
+- Aufteilung der Nachrichten (Partitionierung) nach der UUID des Kunden 
+  (Events für denselben Kunden müssen in derselben Partition landen)
+  - Message Key: `UUID`
+  - Format: `String`
+- Customer-Daten sowie die Art des Events als Payload
+  - Message Payload: `CustomerEventRecord`
+  - Format: `JSON`
+
+Hier ein Beispiel für den Payload:
+
+```json
+{
+  "event_type": "created|replaced|deleted",
+  "uuid": "12345",
+  "customer": {
+    "name": "Tom Mayer",
+    "birthdate": "2002-10-05",
+    "state": "active|locked|disabled"
+  }
+}
+```

account-service-provider/pom.xml

@@ -55,6 +55,15 @@
       <artifactId>h2</artifactId>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka-test</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-test</artifactId>

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

@@ -0,0 +1,23 @@
+package de.sample.schulung.accounts.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());
+  }
+}

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

@@ -0,0 +1,10 @@
+package de.sample.schulung.accounts.kafka;
+
+import java.util.UUID;
+
+public record CustomerEventRecord(
+  String eventType,
+  UUID uuid,
+  CustomerRecord customer
+) {
+}

account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventsProducer.java

@@ -0,0 +1,42 @@
+package de.sample.schulung.accounts.kafka;
+
+import de.sample.schulung.accounts.domain.events.CustomerCreatedEvent;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.event.EventListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.UUID;
+
+@Component
+@RequiredArgsConstructor
+public class CustomerEventsProducer {
+
+  private final KafkaTemplate<UUID, Object> kafkaTemplate;
+
+  @EventListener
+  public void handleCustomerCreatedEvent(CustomerCreatedEvent event) {
+    // map event to record
+    var customer = event.customer();
+    var payload = new CustomerEventRecord(
+      "created",
+      customer.getUuid(),
+      new CustomerRecord(
+        customer.getName(),
+        customer.getDateOfBirth(),
+        switch(customer.getState()) {
+          case ACTIVE -> "active";
+          case LOCKED -> "locked";
+          case DISABLED -> "disabled";
+        }
+      )
+    );
+    // send message
+    kafkaTemplate.send(
+      KafkaConstants.CUSTOMER_EVENTS_TOPIC,
+      event.customer().getUuid(),
+      payload
+    );
+  }
+
+}

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

@@ -0,0 +1,10 @@
+package de.sample.schulung.accounts.kafka;
+
+import java.time.LocalDate;
+
+public record CustomerRecord(
+  String name,
+  LocalDate birthdate,
+  String state
+) {
+}

account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/KafkaConfiguration.java

@@ -0,0 +1,34 @@
+package de.sample.schulung.accounts.kafka;
+
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.support.JacksonUtils;
+
+@Configuration
+public class KafkaConfiguration {
+
+  // Topic beim Start der Anwendung erzeugen
+  // -> nur für lokale Tests, NICHT für Produktion
+  // siehe application.yml: spring.kafka.admin.auto-create
+  @Bean
+  public NewTopic customerEventsTopic() {
+    return TopicBuilder
+      .name(KafkaConstants.CUSTOMER_EVENTS_TOPIC)
+      // .partitions(3)
+      // .replicas(3)
+      .build();
+  }
+
+  @EventListener(ContextRefreshedEvent.class)
+  public void configureJsonSerializer() {
+    JacksonUtils
+      .enhancedObjectMapper()
+      .setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
+  }
+
+}

account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/KafkaConstants.java

@@ -0,0 +1,10 @@
+package de.sample.schulung.accounts.kafka;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class KafkaConstants {
+
+  public final String CUSTOMER_EVENTS_TOPIC = "customer-events";
+
+}

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

@@ -16,6 +16,17 @@ spring:
     console:
       path: /db
       enabled: true
+  kafka:
+    admin:
+      auto-create: true
+    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.UUIDSerializer
+      # https://docs.spring.io/spring-kafka/reference/kafka/serdes.html#json-serde (Jackson)
+      value-serializer: de.sample.schulung.accounts.kafka.CustomJsonSerializer
+      properties:
+        # do not serialize the class name into the message
+        "[spring.json.add.type.headers]": false
 application:
   customers:
     initialization:

account-service-provider/src/test/java/de/sample/schulung/accounts/AccountsApiTests.java

@@ -1,5 +1,6 @@
 package de.sample.schulung.accounts;
 
+import de.sample.schulung.accounts.kafka.AutoConfigureKafkaTemplateMock;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
@@ -18,6 +19,7 @@
 @SpringBootTest
 @AutoConfigureMockMvc
 @AutoConfigureTestDatabase
+@AutoConfigureKafkaTemplateMock
 class AccountsApiTests {
 
   @Autowired

account-service-provider/src/test/java/de/sample/schulung/accounts/IndexPageTests.java

@@ -1,5 +1,6 @@
 package de.sample.schulung.accounts;
 
+import de.sample.schulung.accounts.kafka.AutoConfigureKafkaTemplateMock;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
@@ -15,6 +16,7 @@
 @SpringBootTest
 @AutoConfigureMockMvc
 @AutoConfigureTestDatabase
+@AutoConfigureKafkaTemplateMock
 public class IndexPageTests {
 
   @Autowired

account-service-provider/src/test/java/de/sample/schulung/accounts/boundary/AccountsBoundaryTests.java

@@ -2,6 +2,7 @@
 
 import de.sample.schulung.accounts.domain.CustomersService;
 import de.sample.schulung.accounts.domain.NotFoundException;
+import de.sample.schulung.accounts.kafka.AutoConfigureKafkaTemplateMock;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
@@ -28,6 +29,7 @@
 )
 @AutoConfigureMockMvc
 @AutoConfigureTestDatabase
+@AutoConfigureKafkaTemplateMock
 public class AccountsBoundaryTests {
 
   @Autowired

account-service-provider/src/test/java/de/sample/schulung/accounts/domain/CustomersInitializerTests.java

@@ -1,5 +1,6 @@
 package de.sample.schulung.accounts.domain;
 
+import de.sample.schulung.accounts.kafka.AutoConfigureKafkaTemplateMock;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -15,6 +16,7 @@
   }
 )
 @AutoConfigureTestDatabase
+@AutoConfigureKafkaTemplateMock
 public class CustomersInitializerTests {
 
   @MockBean

account-service-provider/src/test/java/de/sample/schulung/accounts/domain/CustomersServiceTest.java

@@ -1,5 +1,6 @@
 package de.sample.schulung.accounts.domain;
 
+import de.sample.schulung.accounts.kafka.AutoConfigureKafkaTemplateMock;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
@@ -12,6 +13,7 @@
 
 @SpringBootTest
 @AutoConfigureTestDatabase
+@AutoConfigureKafkaTemplateMock
 public class CustomersServiceTest {
 
   @Autowired

account-service-provider/src/test/java/de/sample/schulung/accounts/kafka/AutoConfigureKafkaTemplateMock.java

@@ -0,0 +1,26 @@
+package de.sample.schulung.accounts.kafka;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.kafka.core.KafkaTemplate;
+
+import java.lang.annotation.*;
+
+/**
+ * Auto-configures a {@link KafkaTemplate} mock in the test context.
+ * You can get the mock injected by simply using
+ * <pre>
+ * \u0040Autowired
+ * KafkaTemplate&lt;String, CustomerDto&gt; templateMock;
+ * </pre>
+ */
+@Documented
+@Inherited
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+@EnableAutoConfiguration(exclude = KafkaAutoConfiguration.class)
+@MockBean(KafkaTemplate.class)
+public @interface AutoConfigureKafkaTemplateMock {
+
+}