MODLD-686: working proto

Author: PBobylev

src/main/java/org/folio/linked/data/configuration/kafka/KafkaListenerConfiguration.java

@@ -9,7 +9,8 @@
 import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.common.serialization.StringDeserializer;
-import org.folio.linked.data.domain.dto.InventoryDomainEvent;
+import org.folio.linked.data.domain.dto.InventoryDomainHoldingEvent;
+import org.folio.linked.data.domain.dto.InventoryDomainItemEvent;
 import org.folio.linked.data.domain.dto.InventoryInstanceEvent;
 import org.folio.linked.data.domain.dto.SourceRecordDomainEvent;
 import org.folio.spring.tools.kafka.FolioKafkaProperties;
@@ -63,10 +64,27 @@ public ConsumerFactory<String, InventoryInstanceEvent> inventoryInstanceEventCon
   }
 
   @Bean
-  public ConcurrentKafkaListenerContainerFactory<String, InventoryDomainEvent> inventoryDomainEventListenerFactory(
-    ConsumerFactory<String, InventoryDomainEvent> inventoryDomainEventListenerFactory
+  public ConsumerFactory<String, InventoryDomainHoldingEvent> holdingEventConsumerFactory() {
+    return errorHandlingConsumerFactory(InventoryDomainHoldingEvent.class);
+  }
+
+  @Bean
+  public ConcurrentKafkaListenerContainerFactory<String, InventoryDomainHoldingEvent> holdingListenerFactory(
+    ConsumerFactory<String, InventoryDomainHoldingEvent> inventoryDomainEventConsumerFactory
+  ) {
+    return concurrentKafkaBatchListenerContainerFactory(inventoryDomainEventConsumerFactory);
+  }
+
+  @Bean
+  public ConsumerFactory<String, InventoryDomainItemEvent> itemEventConsumerFactory() {
+    return errorHandlingConsumerFactory(InventoryDomainItemEvent.class);
+  }
+
+  @Bean
+  public ConcurrentKafkaListenerContainerFactory<String, InventoryDomainItemEvent> itemListenerFactory(
+    ConsumerFactory<String, InventoryDomainItemEvent> inventoryDomainEventConsumerFactory
   ) {
-    return concurrentKafkaBatchListenerContainerFactory(inventoryDomainEventListenerFactory);
+    return concurrentKafkaBatchListenerContainerFactory(inventoryDomainEventConsumerFactory);
   }
 
   private <K, V> ConcurrentKafkaListenerContainerFactory<K, V> concurrentKafkaBatchListenerContainerFactory(

src/main/java/org/folio/linked/data/integration/kafka/listener/InventoryDomainEventListener.java

@@ -6,7 +6,8 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.folio.linked.data.domain.dto.InventoryDomainEvent;
+import org.folio.linked.data.domain.dto.InventoryDomainHoldingEvent;
+import org.folio.linked.data.domain.dto.InventoryDomainItemEvent;
 import org.springframework.context.annotation.Profile;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
@@ -22,26 +23,26 @@ public class InventoryDomainEventListener {
 
   @KafkaListener(
     id = ITEM_DOMAIN_EVENT_LISTENER,
-    containerFactory = "inventoryDomainEventListenerFactory",
+    containerFactory = "itemListenerFactory",
     groupId = "#{folioKafkaProperties.listener['inventory-item-domain-event'].groupId}",
     concurrency = "#{folioKafkaProperties.listener['inventory-item-domain-event'].concurrency}",
     topicPattern = "#{folioKafkaProperties.listener['inventory-item-domain-event'].topicPattern}")
-  public void handleInventoryItemDomainEvent(List<ConsumerRecord<String, InventoryDomainEvent>> consumerRecords) {
+  public void handleInventoryItemDomainEvent(List<ConsumerRecord<String, InventoryDomainItemEvent>> consumerRecords) {
     consumerRecords.forEach(cr -> processRecord(cr, "Item"));
   }
 
   @KafkaListener(
     id = HOLDING_DOMAIN_EVENT_LISTENER,
-    containerFactory = "inventoryDomainEventListenerFactory",
+    containerFactory = "holdingListenerFactory",
     groupId = "#{folioKafkaProperties.listener['inventory-holding-domain-event'].groupId}",
     concurrency = "#{folioKafkaProperties.listener['inventory-holding-domain-event'].concurrency}",
     topicPattern = "#{folioKafkaProperties.listener['inventory-holding-domain-event'].topicPattern}")
-  public void handleInventoryHoldingDomainEvent(List<ConsumerRecord<String, InventoryDomainEvent>> consumerRecords) {
+  public void handleInventoryHoldingDomainEvent(List<ConsumerRecord<String, InventoryDomainHoldingEvent>> consumerRecords) {
     consumerRecords.forEach(cr -> processRecord(cr, "Holding"));
   }
 
-  private void processRecord(ConsumerRecord<String, InventoryDomainEvent> event, String entityType) {
-    log.info("Received [{}] Domain Event: [key {}], value [{}]", entityType, event.key(), event.value());
+  private void processRecord(Object event, String entityType) {
+    log.info("Received [{}] Domain Event: {}", entityType, event);
   }
 
 }

src/main/resources/swagger.api/folio-modules.yaml

@@ -26,5 +26,7 @@ components:
       $ref: folio-modules/srs/sourceRecordDomainEvent.json
     inventoryInstanceEvent:
       $ref: folio-modules/inventory/inventoryInstanceEvent.json
-    inventoryDomainEvent:
-      $ref: folio-modules/inventory/storage/inventoryDomainEvent.json
+    inventoryDomainHoldingEvent:
+      $ref: folio-modules/inventory/storage/inventoryDomainHoldingEvent.json
+    inventoryDomainItemEvent:
+      $ref: folio-modules/inventory/storage/inventoryDomainItemEvent.json

src/main/resources/swagger.api/folio-modules/inventory/storage/inventoryDomainEventPayload.json

@@ -1,6 +1,6 @@
 {
   "$schema": "http://json-schema.org/draft-04/schema#",
-  "description": "Inventory domain event data model",
+  "description": "Inventory Holding domain event data model",
   "type": "object",
   "additionalProperties": false,
   "properties": {
@@ -12,13 +12,13 @@
       "description": "Event timestamp",
       "type": "string"
     },
-    "oldEntity": {
+    "old": {
       "description": "Old entity",
-      "$ref": "inventoryDomainEventPayload.json"
+      "$ref": "holdingsRecord.json"
     },
-    "newEntity": {
+    "new": {
       "description": "New entity",
-      "$ref": "inventoryDomainEventPayload.json"
+      "$ref": "holdingsRecord.json"
     },
     "type": {
       "type": "string",

src/main/resources/swagger.api/folio-modules/inventory/storage/inventoryDomainEvent.json renamed to src/main/resources/swagger.api/folio-modules/inventory/storage/inventoryDomainHoldingEvent.json

@@ -0,0 +1,53 @@
+{
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "description": "Inventory Item domain event data model",
+  "type": "object",
+  "additionalProperties": false,
+  "properties": {
+    "eventId": {
+      "description": "UUID",
+      "$ref": "../../common/uuid.json"
+    },
+    "eventTs": {
+      "description": "Event timestamp",
+      "type": "string"
+    },
+    "old": {
+      "description": "Old entity",
+      "$ref": "inventoryItem.json"
+    },
+    "new": {
+      "description": "New entity",
+      "$ref": "inventoryItem.json"
+    },
+    "type": {
+      "type": "string",
+      "enum": [
+        "UPDATE",
+        "DELETE",
+        "CREATE",
+        "DELETE_ALL",
+        "REINDEX",
+        "ITERATE",
+        "MIGRATION"
+      ],
+      "description": "Inventory domain event type"
+    },
+    "eventMetadata": {
+      "type": "object",
+      "$ref": "../../common/eventMetadata.json",
+      "description": "Event metadata"
+    },
+    "tenant": {
+      "type": "string",
+      "description": "Tenant id"
+    }
+  },
+  "excludedFromEqualsAndHashCode": [
+    "eventMetadata"
+  ],
+  "required": [
+    "id",
+    "eventType"
+  ]
+}

src/main/resources/swagger.api/folio-modules/inventory/storage/inventoryDomainItemEvent.json

@@ -1,21 +1,27 @@
 package org.folio.linked.data.integration.kafka.listener;
 
 import static org.folio.linked.data.test.TestUtil.TENANT_ID;
+import static org.folio.linked.data.test.TestUtil.awaitAndAssert;
 import static org.folio.linked.data.test.kafka.KafkaEventsTestDataFixture.getInventoryDomainEventHoldingSampleProducerRecord;
 import static org.folio.linked.data.test.kafka.KafkaEventsTestDataFixture.getInventoryDomainEventItemSampleProducerRecord;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
 
 import org.folio.linked.data.e2e.base.IntegrationTest;
 import org.folio.spring.tools.kafka.KafkaAdminService;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
 
 @IntegrationTest
 class InventoryDomainEventListenerIT {
 
   @Autowired
   private KafkaTemplate<String, String> eventKafkaTemplate;
+  @MockitoSpyBean
+  private InventoryDomainEventListener inventoryDomainEventListener;
 
   @BeforeAll
   static void setup(@Autowired KafkaAdminService kafkaAdminService) {
@@ -43,7 +49,7 @@ void shouldHandleHoldingDomainEvent() {
     eventKafkaTemplate.send(eventProducerRecord);
 
     // then
-    // check logs, that's enough for Spike
+    awaitAndAssert(() -> verify(inventoryDomainEventListener).handleInventoryHoldingDomainEvent(any()));
   }
 
 }

src/test/java/org/folio/linked/data/integration/kafka/listener/InventoryDomainEventListenerIT.java

@@ -18,6 +18,10 @@ folio:
         num-partitions: 1
       - name: inventory.instance
         num-partitions: 1
+      - name: inventory.item
+        num-partitions: 1
+      - name: inventory.holdings-record
+        num-partitions: 1
 
 mod-linked-data:
   reindex:

src/test/resources/application-test.yml

@@ -0,0 +1,31 @@
+{
+	"new": {
+		"id": "d48996b4-170e-43e4-8f4d-cafaa75e7b52",
+		"_version": 1,
+		"sourceId": "f32d531e-df79-46b3-8932-cdd35f7a2264",
+		"hrid": "oho00056184712",
+		"formerIds": [],
+		"instanceId": "c1ef011b-f97e-4ed0-aec3-636cd2f64107",
+		"permanentLocationId": "574326a1-f240-4222-82b3-21ccd6d6e5e5",
+		"effectiveLocationId": "574326a1-f240-4222-82b3-21ccd6d6e5e5",
+		"electronicAccess": [],
+		"callNumberTypeId": "95467209-6d7b-468b-94df-0f5d7ad2747d",
+		"callNumber": "1986",
+		"administrativeNotes": [],
+		"notes": [],
+		"holdingsStatements": [],
+		"holdingsStatementsForIndexes": [],
+		"holdingsStatementsForSupplements": [],
+		"statisticalCodeIds": [],
+		"metadata": {
+			"createdDate": "2025-04-11T11:39:26.882+00:00",
+			"createdByUserId": "cee695b3-8732-447c-90d2-534f1fa16991",
+			"updatedDate": "2025-04-11T11:39:26.882+00:00",
+			"updatedByUserId": "cee695b3-8732-447c-90d2-534f1fa16991"
+		}
+	},
+	"type": "CREATE",
+	"tenant": "fs09000000",
+	"eventId": "38daa6bb-76cc-4492-a1de-80f3a4728182",
+	"eventTs": 1744371567012
+}

src/test/resources/samples/inventoryDomainEventHolding.json

@@ -0,0 +1,62 @@
+{
+	"new": {
+		"instanceId": "c1ef011b-f97e-4ed0-aec3-636cd2f64107",
+		"id": "bafdd67b-f8f5-45e7-971a-8ab0fb76b03a",
+		"_version": 1,
+		"hrid": "oit00160053757",
+		"holdingsRecordId": "d48996b4-170e-43e4-8f4d-cafaa75e7b52",
+		"formerIds": [],
+		"displaySummary": "Enum display summary",
+		"barcode": "Barcode",
+		"effectiveShelvingOrder": "41509 VOLUME VALUE ENUMERATION VALUE CHRONOLOGY VALUE",
+		"itemLevelCallNumber": "1509",
+		"effectiveCallNumberComponents": {
+			"callNumber": "1509",
+			"typeId": "95467209-6d7b-468b-94df-0f5d7ad2747d"
+		},
+		"volume": "Volume value",
+		"enumeration": "Enumeration value",
+		"chronology": "Chronology value",
+		"yearCaption": [
+			"Year, caption"
+		],
+		"itemDamagedStatusId": "516b82eb-1f19-4a63-8c48-8f1a3e9ff311",
+		"itemDamagedStatusDate": "2025-04-11",
+		"administrativeNotes": [],
+		"notes": [
+			{
+				"itemNoteTypeId": "f618d7d8-0ab2-40ed-b76a-9a6cff141643",
+				"note": "Action note",
+				"staffOnly": false
+			},
+			{
+				"itemNoteTypeId": "2864ca78-7d78-4a90-8ff6-13a4e30128b4",
+				"note": "Note",
+				"staffOnly": false
+			}
+		],
+		"circulationNotes": [],
+		"status": {
+			"name": "Available",
+			"date": "2025-04-11T11:44:08.630+00:00"
+		},
+		"materialTypeId": "025ba2c5-5e96-4667-a677-8186463aee69",
+		"permanentLoanTypeId": "668c4500-fc4e-4140-8a9f-872b54f5ef31",
+		"effectiveLocationId": "574326a1-f240-4222-82b3-21ccd6d6e5e5",
+		"electronicAccess": [],
+		"statisticalCodeIds": [],
+		"tags": {
+			"tagList": []
+		},
+		"metadata": {
+			"createdDate": "2025-04-11T11:44:08.629+00:00",
+			"createdByUserId": "cee695b3-8732-447c-90d2-534f1fa16991",
+			"updatedDate": "2025-04-11T11:44:08.629+00:00",
+			"updatedByUserId": "cee695b3-8732-447c-90d2-534f1fa16991"
+		}
+	},
+	"type": "CREATE",
+	"tenant": "fs09000000",
+	"eventId": "ae3aa2db-c71a-4a32-83d7-3d7534178f5a",
+	"eventTs": 1744371848744
+}