From e6652549b6b45a58ba6117ab82a95e668fd9c5ef Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 12:51:27 +0400 Subject: [PATCH 01/12] [MODNOTES-295] Update pom and properties to include new deps --- pom.xml | 21 +++++++++++++++++++++ src/main/resources/application.yaml | 23 ++++++++++++++++++++++- 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 174c78f7..05cfc563 100644 --- a/pom.xml +++ b/pom.xml @@ -78,6 +78,16 @@ org.springframework.boot spring-boot-starter-validation + + org.springframework.boot + spring-boot-starter-kafka + + + org.springframework.boot + spring-boot-starter-logging + + + org.springframework.boot spring-boot-starter-actuator @@ -150,6 +160,17 @@ ${folio-spring-support.version} test + + org.springframework.boot + spring-boot-starter-kafka-test + test + + + org.springframework.boot + spring-boot-starter-logging + + + org.springframework.boot spring-boot-starter-webmvc-test diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 5f959253..6050dd93 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -1,5 +1,6 @@ # Module properties folio: + environment: ${ENV:folio} exchange: enabled: true tenant: @@ -47,11 +48,31 @@ folio: - target response: limit: ${MAX_RECORDS_COUNT:1000} + kafka: + topics: + note: notes.note # Spring properties spring: application: name: mod-notes + kafka: + bootstrap-servers: ${KAFKA_HOST:localhost}:${KAFKA_PORT:9092} + producer: + client-id: mod-notes + key-serializer: org.apache.kafka.common.serialization.UUIDSerializer + value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer + acks: all + retries: 3 + properties: + max.block.ms: ${KAFKA_PRODUCER_MAX_BLOCK_MS:10000} + security: + protocol: ${KAFKA_SECURITY_PROTOCOL:PLAINTEXT} + ssl: + key-store-password: ${KAFKA_SSL_KEYSTORE_PASSWORD:} + key-store-location: ${KAFKA_SSL_KEYSTORE_LOCATION:} + trust-store-password: ${KAFKA_SSL_TRUSTSTORE_PASSWORD:} + trust-store-location: ${KAFKA_SSL_TRUSTSTORE_LOCATION:} sql: init: continue-on-error: true @@ -94,4 +115,4 @@ server: logging: level: - org.folio.spring.filter.IncomingRequestLoggingFilter: DEBUG \ No newline at end of file + org.folio.spring.filter.IncomingRequestLoggingFilter: DEBUG From 2249419af086c0b97946e49d5c2f733a30924ec9 Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 12:51:51 +0400 Subject: [PATCH 02/12] [MODNOTES-295] Create kafka configuration and producer --- .../notes/config/KafkaConfiguration.java | 33 +++++++++++ .../integration/kafka/NoteEventProducer.java | 58 +++++++++++++++++++ 2 files changed, 91 insertions(+) create mode 100644 src/main/java/org/folio/notes/config/KafkaConfiguration.java create mode 100644 src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java diff --git a/src/main/java/org/folio/notes/config/KafkaConfiguration.java b/src/main/java/org/folio/notes/config/KafkaConfiguration.java new file mode 100644 index 00000000..841c9b56 --- /dev/null +++ b/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. + * + *

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).

+ */ +@Configuration +public class KafkaConfiguration { + + @Bean + public ProducerFactory> noteEventProducerFactory(KafkaProperties kafkaProperties) { + return new DefaultKafkaProducerFactory<>(kafkaProperties.buildProducerProperties()); + } + + @Bean + public KafkaTemplate> noteEventKafkaTemplate( + ProducerFactory> noteEventProducerFactory) { + return new KafkaTemplate<>(noteEventProducerFactory); + } +} diff --git a/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java new file mode 100644 index 00000000..29ac3b1e --- /dev/null +++ b/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. + * + *

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}.

