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..5e41f685 100644 --- a/descriptors/ModuleDescriptor-template.json +++ b/descriptors/ModuleDescriptor-template.json @@ -313,6 +313,49 @@ { "name": "NOTES_TYPES_DEFAULTS_LIMIT", "value": "25" + }, + { + "name": "KAFKA_HOST", + "value": "kafka" + }, + { + "name": "KAFKA_PORT", + "value": "9092" + }, + { + "name": "KAFKA_SECURITY_PROTOCOL", + "value": "PLAINTEXT", + "description": "Kafka security protocol used to communicate with brokers (SSL or PLAINTEXT)" + }, + { + "name": "KAFKA_SSL_KEYSTORE_LOCATION", + "description": "The location of the Kafka key store file. This is optional for client and can be used for two-way authentication for client." + }, + { + "name": "KAFKA_SSL_KEYSTORE_PASSWORD", + "description": "The store password for the Kafka key store file. This is optional for client and only needed if 'ssl.keystore.location' is configured." + }, + { + "name": "KAFKA_SSL_TRUSTSTORE_LOCATION", + "description": "The location of the Kafka trust store file." + }, + { + "name": "KAFKA_SSL_TRUSTSTORE_PASSWORD", + "description": "The password for the Kafka trust store file. If a password is not set, trust store file configured will still be used, but integrity checking is disabled." + }, + { + "name": "KAFKA_NOTE_TOPIC_PARTITIONS", + "value": "1", + "description": "Number of partitions for the `notes.note` topic." + }, + { + "name": "KAFKA_NOTE_TOPIC_REPLICATION_FACTOR", + "description": "Replication factor for the `notes.note` topic. When not set, the broker default is used." + }, + { + "name": "ENV", + "value": "folio", + "description": "The logical name of the deployment, must be unique across all environments using the same shared Kafka/Elasticsearch clusters, `a-z (any case)`, `0-9`, `-`, `_` symbols only allowed" } ] } diff --git a/docker/.env b/docker/.env index bc38ca40..cff3c0f9 100644 --- a/docker/.env +++ b/docker/.env @@ -22,3 +22,12 @@ PGADMIN_DEFAULT_PASSWORD=admin WIREMOCK_PORT=9130 OKAPI_URL=http://wiremock:8080 +# Kafka configuration +ENV=folio +KAFKA_HOST=kafka +KAFKA_PORT=9093 +KAFKA_EXTERNAL_PORT=9092 +# Kafka UI (web) port -> http://localhost:8090 +KAFKA_UI_PORT=8090 + + diff --git a/docker/README.md b/docker/README.md index c784caa7..00e46337 100644 --- a/docker/README.md +++ b/docker/README.md @@ -34,6 +34,11 @@ Configuration is managed via the `.env` file in this directory. | `PGADMIN_PORT` | PgAdmin port | `5050` | | `WIREMOCK_PORT` | WireMock (Okapi mock) port | `9130` | | `OKAPI_URL` | Okapi URL for the module | `http://wiremock:8080` | +| `ENV` | Kafka topic prefix | `folio` | +| `KAFKA_HOST` | Broker host (in-cluster) | `kafka` | +| `KAFKA_PORT` | Broker port (in-cluster) | `9093` | +| `KAFKA_EXTERNAL_PORT` | Broker port from host/IDE | `9092` | +| `KAFKA_UI_PORT` | Kafka UI web port | `8090` | ## 🚀 Services @@ -54,6 +59,17 @@ Configuration is managed via the `.env` file in this directory. - **Access**: http://localhost:9130 (configurable via `WIREMOCK_PORT`) - **Mappings**: Located in `src/test/resources/mappings` +### Kafka +- **Purpose**: Message broker the module publishes Note domain events to +- **Image**: `apache/kafka-native` (KRaft mode, no Zookeeper) +- **Access (host/IDE)**: `localhost:9092` (configurable via `KAFKA_EXTERNAL_PORT`) +- **Access (in-cluster)**: `kafka:9093` +- **Topics**: `{ENV}.{tenant}.notes.note` — e.g. `folio.diku.notes.note`. The module creates the topic when a tenant is enabled (via the `/_/tenant` API); the broker also auto-creates it on first publish. + +### Kafka UI +- **Purpose**: Web UI to browse topics and inspect published event messages +- **Access**: http://localhost:8090 (configurable via `KAFKA_UI_PORT`) + ## 📖 Usage > **Note**: All commands in this guide assume you are in the `docker/` directory. If you're at the project root, run `cd docker` first. diff --git a/docker/app-docker-compose.yml b/docker/app-docker-compose.yml index 3b9d6d60..f51cc7bf 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,7 @@ services: depends_on: - postgres - wiremock + - kafka deploy: replicas: ${MODULE_REPLICAS:-1} restart_policy: diff --git a/docker/infra-docker-compose.yml b/docker/infra-docker-compose.yml index 005e7f10..de07c3bd 100644 --- a/docker/infra-docker-compose.yml +++ b/docker/infra-docker-compose.yml @@ -45,6 +45,40 @@ services: networks: - mod-notes-local + kafka: + image: apache/kafka-native + networks: + - mod-notes-local + ports: + - ${KAFKA_EXTERNAL_PORT}:${KAFKA_EXTERNAL_PORT} + - ${KAFKA_PORT}:9093 + environment: + KAFKA_LISTENERS: CONTROLLER://localhost:9091,HOST://0.0.0.0:${KAFKA_EXTERNAL_PORT},DOCKER://0.0.0.0:9093 + KAFKA_ADVERTISED_LISTENERS: HOST://localhost:${KAFKA_EXTERNAL_PORT},DOCKER://kafka:9093 + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,DOCKER:PLAINTEXT,HOST:PLAINTEXT + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9091 + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_INTER_BROKER_LISTENER_NAME: DOCKER + KAFKA_LOG_DIRS: /tmp/kraft-combined-logs + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true' + + kafka-ui: + image: provectuslabs/kafka-ui:latest + networks: + - mod-notes-local + ports: + - ${KAFKA_UI_PORT}:8080 + environment: + DYNAMIC_CONFIG_ENABLED: 'true' + KAFKA_CLUSTERS_0_NAME: local + KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9093 + depends_on: + - kafka + + networks: mod-notes-local: driver: bridge diff --git a/pom.xml b/pom.xml index 174c78f7..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 @@ -78,6 +84,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 +166,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/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/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/integration/kafka/NoteEventProducer.java b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java new file mode 100644 index 00000000..0b6c2c7c --- /dev/null +++ b/src/main/java/org/folio/notes/integration/kafka/NoteEventProducer.java @@ -0,0 +1,85 @@ +package org.folio.notes.integration.kafka; + +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; +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.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. + * + *

