[MODNOTES-295] Kafka producer foundation + Note create events (#373)

Author: Saba-Zedginidze-EPAM

README.md

@@ -39,7 +39,7 @@ quite do yet.
 
 ### Compiling
 
-Compile with 
+Compile with
 ```shell
 mvn clean install
 ```
@@ -71,6 +71,10 @@ requires and provides, the permissions, and the additional module metadata.
 
 Use `DB_HOST`, `DB_PORT`, `DB_DATABASE`, `DB_USERNAME`, `DB_PASSWORD` to configure the PostgreSQL database.
 
+Use `KAFKA_HOST` (default `localhost`) and `KAFKA_PORT` (default `9092`) to configure the Kafka broker the module
+publishes domain events to. `ENV` (default `folio`) is the environment prefix used when building the tenant-scoped topic
+name `{ENV}.{tenant}.notes.note`.
+
 `NOTES_TYPES_DEFAULTS_LIMIT` defaults to 25.
 
 `MAX_RECORDS_COUNT` defaults to 1000.

descriptors/ModuleDescriptor-template.json

@@ -313,6 +313,49 @@
       {
         "name": "NOTES_TYPES_DEFAULTS_LIMIT",
         "value": "25"
+      },
+      {
+        "name": "KAFKA_HOST",
+        "value": "kafka"
+      },
+      {
+        "name": "KAFKA_PORT",
+        "value": "9092"
+      },
+      {
+        "name": "KAFKA_SECURITY_PROTOCOL",
+        "value": "PLAINTEXT",
+        "description": "Kafka security protocol used to communicate with brokers (SSL or PLAINTEXT)"
+      },
+      {
+        "name": "KAFKA_SSL_KEYSTORE_LOCATION",
+        "description": "The location of the Kafka key store file. This is optional for client and can be used for two-way authentication for client."
+      },
+      {
+        "name": "KAFKA_SSL_KEYSTORE_PASSWORD",
+        "description": "The store password for the Kafka key store file. This is optional for client and only needed if 'ssl.keystore.location' is configured."
+      },
+      {
+        "name": "KAFKA_SSL_TRUSTSTORE_LOCATION",
+        "description": "The location of the Kafka trust store file."
+      },
+      {
+        "name": "KAFKA_SSL_TRUSTSTORE_PASSWORD",
+        "description": "The password for the Kafka trust store file. If a password is not set, trust store file configured will still be used, but integrity checking is disabled."
+      },
+      {
+        "name": "KAFKA_NOTE_TOPIC_PARTITIONS",
+        "value": "1",
+        "description": "Number of partitions for the `notes.note` topic."
+      },
+      {
+        "name": "KAFKA_NOTE_TOPIC_REPLICATION_FACTOR",
+        "description": "Replication factor for the `notes.note` topic. When not set, the broker default is used."
+      },
+      {
+        "name": "ENV",
+        "value": "folio",
+        "description": "The logical name of the deployment, must be unique across all environments using the same shared Kafka/Elasticsearch clusters, `a-z (any case)`, `0-9`, `-`, `_` symbols only allowed"
       }
     ]
   }

docker/.env

@@ -22,3 +22,12 @@ PGADMIN_DEFAULT_PASSWORD=admin
 WIREMOCK_PORT=9130
 OKAPI_URL=http://wiremock:8080
 
+# Kafka configuration
+ENV=folio
+KAFKA_HOST=kafka
+KAFKA_PORT=9093
+KAFKA_EXTERNAL_PORT=9092
+# Kafka UI (web) port -> http://localhost:8090
+KAFKA_UI_PORT=8090
+
+

docker/README.md

@@ -34,6 +34,11 @@ Configuration is managed via the `.env` file in this directory.
 | `PGADMIN_PORT`            | PgAdmin port                  | `5050`                 |
 | `WIREMOCK_PORT`           | WireMock (Okapi mock) port    | `9130`                 |
 | `OKAPI_URL`               | Okapi URL for the module      | `http://wiremock:8080` |