+ */ +@Slf4j +@Component +public class NoteEventProducer { + + private final KafkaTemplate> kafkaTemplate; + private final FolioExecutionContext context; + private final String env; + private final String noteTopic; + + public NoteEventProducer(KafkaTemplate> 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 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); + } +} From b6aa62696ed7e10a9ec5b1372033b358e4780197 Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 12:52:24 +0400 Subject: [PATCH 03/12] [MODNOTES-295] Create publisher service for domain events --- .../folio/notes/domain/event/DomainEvent.java | 59 ++++++++++++++++++ .../notes/domain/event/DomainEventType.java | 5 ++ .../service/DomainEventPublisherService.java | 60 +++++++++++++++++++ 3 files changed, 124 insertions(+) create mode 100644 src/main/java/org/folio/notes/domain/event/DomainEvent.java create mode 100644 src/main/java/org/folio/notes/domain/event/DomainEventType.java create mode 100644 src/main/java/org/folio/notes/service/DomainEventPublisherService.java diff --git a/src/main/java/org/folio/notes/domain/event/DomainEvent.java b/src/main/java/org/folio/notes/domain/event/DomainEvent.java new file mode 100644 index 00000000..1e8a9d2e --- /dev/null +++ b/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 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 { + + 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 DomainEvent createEvent(UUID id, T newEntity, String tenant) { + return new DomainEvent<>(id, null, newEntity, DomainEventType.CREATE, tenant, currentTs()); + } + + public static DomainEvent updateEvent(UUID id, T oldEntity, T newEntity, String tenant) { + return new DomainEvent<>(id, oldEntity, newEntity, DomainEventType.UPDATE, tenant, currentTs()); + } + + public static DomainEvent 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()); + } +} diff --git a/src/main/java/org/folio/notes/domain/event/DomainEventType.java b/src/main/java/org/folio/notes/domain/event/DomainEventType.java new file mode 100644 index 00000000..cb56086d --- /dev/null +++ b/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 +} diff --git a/src/main/java/org/folio/notes/service/DomainEventPublisherService.java b/src/main/java/org/folio/notes/service/DomainEventPublisherService.java new file mode 100644 index 00000000..1ecc2ee7 --- /dev/null +++ b/src/main/java/org/folio/notes/service/DomainEventPublisherService.java @@ -0,0 +1,60 @@ +package org.folio.notes.service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.folio.notes.domain.dto.Note; +import org.folio.notes.domain.event.DomainEvent; +import org.folio.notes.integration.kafka.NoteEventProducer; +import org.folio.spring.FolioExecutionContext; +import org.springframework.stereotype.Service; + +/** + * Builds {@link DomainEvent} envelopes for Note changes and hands them to the {@link NoteEventProducer}. + * + *

Callers use the event-specific methods ({@link #publishNoteCreatedEvent(Note)}, + * {@link #publishNoteUpdatedEvent(Note, Note)}, {@link #publishNoteDeletedEvent(Note)}) and only need to provide the + * relevant Note snapshot(s); this service takes care of resolving the current tenant via {@link FolioExecutionContext}, + * choosing the Kafka record key and populating the {@code old}/{@code new} envelope fields for each event type.

+ */ +@Slf4j +@Service +@RequiredArgsConstructor +public class DomainEventPublisherService { + + private final NoteEventProducer noteEventProducer; + private final FolioExecutionContext context; + + /** + * Publishes a {@code CREATE} Note event carrying the new snapshot. + * + * @param newNote the newly-created Note snapshot + */ + public void publishNoteCreatedEvent(Note newNote) { + publish(DomainEvent.createEvent(newNote.getId(), newNote, context.getTenantId())); + } + + /** + * Publishes an {@code UPDATE} Note event carrying both the pre- and post-change snapshots. + * + * @param oldNote the pre-change Note snapshot + * @param newNote the post-change Note snapshot + */ + public void publishNoteUpdatedEvent(Note oldNote, Note newNote) { + publish(DomainEvent.updateEvent(newNote.getId(), oldNote, newNote, context.getTenantId())); + } + + /** + * Publishes a {@code DELETE} Note event carrying the pre-delete snapshot. + * + * @param oldNote the pre-delete Note snapshot + */ + public void publishNoteDeletedEvent(Note oldNote) { + publish(DomainEvent.deleteEvent(oldNote.getId(), oldNote, context.getTenantId())); + } + + private void publish(DomainEvent event) { + log.debug("publish:: publishing Note event [id: {}, type: {}, tenant: {}]", + event.getId(), event.getType(), event.getTenant()); + noteEventProducer.publish(event.getId(), event); + } +} From 829a2a0fbce951f757e1045aa41e26b17e7d27ba Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 12:52:40 +0400 Subject: [PATCH 04/12] [MODNOTES-295] Update servce to send out events --- .../java/org/folio/notes/service/impl/NotesServiceImpl.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.java b/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.java index 8de228eb..8b4a0788 100644 --- a/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.java +++ b/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.java @@ -33,6 +33,7 @@ import org.folio.notes.domain.repository.NoteRepository; import org.folio.notes.domain.repository.NoteTypesRepository; import org.folio.notes.exception.NoteNotFoundException; +import org.folio.notes.service.DomainEventPublisherService; import org.folio.notes.service.NotesService; import org.folio.notes.util.HtmlSanitizer; import org.folio.spring.data.OffsetRequest; @@ -76,6 +77,7 @@ public class NotesServiceImpl implements NotesService { private final NotesMapper notesMapper; private final NoteCollectionMapper noteCollectionMapper; private final HtmlSanitizer sanitizer; + private final DomainEventPublisherService domainEventPublisherService; @Value("${folio.notes.response.limit}") private Integer responseLimit; @@ -131,7 +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()); - return notesMapper.toDto(entity); + Note newSnapshot = notesMapper.toDto(entity); + domainEventPublisherService.publishNoteCreatedEvent(newSnapshot); + return newSnapshot; } @Transactional From 66c407933179075c764336f66131c0a114b1e91a Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 12:53:36 +0400 Subject: [PATCH 05/12] [MODNOTES-295] Create new unit tests --- .../notes/domain/event/DomainEventTest.java | 67 ++++++++++++++ .../DomainEventPublisherServiceTest.java | 87 +++++++++++++++++++ .../org/folio/notes/support/TestApiBase.java | 2 + 3 files changed, 156 insertions(+) create mode 100644 src/test/java/org/folio/notes/domain/event/DomainEventTest.java create mode 100644 src/test/java/org/folio/notes/service/DomainEventPublisherServiceTest.java diff --git a/src/test/java/org/folio/notes/domain/event/DomainEventTest.java b/src/test/java/org/folio/notes/domain/event/DomainEventTest.java new file mode 100644 index 00000000..2a38acbf --- /dev/null +++ b/src/test/java/org/folio/notes/domain/event/DomainEventTest.java @@ -0,0 +1,67 @@ +package org.folio.notes.domain.event; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.UUID; +import org.folio.notes.domain.dto.Note; +import org.folio.spring.testing.type.UnitTest; +import org.junit.jupiter.api.Test; +import tools.jackson.databind.json.JsonMapper; + +@UnitTest +class DomainEventTest { + + private static final JsonMapper MAPPER = JsonMapper.builder().build(); + + @Test + void createEvent_setsCreateTypeAndOmitsOld() { + var id = UUID.randomUUID(); + var note = new Note().title("t").domain("orders").typeId(UUID.randomUUID()); + + var event = DomainEvent.createEvent(id, note, "diku"); + + assertEquals(id, event.getId()); + assertEquals(DomainEventType.CREATE, event.getType()); + assertEquals("diku", event.getTenant()); + assertNull(event.getOldEntity()); + assertEquals(note, event.getNewEntity()); + assertNotNull(event.getTs()); + + var json = MAPPER.writeValueAsString(event); + assertTrue(json.contains("\"type\":\"CREATE\"")); + assertTrue(json.contains("\"new\"")); + assertFalse(json.contains("\"old\""), "NON_NULL must omit the null 'old' field"); + } + + @Test + void updateEvent_carriesOldAndNew() { + var id = UUID.randomUUID(); + var oldNote = new Note().title("old"); + var newNote = new Note().title("new"); + + var event = DomainEvent.updateEvent(id, oldNote, newNote, "diku"); + + assertEquals(DomainEventType.UPDATE, event.getType()); + assertEquals(oldNote, event.getOldEntity()); + assertEquals(newNote, event.getNewEntity()); + } + + @Test + void deleteEvent_carriesOldAndOmitsNew() { + var id = UUID.randomUUID(); + var oldNote = new Note().title("old"); + + var event = DomainEvent.deleteEvent(id, oldNote, "diku"); + + assertEquals(DomainEventType.DELETE, event.getType()); + assertEquals(oldNote, event.getOldEntity()); + assertNull(event.getNewEntity()); + } +} + + + diff --git a/src/test/java/org/folio/notes/service/DomainEventPublisherServiceTest.java b/src/test/java/org/folio/notes/service/DomainEventPublisherServiceTest.java new file mode 100644 index 00000000..d713f9a2 --- /dev/null +++ b/src/test/java/org/folio/notes/service/DomainEventPublisherServiceTest.java @@ -0,0 +1,87 @@ +package org.folio.notes.service; + +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.UUID; +import org.folio.notes.domain.dto.Note; +import org.folio.notes.domain.event.DomainEvent; +import org.folio.notes.domain.event.DomainEventType; +import org.folio.notes.integration.kafka.NoteEventProducer; +import org.folio.spring.FolioExecutionContext; +import org.folio.spring.testing.type.UnitTest; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@UnitTest +@ExtendWith(MockitoExtension.class) +class DomainEventPublisherServiceTest { + + @Mock + private NoteEventProducer noteEventProducer; + @Mock + private FolioExecutionContext context; + @InjectMocks + private DomainEventPublisherService service; + + @Test + void publishNoteCreatedEvent_buildsCreateEnvelopeWithoutOld() { + var id = UUID.randomUUID(); + var note = new Note().id(id).title("t"); + when(context.getTenantId()).thenReturn("diku"); + + service.publishNoteCreatedEvent(note); + + var captor = eventCaptor(); + verify(noteEventProducer).publish(eq(id), captor.capture()); + var event = captor.getValue(); + org.junit.jupiter.api.Assertions.assertEquals(DomainEventType.CREATE, event.getType()); + org.junit.jupiter.api.Assertions.assertEquals("diku", event.getTenant()); + org.junit.jupiter.api.Assertions.assertEquals(note, event.getNewEntity()); + org.junit.jupiter.api.Assertions.assertNull(event.getOldEntity()); + } + + @Test + void publishNoteUpdatedEvent_carriesOldAndNew() { + var id = UUID.randomUUID(); + var oldNote = new Note().id(id).title("old"); + var newNote = new Note().id(id).title("new"); + when(context.getTenantId()).thenReturn("diku"); + + service.publishNoteUpdatedEvent(oldNote, newNote); + + var captor = eventCaptor(); + verify(noteEventProducer).publish(eq(id), captor.capture()); + var event = captor.getValue(); + org.junit.jupiter.api.Assertions.assertEquals(DomainEventType.UPDATE, event.getType()); + org.junit.jupiter.api.Assertions.assertEquals(oldNote, event.getOldEntity()); + org.junit.jupiter.api.Assertions.assertEquals(newNote, event.getNewEntity()); + } + + @Test + void publishNoteDeletedEvent_carriesOldAndOmitsNew() { + var id = UUID.randomUUID(); + var oldNote = new Note().id(id).title("old"); + when(context.getTenantId()).thenReturn("diku"); + + service.publishNoteDeletedEvent(oldNote); + + var captor = eventCaptor(); + verify(noteEventProducer).publish(eq(id), captor.capture()); + var event = captor.getValue(); + org.junit.jupiter.api.Assertions.assertEquals(DomainEventType.DELETE, event.getType()); + org.junit.jupiter.api.Assertions.assertEquals(oldNote, event.getOldEntity()); + org.junit.jupiter.api.Assertions.assertNull(event.getNewEntity()); + } + + @SuppressWarnings("unchecked") + private ArgumentCaptor> eventCaptor() { + return ArgumentCaptor.forClass(DomainEvent.class); + } +} + diff --git a/src/test/java/org/folio/notes/support/TestApiBase.java b/src/test/java/org/folio/notes/support/TestApiBase.java index aee8a5fe..022ce2d6 100644 --- a/src/test/java/org/folio/notes/support/TestApiBase.java +++ b/src/test/java/org/folio/notes/support/TestApiBase.java @@ -16,6 +16,7 @@ import org.folio.notes.domain.dto.User; import org.folio.spring.FolioModuleMetadata; import org.folio.spring.integration.XOkapiHeaders; +import org.folio.spring.testing.extension.EnableKafka; import org.folio.spring.testing.extension.EnableOkapi; import org.folio.spring.testing.extension.EnablePostgres; import org.folio.spring.testing.extension.impl.OkapiConfiguration; @@ -41,6 +42,7 @@ @EnableOkapi @EnablePostgres +@EnableKafka @AutoConfigureMockMvc @SpringBootTest @ContextConfiguration From c7293b8b775a108a3bac02298ea6615267b11b64 Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 12:54:28 +0400 Subject: [PATCH 06/12] [MODNOTES-295] Implement integration tests for domain events --- .../notes/controller/NoteDomainEventIT.java | 129 ++++++++++++++++ .../notes/support/TestKafkaConsumer.java | 143 ++++++++++++++++++ 2 files changed, 272 insertions(+) create mode 100644 src/test/java/org/folio/notes/controller/NoteDomainEventIT.java create mode 100644 src/test/java/org/folio/notes/support/TestKafkaConsumer.java diff --git a/src/test/java/org/folio/notes/controller/NoteDomainEventIT.java b/src/test/java/org/folio/notes/controller/NoteDomainEventIT.java new file mode 100644 index 00000000..0b2c1661 --- /dev/null +++ b/src/test/java/org/folio/notes/controller/NoteDomainEventIT.java @@ -0,0 +1,129 @@ +package org.folio.notes.controller; + +import static org.folio.notes.support.DatabaseHelper.LINK; +import static org.folio.notes.support.DatabaseHelper.NOTE; +import static org.folio.notes.support.DatabaseHelper.TYPE; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import java.time.Duration; +import java.util.List; +import java.util.UUID; +import org.folio.notes.domain.dto.Link; +import org.folio.notes.domain.dto.Note; +import org.folio.notes.domain.dto.User; +import org.folio.notes.domain.entity.NoteTypeEntity; +import org.folio.notes.support.TestApiBase; +import org.folio.notes.support.TestKafkaConsumer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.kafka.autoconfigure.KafkaProperties; +import tools.jackson.databind.JsonNode; + +@DisplayName("Note create domain events") +class NoteDomainEventIT extends TestApiBase { + + private static final String NOTE_URL = "/notes"; + private static final String NOTE_TOPIC = "folio.test.notes.note"; + private static final String ORDER_LINE_TYPE = "order-line"; + + private static final UUID NOTE_TYPE_ID = UUID.fromString("2af21797-d25b-46dc-8427-1759d1db2057"); + private static final String NOTE_TYPE_NAME = "General note"; + + @Autowired + private KafkaProperties kafkaProperties; + + private TestKafkaConsumer consumer; + + @BeforeEach + void setUp() { + stubUser(new User(USER_ID, "test_user", null)); + databaseHelper.clearTable(TENANT, NOTE); + databaseHelper.clearTable(TENANT, LINK); + databaseHelper.clearTable(TENANT, TYPE); + + var noteType = new NoteTypeEntity(); + noteType.setId(NOTE_TYPE_ID); + noteType.setName(NOTE_TYPE_NAME); + noteType.setCreatedBy(USER_ID); + databaseHelper.saveNoteType(noteType, TENANT); + + consumer = TestKafkaConsumer.subscribe(NOTE_TOPIC, kafkaProperties); + } + + @AfterEach + void tearDown() { + if (consumer != null) { + consumer.close(); + } + } + + @Test + @DisplayName("create publishes a CREATE event with the full snapshot and no 'old' field") + void createNote_publishesCreateEvent() throws Exception { + var link = new Link().id(UUID.randomUUID().toString()).type(ORDER_LINE_TYPE); + var note = new Note() + .title("Kafka title") + .content("Kafka details") + .domain("orders") + .typeId(NOTE_TYPE_ID) + .links(List.of(link)); + + var response = mockMvc.perform(post(NOTE_URL).headers(okapiHeaders()).content(asJsonString(note))) + .andExpect(status().isCreated()) + .andReturn().getResponse().getContentAsString(); + + var event = consumer.poll(); + var envelope = OBJECT_MAPPER.readTree(event.value()); + + assertEnvelope(envelope, event.value()); + assertPayload(envelope.get("new"), OBJECT_MAPPER.readTree(response).get("id").asString(), link); + } + + @Test + @DisplayName("invalid payload (missing title) publishes no event") + void createNote_invalidPayload_publishesNoEvent() throws Exception { + var note = new Note() + .domain("orders") + .typeId(NOTE_TYPE_ID) + .links(List.of(new Link().id(UUID.randomUUID().toString()).type(ORDER_LINE_TYPE))); + + mockMvc.perform(post(NOTE_URL).headers(okapiHeaders()).content(asJsonString(note))) + .andExpect(status().is4xxClientError()); + + var event = consumer.pollNullable(Duration.ofSeconds(10)); + assertNull(event, "No event should be published for an invalid create"); + } + + private void assertEnvelope(JsonNode envelope, String rawJson) { + assertNotNull(envelope.get("id"), "eventId (id) must be present"); + assertEquals("CREATE", envelope.get("type").asString()); + assertEquals(TENANT, envelope.get("tenant").asString()); + assertNull(envelope.get("old"), "CREATE event must not carry an 'old' field"); + assertFalse(rawJson.contains("\"old\""), "JSON must not contain an 'old' field"); + } + + private void assertPayload(JsonNode newNode, String createdId, Link link) { + assertNotNull(newNode); + assertEquals(createdId, newNode.get("id").asString()); + assertEquals("Kafka title", newNode.get("title").asString()); + assertEquals("Kafka details", newNode.get("content").asString()); + assertEquals(NOTE_TYPE_ID.toString(), newNode.get("typeId").asString()); + assertEquals(USER_ID.toString(), newNode.get("metadata").get("createdByUserId").asString()); + + var links = newNode.get("links"); + assertEquals(1, links.size()); + assertEquals(ORDER_LINE_TYPE, links.get(0).get("type").asString()); + assertEquals(link.getId(), links.get(0).get("id").asString()); + } +} + + + diff --git a/src/test/java/org/folio/notes/support/TestKafkaConsumer.java b/src/test/java/org/folio/notes/support/TestKafkaConsumer.java new file mode 100644 index 00000000..fb7cd226 --- /dev/null +++ b/src/test/java/org/folio/notes/support/TestKafkaConsumer.java @@ -0,0 +1,143 @@ +package org.folio.notes.support; + +import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.io.Closeable; +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.errors.TopicExistsException; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.springframework.boot.kafka.autoconfigure.KafkaProperties; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.kafka.listener.MessageListener; + +/** + * Reusable test consumer that subscribes to a single Kafka topic with a raw {@code String} value deserializer and + * buffers every received record. Integration tests use it to assert on the JSON payload of published domain events. + * + *

The consumer owns its listener container and the record buffer; callers only need to + * {@link #subscribe(String, KafkaProperties)} it, {@link #poll()} the buffered records, and {@link #close()} it when + * done (it is {@link Closeable}, so it also works with try-with-resources or an {@code @AfterEach} hook).

+ */ +public final class TestKafkaConsumer implements Closeable { + + private static final Duration DEFAULT_POLL_TIMEOUT = Duration.ofMinutes(1); + private static final Duration POLL_INTERVAL = Duration.ofSeconds(1); + + private final KafkaMessageListenerContainer container; + private final BlockingQueue> records = new LinkedBlockingQueue<>(); + + private TestKafkaConsumer(KafkaMessageListenerContainer container) { + this.container = container; + } + + /** + * Creates and starts a consumer subscribed to the given topic. + * + * @param topic the topic to consume from (already env/tenant qualified) + * @param properties Spring Kafka properties (bootstrap servers point at the embedded broker) + * @return a started consumer; close it when done + */ + public static TestKafkaConsumer subscribe(String topic, KafkaProperties properties) { + createTopic(topic, properties); + properties.getConsumer().setGroupId("mod-notes-test-group"); + properties.getConsumer().setAutoOffsetReset("earliest"); + Map config = new HashMap<>(properties.buildConsumerProperties()); + config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + + var consumerFactory = new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), new StringDeserializer()); + var containerProperties = new ContainerProperties(topic); + var container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties); + + var consumer = new TestKafkaConsumer(container); + container.setupMessageListener((MessageListener) consumer.records::add); + container.start(); + return consumer; + } + + /** + * Eagerly creates the topic via an admin client so the producer does not race the broker's lazy auto-creation + * (which otherwise surfaces as {@code Topic ... not present in metadata} on the first send). + * + * @param topic the topic to create (no-op if it already exists) + * @param properties Spring Kafka properties (used for the bootstrap servers) + */ + private static void createTopic(String topic, KafkaProperties properties) { + try (var admin = Admin.create(properties.buildAdminProperties())) { + admin.createTopics(List.of(new NewTopic(topic, 1, (short) 1))).all().get(); + } catch (ExecutionException e) { + if (!(e.getCause() instanceof TopicExistsException)) { + throw new IllegalStateException("Failed to create test topic " + topic, e); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while creating test topic " + topic, e); + } + } + + /** + * Waits (up to one minute) for the next record and returns it. + * + * @return the received record + */ + public ConsumerRecord poll() { + return poll(DEFAULT_POLL_TIMEOUT); + } + + /** + * Waits up to {@code timeout} for the next record and returns it, failing the calling test if none arrives. + * + * @param timeout the maximum time to wait + * @return the received record + */ + public ConsumerRecord poll(Duration timeout) { + var holder = new AtomicReference>(); + await().pollInterval(POLL_INTERVAL).atMost(timeout) + .untilAsserted(() -> { + var record = records.poll(); + assertNotNull(record, "Expected a record on the topic"); + holder.set(record); + }); + return holder.get(); + } + + /** + * Returns the next record if one arrives within {@code timeout}, or {@code null} otherwise (no assertion). Useful for + * verifying that no event was published. + * + * @param timeout the maximum time to wait + * @return the received record, or {@code null} if none arrived + */ + public ConsumerRecord pollNullable(Duration timeout) { + try { + return records.poll(timeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while polling for records", e); + } + } + + @Override + public void close() { + container.stop(); + } +} + + + From 555c47cb89c8d3f0faa5015388e641e7f1fbaf5a Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 13:54:39 +0400 Subject: [PATCH 07/12] [MODNOTES-295] Update local setup to include kafka --- docker/.env | 12 ++++++++ docker/README.md | 18 ++++++++++++ docker/app-docker-compose.yml | 5 ++++ docker/infra-docker-compose.yml | 49 +++++++++++++++++++++++++++++++++ docker/kafka-init.sh | 34 +++++++++++++++++++++++ 5 files changed, 118 insertions(+) create mode 100644 docker/kafka-init.sh diff --git a/docker/.env b/docker/.env index bc38ca40..5bc33870 100644 --- a/docker/.env +++ b/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 + + diff --git a/docker/README.md b/docker/README.md index c784caa7..439fa568 100644 --- a/docker/README.md +++ b/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. diff --git a/docker/app-docker-compose.yml b/docker/app-docker-compose.yml index 3b9d6d60..1f03ba97 100644 --- a/docker/app-docker-compose.yml +++ b/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: diff --git a/docker/infra-docker-compose.yml b/docker/infra-docker-compose.yml index 005e7f10..5a611df7 100644 --- a/docker/infra-docker-compose.yml +++ b/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 diff --git a/docker/kafka-init.sh b/docker/kafka-init.sh new file mode 100644 index 00000000..98d7635b --- /dev/null +++ b/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." + From 1c855b479d3e44eb1e82cf221b1b276b4316347d Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Mon, 20 Jul 2026 18:21:36 +0400 Subject: [PATCH 08/12] [MODNOTES-295] Update module descriptor --- README.md | 6 +++++- descriptors/ModuleDescriptor-template.json | 12 ++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 82296338..cc0436f9 100644 --- a/README.md +++ b/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. diff --git a/descriptors/ModuleDescriptor-template.json b/descriptors/ModuleDescriptor-template.json index 9ea8fef4..f794f230 100644 --- a/descriptors/ModuleDescriptor-template.json +++ b/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" } ] } From 222693d6c88c50e0f31b9fd85cc711b301f6f600 Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Wed, 22 Jul 2026 13:35:01 +0400 Subject: [PATCH 09/12] [MODNOTES-295] Address comments --- descriptors/ModuleDescriptor-template.json | 33 ++++- docker/.env | 3 - docker/README.md | 4 +- docker/app-docker-compose.yml | 1 - docker/infra-docker-compose.yml | 15 -- docker/kafka-init.sh | 34 ----- pom.xml | 6 + .../integration/kafka/NoteEventProducer.java | 42 ++++-- .../notes/service/impl/NoteTenantService.java | 15 +- .../notes/service/impl/NotesServiceImpl.java | 6 +- src/main/resources/application.yaml | 4 +- .../notes/controller/NoteDomainEventIT.java | 129 ------------------ .../notes/controller/NotesControllerIT.java | 71 ++++++++++ .../notes/service/NoteTenantServiceTest.java | 21 ++- .../notes/support/TestKafkaConsumer.java | 38 ++---- 15 files changed, 191 insertions(+), 231 deletions(-) delete mode 100644 docker/kafka-init.sh delete mode 100644 src/test/java/org/folio/notes/controller/NoteDomainEventIT.java diff --git a/descriptors/ModuleDescriptor-template.json b/descriptors/ModuleDescriptor-template.json index f794f230..5e41f685 100644 --- a/descriptors/ModuleDescriptor-template.json +++ b/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" } ] } diff --git a/docker/.env b/docker/.env index 5bc33870..cff3c0f9 100644 --- a/docker/.env +++ b/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 diff --git a/docker/README.md b/docker/README.md index 439fa568..00e46337 100644 --- a/docker/README.md +++ b/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 diff --git a/docker/app-docker-compose.yml b/docker/app-docker-compose.yml index 1f03ba97..f51cc7bf 100644 --- a/docker/app-docker-compose.yml +++ b/docker/app-docker-compose.yml @@ -30,7 +30,6 @@ services: - postgres - wiremock - kafka - - kafka-topic-init deploy: replicas: ${MODULE_REPLICAS:-1} restart_policy: diff --git a/docker/infra-docker-compose.yml b/docker/infra-docker-compose.yml index 5a611df7..de07c3bd 100644 --- a/docker/infra-docker-compose.yml +++ b/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: diff --git a/docker/kafka-init.sh b/docker/kafka-init.sh deleted file mode 100644 index 98d7635b..00000000 --- a/docker/kafka-init.sh +++ /dev/null @@ -1,34 +0,0 @@ -#!/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." - diff --git a/pom.xml b/pom.xml index 05cfc563..e3c3de2c 100644 --- a/pom.xml +++ b/pom.xml @@ -27,6 +27,7 @@ UTF-8 10.0.0 + 6.0.0 1.6.3 1.22.2 1.18.44 @@ -64,6 +65,11 @@ folio-spring-cql ${folio-spring-support.version} + + org.folio + folio-service-tools-spring-dev + ${folio-service-tools.version} + org.springframework.boot spring-boot-starter-cache diff --git a/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java index 29ac3b1e..30c9be9c 100644 --- a/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java +++ b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java @@ -1,11 +1,18 @@ 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; @@ -13,26 +20,28 @@ * Publishes Note {@link DomainEvent}s to the tenant-scoped Kafka topic. * *

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}.

