Merge branch 'master' into feature/ecs-member-reindex

# Conflicts:
#	src/main/java/org/folio/search/service/reindex/jdbc/CallNumberRepository.java
#	src/test/java/org/folio/search/service/reindex/jdbc/CallNumberRepositoryIT.java

Author: viacheslavkol

NEWS.md

@@ -23,9 +23,11 @@
   * Use unique consumer group per instance for inventory classification-type and call-number-type topics ([MSEARCH-1092](https://folio-org.atlassian.net/browse/MSEARCH-1092))
   * Remove identifiers from keyword alias for instance and authority mappings ([MSEARCH-1118](https://folio-org.atlassian.net/browse/MSEARCH-1118))
   * Change `all` query builder to use full term for multimatch search ([MSEARCH-1112](https://folio-org.atlassian.net/browse/MSEARCH-1112))
+  * Create a new listener for the CONSORTIUM_INSTANCE_SHARING_COMPLETE event to update the call number’s last_updated_date when an instance is shared to the central tenant ([MSEARCH-1168](https://folio-org.atlassian.net/browse/MSEARCH-1168))
 * **Authority Search**
   * Implement two-stage Kafka processing with event aggregation for instance indexing ([MSEARCH-1157](https://folio-org.atlassian.net/browse/MSEARCH-1157))
   * Separate LCCN and Canceled LCCN identifiers search to lccn and canceledLccn options ([MSEARCH-1066](https://folio-org.atlassian.net/browse/MSEARCH-1066))
+  * Use Authority-Specific Identifier Types for LCCN Lookups ([MSEARCH-1193](https://folio-org.atlassian.net/browse/MSEARCH-1193))
 * **Classification Browse**
   * Add title and contributors to classification browse response ([MSEARCH-1045](https://folio-org.atlassian.net/browse/MSEARCH-1045))
   * Add id to classification browse response ([MSEARCH-1093](https://folio-org.atlassian.net/browse/MSEARCH-1093))

README.md

@@ -293,6 +293,7 @@ and [Cross-cluster replication](https://docs.aws.amazon.com/opensearch-service/l
 | KAFKA_AUTHORITIES_CONCURRENCY                      | 1                                                    | Custom number of kafka concurrent threads for authority message consuming.                                                                                                            |
 | KAFKA_LOCATION_CONCURRENCY                         | 1                                                    | Custom number of kafka concurrent threads for inventory.location, inventory.campus, inventory.institution and inventory.library message consuming.                                    |
 | KAFKA_LINKED_DATA_CONCURRENCY                      | 1                                                    | Custom number of kafka concurrent threads for linked data message consuming.                                                                                                          |
+| KAFKA_CONSORTIUM_INSTANCE_SHARING_COMPLETE_CONCURRENCY        | 1                                                    | Custom number of kafka concurrent threads for consortium instance sharing complete message consuming.                                                                                 |
 | KAFKA_REINDEX_RANGE_INDEX_CONCURRENCY              | 1                                                    | Custom number of kafka concurrent threads for `search.reindex.range-index` message consuming.                                                                                         |
 | KAFKA_REINDEX_RANGE_INDEX_TOPIC_PARTITIONS         | 16                                                   | Amount of partitions for `search.reindex.range-index` topic.                                                                                                                          |
 | KAFKA_REINDEX_RANGE_INDEX_TOPIC_REPLICATION_FACTOR | -                                                    | Replication factor for `search.reindex.range-index` topic.                                                                                                                            |

descriptors/ModuleDescriptor-template.json

@@ -652,6 +652,10 @@
       "id": "item-storage",
       "version": "11.0"
     },
+    {
+      "id": "authority-identifier-types",
+      "version": "1.0"
+    },
     {
       "id": "authority-reindex",
       "version": "0.1"
@@ -932,6 +936,7 @@
         "perms.users.item.post",
         "perms.users.assign.immutable",
         "perms.users.assign.mutable",
+        "authority-identifier-types.collection.get",
         "inventory-storage.instance.reindex.post",
         "authority-storage.authority.reindex.post",
         "inventory-storage.inventory-view.instances.collection.get",
@@ -1039,6 +1044,11 @@
         "value": "1",
         "description": "Custom number of kafka concurrent threads for authority message consuming."
       },
+      {
+        "name": "KAFKA_CONSORTIUM_INSTANCE_SHARING_COMPLETE_CONCURRENCY",
+        "value": "1",
+        "description": "Custom number of kafka concurrent threads for consortium instance sharing complete message consuming."
+      },
       {
         "name": "KAFKA_REINDEX_RANGE_INDEX_CONCURRENCY",
         "value": "1",

docker/kafka-init.sh

@@ -26,6 +26,7 @@ TOPICS=(
 "${ENV}.Default.linked-data.hub"
 "${ENV}.Default.search.reindex.range-index"
 "${ENV}.Default.inventory.reindex-records"
+"${ENV}.Default.ALL.CONSORTIUM_INSTANCE_SHARING_COMPLETE"
 )
 
 # Updated to use the full path to kafka-topics.sh

pom.xml

@@ -7,7 +7,7 @@
   <parent>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-parent</artifactId>
-    <version>4.0.3</version>
+    <version>4.0.4</version>
     <relativePath/>
   </parent>
 
@@ -55,7 +55,7 @@
 
     <!-- Plugins versions -->
     <folio-module-descriptor-validator.version>1.0.1</folio-module-descriptor-validator.version>
-    <maven-openapi-generator-plugin.version>7.20.0</maven-openapi-generator-plugin.version>
+    <maven-openapi-generator-plugin.version>7.21.0</maven-openapi-generator-plugin.version>
     <maven-copy-rename-plugin.version>1.0.1</maven-copy-rename-plugin.version>
     <maven-build-helper-plugin.version>3.6.1</maven-build-helper-plugin.version>
     <maven-clean-plugin.version>3.5.0</maven-clean-plugin.version>
@@ -313,7 +313,7 @@
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>testcontainers-elasticsearch</artifactId>
-      <version>2.0.3</version>
+      <version>2.0.4</version>
       <scope>test</scope>
     </dependency>
 

src/main/java/org/folio/search/client/InventoryReferenceDataClient.java

@@ -27,6 +27,7 @@ public interface InventoryReferenceDataClient {
   @Getter
   enum ReferenceDataType {
 
+    AUTHORITY_IDENTIFIER_TYPES("authority-identifier-types"),
     IDENTIFIER_TYPES("identifier-types"),
     ALTERNATIVE_TITLE_TYPES("alternative-title-types"),
     CALL_NUMBER_TYPES("call-number-types"),

src/main/java/org/folio/search/configuration/kafka/InstanceSharingCompleteEventKafkaConfiguration.java

@@ -0,0 +1,29 @@
+package org.folio.search.configuration.kafka;
+
+import java.util.Collections;
+import lombok.RequiredArgsConstructor;
+import org.folio.search.model.event.InstanceSharingCompleteEvent;
+import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.listener.CommonErrorHandler;
+import org.springframework.kafka.support.serializer.JacksonJsonDeserializer;
+
+@Configuration
+@RequiredArgsConstructor
+public class InstanceSharingCompleteEventKafkaConfiguration extends KafkaConfiguration {
+
+  private final KafkaProperties kafkaProperties;
+
+  @Bean
+  public ConcurrentKafkaListenerContainerFactory<String, InstanceSharingCompleteEvent>
+    instanceSharingCompleteListenerContainerFactory(CommonErrorHandler commonErrorHandler) {
+
+    var factory = new ConcurrentKafkaListenerContainerFactory<String, InstanceSharingCompleteEvent>();
+    var deserializer = new JacksonJsonDeserializer<>(InstanceSharingCompleteEvent.class, false);
+    factory.setConsumerFactory(getConsumerFactory(deserializer, kafkaProperties, Collections.emptyMap()));
+    factory.setCommonErrorHandler(commonErrorHandler);
+    return factory;
+  }
+}

src/main/java/org/folio/search/integration/message/KafkaMessageListener.java

@@ -13,14 +13,18 @@
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.Strings;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.logging.log4j.message.FormattedMessage;
 import org.folio.search.domain.dto.ResourceEvent;
 import org.folio.search.model.event.IndexInstanceEvent;
+import org.folio.search.model.event.InstanceSharingCompleteEvent;
 import org.folio.search.model.types.ResourceType;
 import org.folio.search.service.ResourceService;
 import org.folio.search.service.config.ConfigSynchronizationService;
+import org.folio.search.service.consortium.ConsortiumTenantProvider;
+import org.folio.search.service.reindex.jdbc.CallNumberRepository;
 import org.folio.search.utils.KafkaConstants;
 import org.folio.search.utils.SearchConverterUtils;
 import org.folio.spring.service.SystemUserScopedExecutionService;
@@ -43,6 +47,8 @@ public class KafkaMessageListener {
   private final ConfigSynchronizationService configSynchronizationService;
   private final KafkaTemplate<String, IndexInstanceEvent> instanceEventProducer;
   private final InstanceEventMapper instanceEventMapper;
+  private final CallNumberRepository callNumberRepository;
+  private final ConsortiumTenantProvider consortiumTenantProvider;
 
   /**
    * Handles instance events and indexes them by id.
@@ -172,6 +178,30 @@ public void handleLinkedDataEvents(List<ConsumerRecord<String, ResourceEvent>> c
     indexResources(batch, resourceService::indexResources);
   }
 
+  @KafkaListener(
+    id = KafkaConstants.INSTANCE_SHARING_COMPLETE_LISTENER_ID,
+    containerFactory = "instanceSharingCompleteListenerContainerFactory",
+    groupId = "#{folioKafkaProperties.listener['instance-sharing-complete'].groupId}",
+    concurrency = "#{folioKafkaProperties.listener['instance-sharing-complete'].concurrency}",
+    topicPattern = "#{folioKafkaProperties.listener['instance-sharing-complete'].topicPattern}")
+  public void handleInstanceSharingCompleteEvent(InstanceSharingCompleteEvent instanceSharingCompleteEvent) {
+    log.info("Processing consortium instance sharing complete event from Kafka ");
+
+    if (InstanceSharingCompleteEvent.Status.COMPLETE.equals(instanceSharingCompleteEvent.getStatus())
+      && StringUtils.isEmpty(instanceSharingCompleteEvent.getError())) {
+
+      var tenant = instanceSharingCompleteEvent.getTargetTenantId();
+      executionService.executeSystemUserScoped(tenant, () -> {
+        if (consortiumTenantProvider.isCentralTenant(tenant)) {
+          log.info("handleInstanceSharingCompleteEvent: Updating lastUpdatedDate for call numbers of instance "
+            + "in central tenant {}", tenant);
+          callNumberRepository.updateLastUpdatedDate(instanceSharingCompleteEvent.getInstanceIdentifier());
+        }
+        return null;
+      });
+    }
+  }
+
   private void indexResources(List<ResourceEvent> batch, Consumer<List<ResourceEvent>> indexConsumer) {
     var batchByTenant = batch.stream().collect(Collectors.groupingBy(ResourceEvent::getTenant));
 

src/main/java/org/folio/search/model/client/CqlQueryParam.java

@@ -7,6 +7,7 @@
 public enum CqlQueryParam {
 
   ID("id"),
+  CODE("code"),
   NAME("name"),
   SOURCE("source"),
   HOLDINGS_ID("holdings.id");

src/main/java/org/folio/search/model/event/InstanceSharingCompleteEvent.java

@@ -0,0 +1,29 @@
+package org.folio.search.model.event;
+
+import com.fasterxml.jackson.annotation.JsonValue;
+import java.util.UUID;
+import lombok.Data;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@Data
+public class InstanceSharingCompleteEvent {
+  private UUID id;
+  private String instanceIdentifier;
+  private String sourceTenantId;
+  private String targetTenantId;
+  private Status status;
+  private String error;
+
+  @Getter
+  @RequiredArgsConstructor
+  public enum Status {
+
+    IN_PROGRESS("IN_PROGRESS"),
+    COMPLETE("COMPLETE"),
+    ERROR("ERROR");
+
+    @JsonValue
+    private final String value;
+  }
+}