[MODNOTES-295] Address comments

Author: Saba-Zedginidze-EPAM

descriptors/ModuleDescriptor-template.json

@@ -322,9 +322,40 @@
         "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"
+        "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

@@ -29,8 +29,5 @@ 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

@@ -39,8 +39,6 @@ Configuration is managed via the `.env` file in this directory.
 | `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
 
@@ -66,7 +64,7 @@ Configuration is managed via the `.env` file in this directory.
 - **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.
+- **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

docker/app-docker-compose.yml

@@ -30,7 +30,6 @@ services:
       - postgres
       - wiremock
       - kafka
-      - kafka-topic-init
     deploy:
       replicas: ${MODULE_REPLICAS:-1}
       restart_policy:

docker/infra-docker-compose.yml

@@ -78,21 +78,6 @@ services:
     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:

docker/kafka-init.sh

@@ -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>

pom.xml

@@ -1,38 +1,47 @@
 package org.folio.notes.integration.kafka;
 
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
 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.folio.spring.tools.kafka.FolioKafkaProperties;
+import org.folio.spring.tools.kafka.KafkaUtils;
 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>
+ * {@code folio.diku.notes.note}), resolved via {@link KafkaUtils#getTenantTopicName(String, String)} the same way the
+ * topic is created on tenant enable. The tenant is resolved at publish time via {@link FolioExecutionContext}.</p>
+ *
+ * <p>Every message carries headers ({@code X-Okapi-Tenant}, {@code eventType}, {@code domain}) so consumers can
+ * filter by tenant, event type or Note domain without deserializing the JSON payload.</p>
  */
 @Slf4j
 @Component
 public class NoteEventProducer {
 
+  private static final String EVENT_TYPE_HEADER = "domain-event-type";
+
   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) {
+                           FolioKafkaProperties kafkaProperties) {
     this.kafkaTemplate = kafkaTemplate;
     this.context = context;
-    this.env = env;
-    this.noteTopic = noteTopic;
+    this.noteTopic = kafkaProperties.getTopics().getFirst().getName();
   }
 
   /**
@@ -42,17 +51,26 @@ public NoteEventProducer(KafkaTemplate<UUID, DomainEvent<Note>> kafkaTemplate,
    * @param event the domain event envelope to publish
    */
   public void publish(UUID id, DomainEvent<Note> event) {
-    var topic = topicName();
+    var topic = KafkaUtils.getTenantTopicName(noteTopic, context.getTenantId());
     try {
       log.debug("publish:: sending Note event [topic: {}, id: {}, type: {}]", topic, id, event.getType());
-      kafkaTemplate.send(topic, id, event);
+      var producerRecord = new ProducerRecord<>(topic, null, id, event, buildHeaders(event));
+      kafkaTemplate.send(producerRecord);
       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);
+  private List<Header> buildHeaders(DomainEvent<Note> event) {
+    var headers = new ArrayList<Header>();
+    headers.add(header(EVENT_TYPE_HEADER, event.getType().name()));
+    context.getAllHeaders().forEach((key, value) -> headers.add(header(key, value.iterator().next())));
+    return headers;
   }
+
+  private Header header(String key, String value) {
+    return new RecordHeader(key, value == null ? null : value.getBytes(StandardCharsets.UTF_8));
+  }
+
 }

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

@@ -1,29 +1,42 @@
 package org.folio.notes.service.impl;
 
+import lombok.extern.slf4j.Slf4j;
 import org.folio.notes.service.NoteTypesService;
 import org.folio.spring.FolioExecutionContext;
 import org.folio.spring.liquibase.FolioSpringLiquibase;
 import org.folio.spring.service.TenantService;
+import org.folio.spring.tools.kafka.KafkaAdminService;
+import org.folio.tenant.domain.dto.TenantAttributes;
 import org.springframework.context.annotation.Primary;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Service;
 
 @Service
 @Primary
+@Slf4j
 public class NoteTenantService extends TenantService {
 
   private final NoteTypesService noteTypesService;
+  private final KafkaAdminService kafkaAdminService;
 
   public NoteTenantService(JdbcTemplate jdbcTemplate,
                            FolioExecutionContext context,
                            FolioSpringLiquibase folioSpringLiquibase,
-                           NoteTypesService noteTypesService) {
+                           NoteTypesService noteTypesService,
+                           KafkaAdminService kafkaAdminService) {
     super(jdbcTemplate, context, folioSpringLiquibase);
     this.noteTypesService = noteTypesService;
+    this.kafkaAdminService = kafkaAdminService;
   }
 
   @Override
   public void loadReferenceData() {
     noteTypesService.populateDefaultType();
   }
+
+  @Override
+  protected void afterTenantUpdate(TenantAttributes tenantAttributes) {
+    super.afterTenantUpdate(tenantAttributes);
+    kafkaAdminService.createTopics(context.getTenantId());
+  }
 }

src/main/java/org/folio/notes/service/impl/NoteTenantService.java

@@ -133,9 +133,9 @@ public Note createNote(Note note) {
     NoteEntity entity = saveNote(note, dto -> initNewEntity(notesMapper.toEntity(dto)));
     log.info("createNote:: created note by title: {}, domain: {}, type: {}",
       note.getTitle(), note.getDomain(), note.getType());
-    Note newSnapshot = notesMapper.toDto(entity);
-    domainEventPublisherService.publishNoteCreatedEvent(newSnapshot);
-    return newSnapshot;
+    Note createdNote = notesMapper.toDto(entity);
+    domainEventPublisherService.publishNoteCreatedEvent(createdNote);
+    return createdNote;
   }
 
   @Transactional