+ * {@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}.

+ * + *

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.

*/ @Slf4j @Component public class NoteEventProducer { + private static final String EVENT_TYPE_HEADER = "domain-event-type"; + private final KafkaTemplate> kafkaTemplate; private final FolioExecutionContext context; - private final String env; private final String noteTopic; public NoteEventProducer(KafkaTemplate> 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> kafkaTemplate, * @param event the domain event envelope to publish */ public void publish(UUID id, DomainEvent 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
buildHeaders(DomainEvent event) { + var headers = new ArrayList
(); + 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)); + } + } diff --git a/src/main/java/org/folio/notes/service/impl/NoteTenantService.java b/src/main/java/org/folio/notes/service/impl/NoteTenantService.java index 4bf01817..7852a233 100644 --- a/src/main/java/org/folio/notes/service/impl/NoteTenantService.java +++ b/src/main/java/org/folio/notes/service/impl/NoteTenantService.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()); + } } diff --git a/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.java b/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.java index 8b4a0788..00c91e44 100644 --- a/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.java +++ b/src/main/java/org/folio/notes/service/impl/NotesServiceImpl.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 diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 6050dd93..a85d449a 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -50,7 +50,9 @@ folio: limit: ${MAX_RECORDS_COUNT:1000} kafka: topics: - note: notes.note + - name: notes.note + numPartitions: ${KAFKA_NOTE_TOPIC_PARTITIONS:1} + replicationFactor: ${KAFKA_NOTE_TOPIC_REPLICATION_FACTOR:} # Spring properties spring: diff --git a/src/test/java/org/folio/notes/controller/NoteDomainEventIT.java b/src/test/java/org/folio/notes/controller/NoteDomainEventIT.java deleted file mode 100644 index 0b2c1661..00000000 --- a/src/test/java/org/folio/notes/controller/NoteDomainEventIT.java +++ /dev/null @@ -1,129 +0,0 @@ -package org.folio.notes.controller; - -import static org.folio.notes.support.DatabaseHelper.LINK; -import static org.folio.notes.support.DatabaseHelper.NOTE; -import static org.folio.notes.support.DatabaseHelper.TYPE; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; -import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; - -import java.time.Duration; -import java.util.List; -import java.util.UUID; -import org.folio.notes.domain.dto.Link; -import org.folio.notes.domain.dto.Note; -import org.folio.notes.domain.dto.User; -import org.folio.notes.domain.entity.NoteTypeEntity; -import org.folio.notes.support.TestApiBase; -import org.folio.notes.support.TestKafkaConsumer; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.kafka.autoconfigure.KafkaProperties; -import tools.jackson.databind.JsonNode; - -@DisplayName("Note create domain events") -class NoteDomainEventIT extends TestApiBase { - - private static final String NOTE_URL = "/notes"; - private static final String NOTE_TOPIC = "folio.test.notes.note"; - private static final String ORDER_LINE_TYPE = "order-line"; - - private static final UUID NOTE_TYPE_ID = UUID.fromString("2af21797-d25b-46dc-8427-1759d1db2057"); - private static final String NOTE_TYPE_NAME = "General note"; - - @Autowired - private KafkaProperties kafkaProperties; - - private TestKafkaConsumer consumer; - - @BeforeEach - void setUp() { - stubUser(new User(USER_ID, "test_user", null)); - databaseHelper.clearTable(TENANT, NOTE); - databaseHelper.clearTable(TENANT, LINK); - databaseHelper.clearTable(TENANT, TYPE); - - var noteType = new NoteTypeEntity(); - noteType.setId(NOTE_TYPE_ID); - noteType.setName(NOTE_TYPE_NAME); - noteType.setCreatedBy(USER_ID); - databaseHelper.saveNoteType(noteType, TENANT); - - consumer = TestKafkaConsumer.subscribe(NOTE_TOPIC, kafkaProperties); - } - - @AfterEach - void tearDown() { - if (consumer != null) { - consumer.close(); - } - } - - @Test - @DisplayName("create publishes a CREATE event with the full snapshot and no 'old' field") - void createNote_publishesCreateEvent() throws Exception { - var link = new Link().id(UUID.randomUUID().toString()).type(ORDER_LINE_TYPE); - var note = new Note() - .title("Kafka title") - .content("Kafka details") - .domain("orders") - .typeId(NOTE_TYPE_ID) - .links(List.of(link)); - - var response = mockMvc.perform(post(NOTE_URL).headers(okapiHeaders()).content(asJsonString(note))) - .andExpect(status().isCreated()) - .andReturn().getResponse().getContentAsString(); - - var event = consumer.poll(); - var envelope = OBJECT_MAPPER.readTree(event.value()); - - assertEnvelope(envelope, event.value()); - assertPayload(envelope.get("new"), OBJECT_MAPPER.readTree(response).get("id").asString(), link); - } - - @Test - @DisplayName("invalid payload (missing title) publishes no event") - void createNote_invalidPayload_publishesNoEvent() throws Exception { - var note = new Note() - .domain("orders") - .typeId(NOTE_TYPE_ID) - .links(List.of(new Link().id(UUID.randomUUID().toString()).type(ORDER_LINE_TYPE))); - - mockMvc.perform(post(NOTE_URL).headers(okapiHeaders()).content(asJsonString(note))) - .andExpect(status().is4xxClientError()); - - var event = consumer.pollNullable(Duration.ofSeconds(10)); - assertNull(event, "No event should be published for an invalid create"); - } - - private void assertEnvelope(JsonNode envelope, String rawJson) { - assertNotNull(envelope.get("id"), "eventId (id) must be present"); - assertEquals("CREATE", envelope.get("type").asString()); - assertEquals(TENANT, envelope.get("tenant").asString()); - assertNull(envelope.get("old"), "CREATE event must not carry an 'old' field"); - assertFalse(rawJson.contains("\"old\""), "JSON must not contain an 'old' field"); - } - - private void assertPayload(JsonNode newNode, String createdId, Link link) { - assertNotNull(newNode); - assertEquals(createdId, newNode.get("id").asString()); - assertEquals("Kafka title", newNode.get("title").asString()); - assertEquals("Kafka details", newNode.get("content").asString()); - assertEquals(NOTE_TYPE_ID.toString(), newNode.get("typeId").asString()); - assertEquals(USER_ID.toString(), newNode.get("metadata").get("createdByUserId").asString()); - - var links = newNode.get("links"); - assertEquals(1, links.size()); - assertEquals(ORDER_LINE_TYPE, links.get(0).get("type").asString()); - assertEquals(link.getId(), links.get(0).get("id").asString()); - } -} - - - diff --git a/src/test/java/org/folio/notes/controller/NotesControllerIT.java b/src/test/java/org/folio/notes/controller/NotesControllerIT.java index dda567d6..fdf6be00 100644 --- a/src/test/java/org/folio/notes/controller/NotesControllerIT.java +++ b/src/test/java/org/folio/notes/controller/NotesControllerIT.java @@ -14,6 +14,8 @@ import static org.hamcrest.Matchers.not; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.params.provider.Arguments.arguments; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete; @@ -44,6 +46,7 @@ import org.folio.notes.domain.entity.NoteTypeEntity; import org.folio.notes.exception.NoteNotFoundException; import org.folio.notes.support.TestApiBase; +import org.folio.notes.support.TestKafkaConsumer; import org.folio.spring.cql.CqlQueryValidationException; import org.hamcrest.Matcher; import org.hamcrest.MatcherAssert; @@ -55,13 +58,16 @@ import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.kafka.autoconfigure.KafkaProperties; import org.springframework.http.HttpHeaders; import org.springframework.test.context.TestPropertySource; import org.springframework.test.web.servlet.ResultMatcher; import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder; import org.springframework.web.bind.MethodArgumentNotValidException; import org.springframework.web.method.annotation.MethodArgumentTypeMismatchException; +import tools.jackson.databind.JsonNode; @TestPropertySource(properties = {"folio.notes.types.defaults.limit=5"}) class NotesControllerIT extends TestApiBase { @@ -88,8 +94,15 @@ class NotesControllerIT extends TestApiBase { private static final UUID[] TYPE_IDS = new UUID[] {randomUUID(), randomUUID()}; private static final int DEFAULT_LINK_AMOUNT = 1; + private static final String NOTE_EVENT_TOPIC = "folio.test.notes.note"; + private static final String ORDER_LINE_TYPE = "order-line"; + private static final UUID EVENT_NOTE_TYPE_ID = UUID.fromString("2af21797-d25b-46dc-8427-1759d1db2057"); + private static final String EVENT_NOTE_TYPE_NAME = "General note"; + @Value("${folio.notes.types.defaults.limit}") private String defaultNoteTypeLimit; + @Autowired + private KafkaProperties kafkaProperties; @BeforeEach void setUp() { @@ -1065,6 +1078,64 @@ private NoteEntity generateNoteEntityWithParams(String title, String typeId, Str return noteEntity; } + // Tests for domain events + @Test + @DisplayName("create publishes a CREATE event with the full snapshot and no 'old' field") + void createNote_publishesCreateEvent() throws Exception { + saveEventNoteType(); + var link = new Link().id(UUID.randomUUID().toString()).type(ORDER_LINE_TYPE); + var note = new Note() + .title("Kafka title") + .content("Kafka details") + .domain("orders") + .typeId(EVENT_NOTE_TYPE_ID) + .links(List.of(link)); + + try (var consumer = TestKafkaConsumer.subscribe(NOTE_EVENT_TOPIC, kafkaProperties)) { + var response = mockMvc.perform(postNote(note)) + .andExpect(status().isCreated()) + .andReturn().getResponse().getContentAsString(); + var newNoteId = OBJECT_MAPPER.readTree(response).get("id").asString(); + + var event = consumer.poll(newNoteId); + assertNotNull(event); + + var envelope = OBJECT_MAPPER.readTree(event.value()); + assertEventEnvelope(envelope, event.value()); + assertEventPayload(envelope.get("new"), newNoteId, link); + } + } + + private void saveEventNoteType() { + var noteType = new NoteTypeEntity(); + noteType.setId(EVENT_NOTE_TYPE_ID); + noteType.setName(EVENT_NOTE_TYPE_NAME); + noteType.setCreatedBy(USER_ID); + databaseHelper.saveNoteType(noteType, TENANT); + } + + private void assertEventEnvelope(JsonNode envelope, String rawJson) { + assertNotNull(envelope.get("id"), "eventId (id) must be present"); + assertEquals("CREATE", envelope.get("type").asString()); + assertEquals(TENANT, envelope.get("tenant").asString()); + assertNull(envelope.get("old"), "CREATE event must not carry an 'old' field"); + assertFalse(rawJson.contains("\"old\""), "JSON must not contain an 'old' field"); + } + + private void assertEventPayload(JsonNode newNode, String createdId, Link link) { + assertNotNull(newNode); + assertEquals(createdId, newNode.get("id").asString()); + assertEquals("Kafka title", newNode.get("title").asString()); + assertEquals("Kafka details", newNode.get("content").asString()); + assertEquals(EVENT_NOTE_TYPE_ID.toString(), newNode.get("typeId").asString()); + assertEquals(USER_ID.toString(), newNode.get("metadata").get("createdByUserId").asString()); + + var links = newNode.get("links"); + assertEquals(1, links.size()); + assertEquals(ORDER_LINE_TYPE, links.get(0).get("type").asString()); + assertEquals(link.getId(), links.get(0).get("id").asString()); + } + private Note generateNote() throws Exception { var noteType = new NoteType().name(insecure().nextAlphabetic(100)); var notyTypeAsString = mockMvc.perform(postNoteType(noteType)).andExpect(status().isCreated()) diff --git a/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java b/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java index eb6c3816..4e9bab1c 100644 --- a/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java +++ b/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java @@ -3,23 +3,40 @@ import static org.mockito.Mockito.verify; import org.folio.notes.service.impl.NoteTenantService; +import org.folio.spring.FolioExecutionContext; +import org.folio.spring.liquibase.FolioSpringLiquibase; import org.folio.spring.testing.type.UnitTest; +import org.folio.spring.tools.kafka.KafkaAdminService; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.jdbc.core.JdbcTemplate; @UnitTest @ExtendWith(MockitoExtension.class) class NoteTenantServiceTest { + @Mock + private JdbcTemplate jdbcTemplate; + @Mock + private FolioExecutionContext context; + @Mock + private FolioSpringLiquibase folioSpringLiquibase; @Mock private NoteTypesService noteTypesService; + @Mock + private KafkaAdminService kafkaAdminService; - @InjectMocks private NoteTenantService tenantService; + @BeforeEach + void setUp() { + tenantService = new NoteTenantService(jdbcTemplate, context, folioSpringLiquibase, noteTypesService, + kafkaAdminService); + } + @Test void shouldPopulateDefaultTypeOnLoadReferenceData() { tenantService.loadReferenceData(); diff --git a/src/test/java/org/folio/notes/support/TestKafkaConsumer.java b/src/test/java/org/folio/notes/support/TestKafkaConsumer.java index fb7cd226..ccf67b9d 100644 --- a/src/test/java/org/folio/notes/support/TestKafkaConsumer.java +++ b/src/test/java/org/folio/notes/support/TestKafkaConsumer.java @@ -3,17 +3,17 @@ import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; import static org.awaitility.Awaitility.await; -import static org.junit.jupiter.api.Assertions.assertNotNull; import java.io.Closeable; import java.time.Duration; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutionException; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.admin.NewTopic; @@ -31,7 +31,7 @@ * buffers every received record. Integration tests use it to assert on the JSON payload of published domain events. * *