+| `ENV`                     | Kafka topic prefix            | `folio`                |
+| `KAFKA_HOST`              | Broker host (in-cluster)      | `kafka`                |
+| `KAFKA_PORT`              | Broker port (in-cluster)      | `9093`                 |
+| `KAFKA_EXTERNAL_PORT`     | Broker port from host/IDE     | `9092`                 |
+| `KAFKA_UI_PORT`           | Kafka UI web port             | `8090`                 |
 
 ## 🚀 Services
 
@@ -54,6 +59,17 @@ Configuration is managed via the `.env` file in this directory.
 - **Access**: http://localhost:9130 (configurable via `WIREMOCK_PORT`)
 - **Mappings**: Located in `src/test/resources/mappings`
 
+### Kafka
+- **Purpose**: Message broker the module publishes Note domain events to
+- **Image**: `apache/kafka-native` (KRaft mode, no Zookeeper)
+- **Access (host/IDE)**: `localhost:9092` (configurable via `KAFKA_EXTERNAL_PORT`)
+- **Access (in-cluster)**: `kafka:9093`
+- **Topics**: `{ENV}.{tenant}.notes.note` — e.g. `folio.diku.notes.note`. The module creates the topic when a tenant is enabled (via the `/_/tenant` API); the broker also auto-creates it on first publish.
+
+### Kafka UI
+- **Purpose**: Web UI to browse topics and inspect published event messages
+- **Access**: http://localhost:8090 (configurable via `KAFKA_UI_PORT`)
+
 ## 📖 Usage
 
 > **Note**: All commands in this guide assume you are in the `docker/` directory. If you're at the project root, run `cd docker` first.

docker/app-docker-compose.yml

@@ -14,6 +14,9 @@ services:
       - "${MODULE_PORT}:8081"
       - "${DEBUG_PORT}:5005"
     environment:
+      ENV: ${ENV}
+      KAFKA_HOST: ${KAFKA_HOST}
+      KAFKA_PORT: ${KAFKA_PORT}
       DB_HOST: ${DB_HOST}
       DB_PORT: ${DB_PORT}
       DB_DATABASE: ${DB_DATABASE}
@@ -26,6 +29,7 @@ services:
     depends_on:
       - postgres
       - wiremock
+      - kafka
     deploy:
       replicas: ${MODULE_REPLICAS:-1}
       restart_policy:

docker/infra-docker-compose.yml

@@ -45,6 +45,40 @@ services:
     networks:
       - mod-notes-local
 
+  kafka:
+    image: apache/kafka-native
+    networks:
+      - mod-notes-local
+    ports:
+      - ${KAFKA_EXTERNAL_PORT}:${KAFKA_EXTERNAL_PORT}
+      - ${KAFKA_PORT}:9093
+    environment:
+      KAFKA_LISTENERS: CONTROLLER://localhost:9091,HOST://0.0.0.0:${KAFKA_EXTERNAL_PORT},DOCKER://0.0.0.0:9093
+      KAFKA_ADVERTISED_LISTENERS: HOST://localhost:${KAFKA_EXTERNAL_PORT},DOCKER://kafka:9093
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,DOCKER:PLAINTEXT,HOST:PLAINTEXT
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9091
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_INTER_BROKER_LISTENER_NAME: DOCKER
+      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    networks:
+      - mod-notes-local
+    ports:
+      - ${KAFKA_UI_PORT}:8080
+    environment:
+      DYNAMIC_CONFIG_ENABLED: 'true'
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9093
+    depends_on:
+      - kafka
+
+
 networks:
   mod-notes-local:
     driver: bridge

pom.xml

@@ -27,6 +27,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
     <folio-spring-support.version>10.0.0</folio-spring-support.version>
+    <folio-service-tools.version>6.0.0</folio-service-tools.version>
     <mapstruct.version>1.6.3</mapstruct.version>
     <jsoup.version>1.22.2</jsoup.version>
     <lombok.version>1.18.44</lombok.version>
