[MODNOTES-295] Kafka producer foundation + Note create events

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,18 @@
       {
         "name": "NOTES_TYPES_DEFAULTS_LIMIT",
         "value": "25"
+      },
+      {
+        "name": "KAFKA_HOST",
+        "value": "kafka"
+      },
+      {
+        "name": "KAFKA_PORT",
+        "value": "9092"
+      },
+      {
+        "name": "ENV",
+        "value": "folio"
       }
     ]
   }

docker/.env

@@ -22,3 +22,15 @@ 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
+# Topics pre-created by kafka-topic-init (comma-separated tenants) and their partition count
+KAFKA_INIT_TENANTS=diku,test
+KAFKA_TOPIC_PARTITIONS=1
+
+

docker/README.md

@@ -34,6 +34,13 @@ 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`                 |
+| `KAFKA_INIT_TENANTS`      | Tenants to pre-create topics  | `diku,test`            |
+| `KAFKA_TOPIC_PARTITIONS`  | Partitions per topic          | `1`                    |
 
 ## 🚀 Services
 
@@ -54,6 +61,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`. Pre-created for the tenants in `KAFKA_INIT_TENANTS`; the broker also auto-creates the topic 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,8 @@ services:
     depends_on:
       - postgres
       - wiremock
+      - kafka
+      - kafka-topic-init
     deploy:
       replicas: ${MODULE_REPLICAS:-1}
       restart_policy:

docker/infra-docker-compose.yml

@@ -45,6 +45,55 @@ 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
+
+  kafka-topic-init:
+    image: apache/kafka:latest
+    depends_on:
+      - kafka
+    environment:
+      KAFKA_HOST: kafka
+      KAFKA_PORT: 9093
+      ENV: ${ENV}
+      TENANTS: ${KAFKA_INIT_TENANTS}
+      KAFKA_TOPIC_PARTITIONS: ${KAFKA_TOPIC_PARTITIONS}
+    volumes:
+      - ./kafka-init.sh:/usr/bin/kafka-init.sh
+    entrypoint: ["bash", "/usr/bin/kafka-init.sh"]
+    networks:
+      - mod-notes-local
+
 networks:
   mod-notes-local:
     driver: bridge

docker/kafka-init.sh

@@ -0,0 +1,34 @@
+#!/bin/bash
+set -e
+
+# Pre-creates the Note domain-event topics for mod-notes so they are visible in Kafka UI straight away
+KAFKA_BROKER="${KAFKA_HOST}:${KAFKA_PORT}"
+echo "Waiting for Kafka broker at $KAFKA_BROKER..."
+
+until /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server "$KAFKA_BROKER" > /dev/null 2>&1; do
+  echo "Kafka broker not ready yet, waiting..."
+  sleep 2
+done
+
+echo "Kafka broker is ready!"
+
+KAFKA_TOPICS_CMD="/opt/kafka/bin/kafka-topics.sh"
+
+# Comma-separated list of tenants (from the TENANTS env var), default "diku,test".
+IFS=',' read -ra TENANT_LIST <<< "${TENANTS:-diku,test}"
+
+for TENANT in "${TENANT_LIST[@]}"; do
+  TENANT_TRIMMED=$(echo "$TENANT" | xargs)
+  TOPIC="${ENV}.${TENANT_TRIMMED}.notes.note"
+  $KAFKA_TOPICS_CMD \
+    --create \
+    --bootstrap-server "$KAFKA_BROKER" \
+    --replication-factor 1 \
+    --partitions "${KAFKA_TOPIC_PARTITIONS}" \
+    --topic "$TOPIC" \
+    --if-not-exists
+  echo "Created topic: $TOPIC"
+done
+
+echo "Kafka topics created successfully."
+

pom.xml

@@ -78,6 +78,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 +160,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
+}

src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java

@@ -0,0 +1,58 @@
+package org.folio.notes.integration.kafka;
+
+import java.util.UUID;
+import lombok.extern.slf4j.Slf4j;
+import org.folio.notes.domain.dto.Note;
+import org.folio.notes.domain.event.DomainEvent;
+import org.folio.spring.FolioExecutionContext;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * Publishes Note {@link DomainEvent}s to the tenant-scoped Kafka topic.
+ *
+ * <p>The topic name follows the FOLIO convention {@code {env}.{tenant}.notes.note} (for example
+ * {@code folio.diku.notes.note}). The tenant is resolved at publish time via {@link FolioExecutionContext}
+ * and {@code env} defaults to {@code folio}.</p>
+ */
+@Slf4j
+@Component
+public class NoteEventProducer {
+
+  private final KafkaTemplate<UUID, DomainEvent<Note>> kafkaTemplate;
+  private final FolioExecutionContext context;
+  private final String env;
+  private final String noteTopic;
+
+  public NoteEventProducer(KafkaTemplate<UUID, DomainEvent<Note>> kafkaTemplate,
+                           FolioExecutionContext context,
+                           @Value("${folio.environment:folio}") String env,
+                           @Value("${folio.kafka.topics.note:notes.note}") String noteTopic) {
+    this.kafkaTemplate = kafkaTemplate;
+    this.context = context;
+    this.env = env;
+    this.noteTopic = noteTopic;
+  }
+
+  /**
+   * Publishes a Note domain event to the tenant-scoped topic.
+   *
+   * @param id    the Note id used as the Kafka record key
+   * @param event the domain event envelope to publish
+   */
+  public void publish(UUID id, DomainEvent<Note> event) {
+    var topic = topicName();
+    try {
+      log.debug("publish:: sending Note event [topic: {}, id: {}, type: {}]", topic, id, event.getType());
+      kafkaTemplate.send(topic, id, event);
+      log.info("publish:: Note event sent [topic: {}, id: {}, type: {}]", topic, id, event.getType());
+    } catch (Exception e) {
+      log.error("publish:: failed to send Note event [topic: {}, id: {}, type: {}]", topic, id, event.getType(), e);
+    }
+  }
+
+  private String topicName() {
+    return String.format("%s.%s.%s", env, context.getTenantId(), noteTopic);
+  }
+}