The consumer owns its listener container and the record buffer; callers only need to - * {@link #subscribe(String, KafkaProperties)} it, {@link #poll()} the buffered records, and {@link #close()} it when + * {@link #subscribe(String, KafkaProperties)} it, {@link #poll(Duration)} the buffered records, and {@link #close()} it when * done (it is {@link Closeable}, so it also works with try-with-resources or an {@code @AfterEach} hook).

*/ public final class TestKafkaConsumer implements Closeable { @@ -96,8 +96,10 @@ private static void createTopic(String topic, KafkaProperties properties) { * * @return the received record */ - public ConsumerRecord poll() { - return poll(DEFAULT_POLL_TIMEOUT); + public ConsumerRecord poll(String key) { + return poll(DEFAULT_POLL_TIMEOUT).stream() + .filter(e -> Objects.equals(e.key(), key)).findFirst() + .orElse(null); } /** @@ -106,33 +108,17 @@ public ConsumerRecord poll() { * @param timeout the maximum time to wait * @return the received record */ - public ConsumerRecord poll(Duration timeout) { - var holder = new AtomicReference>(); + public List> poll(Duration timeout) { + var holder = new AtomicReference>>(); await().pollInterval(POLL_INTERVAL).atMost(timeout) .untilAsserted(() -> { - var record = records.poll(); - assertNotNull(record, "Expected a record on the topic"); - holder.set(record); + List> events = new ArrayList<>(); + records.drainTo(events); + holder.set(events); }); return holder.get(); } - /** - * Returns the next record if one arrives within {@code timeout}, or {@code null} otherwise (no assertion). Useful for - * verifying that no event was published. - * - * @param timeout the maximum time to wait - * @return the received record, or {@code null} if none arrived - */ - public ConsumerRecord pollNullable(Duration timeout) { - try { - return records.poll(timeout.toMillis(), TimeUnit.MILLISECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IllegalStateException("Interrupted while polling for records", e); - } - } - @Override public void close() { container.stop(); From 3d9d8622823d7540c14fdef5a95ccac5b86a5f4b Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Wed, 22 Jul 2026 13:48:17 +0400 Subject: [PATCH 10/12] [MODNOTES-295] Fix style --- .../org/folio/notes/integration/kafka/NoteEventProducer.java | 1 - 1 file changed, 1 deletion(-) diff --git a/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java index 30c9be9c..b1357223 100644 --- a/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java +++ b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java @@ -72,5 +72,4 @@ private List
buildHeaders(DomainEvent event) { private Header header(String key, String value) { return new RecordHeader(key, value == null ? null : value.getBytes(StandardCharsets.UTF_8)); } - } From 58aace8c1cd4d3a41dab288b6071ba30f10ec499 Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Wed, 22 Jul 2026 15:01:13 +0400 Subject: [PATCH 11/12] [MODNOTES-295] Fix style --- src/test/java/org/folio/notes/support/TestKafkaConsumer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/test/java/org/folio/notes/support/TestKafkaConsumer.java b/src/test/java/org/folio/notes/support/TestKafkaConsumer.java index ccf67b9d..5b47d491 100644 --- a/src/test/java/org/folio/notes/support/TestKafkaConsumer.java +++ b/src/test/java/org/folio/notes/support/TestKafkaConsumer.java @@ -31,8 +31,8 @@ * buffers every received record. Integration tests use it to assert on the JSON payload of published domain events. * *

The consumer owns its listener container and the record buffer; callers only need to - * {@link #subscribe(String, KafkaProperties)} it, {@link #poll(Duration)} the buffered records, and {@link #close()} it when - * done (it is {@link Closeable}, so it also works with try-with-resources or an {@code @AfterEach} hook).

+ * {@link #subscribe(String, KafkaProperties)} it, {@link #poll(Duration)} the buffered records, and {@link #close()} + * it when done (it is {@link Closeable}, so it also works with try-with-resources or an {@code @AfterEach} hook).

*/ public final class TestKafkaConsumer implements Closeable { From a9a22adf7886c56fbe5863c80b12879a77bd5692 Mon Sep 17 00:00:00 2001 From: saba_zedginidze Date: Wed, 22 Jul 2026 15:46:12 +0400 Subject: [PATCH 12/12] [MODNOTES-295] Add missing domain header --- .../notes/integration/kafka/NoteEventProducer.java | 12 +++++++++++- .../folio/notes/service/NoteTenantServiceTest.java | 9 ++------- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java index b1357223..0b6c2c7c 100644 --- a/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java +++ b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java @@ -3,6 +3,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; +import java.util.Optional; import java.util.UUID; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.ProducerRecord; @@ -30,7 +31,8 @@ @Component public class NoteEventProducer { - private static final String EVENT_TYPE_HEADER = "domain-event-type"; + private static final String EVENT_TYPE_HEADER = "eventType"; + private static final String EVENT_DOMAIN_HEADER = "domain"; private final KafkaTemplate> kafkaTemplate; private final FolioExecutionContext context; @@ -65,6 +67,7 @@ public void publish(UUID id, DomainEvent event) { private List
buildHeaders(DomainEvent event) { var headers = new ArrayList
(); headers.add(header(EVENT_TYPE_HEADER, event.getType().name())); + headers.add(header(EVENT_DOMAIN_HEADER, extractDomain(event))); context.getAllHeaders().forEach((key, value) -> headers.add(header(key, value.iterator().next()))); return headers; } @@ -72,4 +75,11 @@ private List
buildHeaders(DomainEvent event) { private Header header(String key, String value) { return new RecordHeader(key, value == null ? null : value.getBytes(StandardCharsets.UTF_8)); } + + private String extractDomain(DomainEvent event) { + return Optional.ofNullable(event.getNewEntity()) + .or(() -> Optional.ofNullable(event.getOldEntity())) + .map(Note::getDomain) + .orElse(null); + } } diff --git a/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java b/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java index 4e9bab1c..d6fac314 100644 --- a/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java +++ b/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java @@ -7,9 +7,9 @@ import org.folio.spring.liquibase.FolioSpringLiquibase; import org.folio.spring.testing.type.UnitTest; import org.folio.spring.tools.kafka.KafkaAdminService; -import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.jdbc.core.JdbcTemplate; @@ -29,14 +29,9 @@ class NoteTenantServiceTest { @Mock private KafkaAdminService kafkaAdminService; + @InjectMocks private NoteTenantService tenantService; - @BeforeEach - void setUp() { - tenantService = new NoteTenantService(jdbcTemplate, context, folioSpringLiquibase, noteTypesService, - kafkaAdminService); - } - @Test void shouldPopulateDefaultTypeOnLoadReferenceData() { tenantService.loadReferenceData();