The topic name follows the FOLIO convention {@code {env}.{tenant}.notes.note} (for example + * {@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 = "eventType"; + private static final String EVENT_DOMAIN_HEADER = "domain"; + + private final KafkaTemplate> kafkaTemplate; + private final FolioExecutionContext context; + private final String noteTopic; + + public NoteEventProducer(KafkaTemplate> kafkaTemplate, + FolioExecutionContext context, + FolioKafkaProperties kafkaProperties) { + this.kafkaTemplate = kafkaTemplate; + this.context = context; + this.noteTopic = kafkaProperties.getTopics().getFirst().getName(); + } + + /** + * 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 = KafkaUtils.getTenantTopicName(noteTopic, context.getTenantId()); + try { + log.debug("publish:: sending Note event [topic: {}, id: {}, type: {}]", topic, id, event.getType()); + 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 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; + } + + 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/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); + } +} 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 8de228eb..00c91e44 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 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 5f959253..a85d449a 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,33 @@ folio: - target response: limit: ${MAX_RECORDS_COUNT:1000} + kafka: + topics: + - name: notes.note + numPartitions: ${KAFKA_NOTE_TOPIC_PARTITIONS:1} + replicationFactor: ${KAFKA_NOTE_TOPIC_REPLICATION_FACTOR:} # 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 +117,4 @@ server: logging: level: - org.folio.spring.filter.IncomingRequestLoggingFilter: DEBUG \ No newline at end of file + org.folio.spring.filter.IncomingRequestLoggingFilter: DEBUG 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/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/service/NoteTenantServiceTest.java b/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java index eb6c3816..d6fac314 100644 --- a/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java +++ b/src/test/java/org/folio/notes/service/NoteTenantServiceTest.java @@ -3,19 +3,31 @@ 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.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; 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 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..5b47d491 --- /dev/null +++ b/src/test/java/org/folio/notes/support/TestKafkaConsumer.java @@ -0,0 +1,129 @@ +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 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.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(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 { + + 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(String key) { + return poll(DEFAULT_POLL_TIMEOUT).stream() + .filter(e -> Objects.equals(e.key(), key)).findFirst() + .orElse(null); + } + + /** + * 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 List> poll(Duration timeout) { + var holder = new AtomicReference>>(); + await().pollInterval(POLL_INTERVAL).atMost(timeout) + .untilAsserted(() -> { + List> events = new ArrayList<>(); + records.drainTo(events); + holder.set(events); + }); + return holder.get(); + } + + @Override + public void close() { + container.stop(); + } +} + + +