@@ -64,6 +65,11 @@
       <artifactId>folio-spring-cql</artifactId>
       <version>${folio-spring-support.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.folio</groupId>
+      <artifactId>folio-service-tools-spring-dev</artifactId>
+      <version>${folio-service-tools.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-cache</artifactId>
@@ -78,6 +84,16 @@
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-validation</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-kafka</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>org.springframework.boot</groupId>
+          <artifactId>spring-boot-starter-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-actuator</artifactId>
@@ -150,6 +166,17 @@
       <version>${folio-spring-support.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-kafka-test</artifactId>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.springframework.boot</groupId>
+          <artifactId>spring-boot-starter-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-webmvc-test</artifactId>

src/main/java/org/folio/notes/config/KafkaConfiguration.java

@@ -0,0 +1,33 @@
+package org.folio.notes.config;
+
+import java.util.UUID;
+import org.folio.notes.domain.dto.Note;
+import org.folio.notes.domain.event.DomainEvent;
+import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+/**
+ * Kafka producer configuration for Note domain events.
+ *
+ * <p>Defines the {@link ProducerFactory} and {@link KafkaTemplate} used to publish {@link DomainEvent} envelopes
+ * keyed by the Note id. Serializers are driven by the {@code spring.kafka.producer} configuration
+ * ({@code UUIDSerializer} for the key, {@code JacksonJsonSerializer} for the value).</p>
+ */
+@Configuration
+public class KafkaConfiguration {
+
+  @Bean
+  public ProducerFactory<UUID, DomainEvent<Note>> noteEventProducerFactory(KafkaProperties kafkaProperties) {
+    return new DefaultKafkaProducerFactory<>(kafkaProperties.buildProducerProperties());
+  }
+
+  @Bean
+  public KafkaTemplate<UUID, DomainEvent<Note>> noteEventKafkaTemplate(
+    ProducerFactory<UUID, DomainEvent<Note>> noteEventProducerFactory) {
+    return new KafkaTemplate<>(noteEventProducerFactory);
+  }
+}

src/main/java/org/folio/notes/domain/event/DomainEvent.java

@@ -0,0 +1,59 @@
+package org.folio.notes.domain.event;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.UUID;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Envelope for a mod-notes domain event.
+ *
+ * @param <T> the payload type carried in the {@code old} and {@code new} fields (a full entity snapshot).
+ */
+@Data
+@NoArgsConstructor
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class DomainEvent<T> {
+
+  private UUID id;
+  @JsonProperty("old")
+  private T oldEntity;
+  @JsonProperty("new")
+  private T newEntity;
+  private DomainEventType type;
+  private String tenant;
+  private String ts;
+
+  @JsonCreator
+  public DomainEvent(@JsonProperty("id") UUID id,
+                     @JsonProperty("old") T oldEntity,
+                     @JsonProperty("new") T newEntity,
+                     @JsonProperty("type") DomainEventType type,
+                     @JsonProperty("tenant") String tenant,
+                     @JsonProperty("ts") String ts) {
+    this.id = id;
+    this.oldEntity = oldEntity;
+    this.newEntity = newEntity;
+    this.type = type;
+    this.tenant = tenant;
+    this.ts = ts;
+  }
+
+  public static <T> DomainEvent<T> createEvent(UUID id, T newEntity, String tenant) {
+    return new DomainEvent<>(id, null, newEntity, DomainEventType.CREATE, tenant, currentTs());
+  }
+
+  public static <T> DomainEvent<T> updateEvent(UUID id, T oldEntity, T newEntity, String tenant) {
+    return new DomainEvent<>(id, oldEntity, newEntity, DomainEventType.UPDATE, tenant, currentTs());
+  }
+
+  public static <T> DomainEvent<T> deleteEvent(UUID id, T oldEntity, String tenant) {
+    return new DomainEvent<>(id, oldEntity, null, DomainEventType.DELETE, tenant, currentTs());
+  }
+
+  private static String currentTs() {
+    return String.valueOf(System.currentTimeMillis());
+  }
+}

src/main/java/org/folio/notes/domain/event/DomainEventType.java

@@ -0,0 +1,5 @@
+package org.folio.notes.domain.event;
+
+public enum DomainEventType {
+  CREATE, UPDATE, DELETE
+}