feat(instance-sharing): 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

Author: SvitlanaKovalova1

NEWS.md

@@ -22,7 +22,7 @@
   * 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 tenantId instead of performing this update within InstanceChildrenResourceService.persistChildren ([MSEARCH-1168](https://folio-org.atlassian.net/browse/MSEARCH-1168))
+  * 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))

README.md

@@ -336,8 +336,6 @@ and [Cross-cluster replication](https://docs.aws.amazon.com/opensearch-service/l
 | INSTANCE_CHILDREN_INDEX_DELAY_MS                   | 60000                                                | Defines the delay for scheduler that indexes subjects/contributors/classifications/call-numbers in a background                                                                       |
 | SUB_RESOURCE_BATCH_SIZE                            | 100                                                  | Defines number of sub-resources to process at a time during background indexing                                                                                                       |
 | STALE_LOCK_THRESHOLD_MS                            | 600000                                               | Threshold to consider a sub-resource lock as stale and eligible for release                                                                                                           |
-| TASK_SCHEDULER_POOL_SIZE                           | 2                                                    | The number of threads to keep in the task scheduler pool.                                                                                                                             |
-| TASK_SCHEDULER_DELAY_MS                            | 70000                                                | The delay in ms for scheduled tasks.                                                                                                                                                  |
 
 The module uses system user to communicate with other modules from Kafka consumers.
 For production deployments you MUST specify the password for this system user via env variable:

descriptors/ModuleDescriptor-template.json

@@ -1256,16 +1256,6 @@
         "value": "500",
         "description": "The capacity of the queue."
       },
-      {
-        "name": "TASK_SCHEDULER_POOL_SIZE",
-        "value": "2",
-        "description": "The number of threads to keep in the task scheduler pool."
-      },
-      {
-        "name": "TASK_SCHEDULER_DELAY_MS",
-        "value": "70000",
-        "description": "The delay in ms for scheduled tasks."
-      },
       {
         "name": "SEARCH_QUERY_TIMEOUT",
         "value": "25s",

src/main/java/org/folio/search/configuration/TaskSchedulerConfiguration.java

@@ -7,7 +7,6 @@
 import static org.folio.search.utils.SearchConverterUtils.getResourceSource;
 import static org.folio.search.utils.SearchUtils.SOURCE_CONSORTIUM_PREFIX;
 
-import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
@@ -19,7 +18,6 @@
 import org.apache.commons.lang3.Strings;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.logging.log4j.message.FormattedMessage;
-import org.folio.search.configuration.properties.TaskSchedulerProperties;
 import org.folio.search.domain.dto.ResourceEvent;
 import org.folio.search.model.event.IndexInstanceEvent;
 import org.folio.search.model.event.InstanceSharingCompleteEvent;
@@ -34,7 +32,6 @@
 import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.scheduling.TaskScheduler;
 import org.springframework.stereotype.Component;
 
 /**
@@ -53,8 +50,6 @@ public class KafkaMessageListener {
   private final InstanceEventMapper instanceEventMapper;
   private final CallNumberRepository callNumberRepository;
   private final ConsortiumTenantProvider consortiumTenantProvider;
-  private final TaskSchedulerProperties taskSchedulerProperties;
-  private final TaskScheduler instanceSharingCompleteTaskScheduler;
 
   /**
    * Handles instance events and indexes them by id.
@@ -192,33 +187,30 @@ public void handleLinkedDataEvents(List<ConsumerRecord<String, ResourceEvent>> c
     topicPattern = "#{folioKafkaProperties.listener['instance-sharing-complete'].topicPattern}")
   public void handleInstanceSharingCompleteEvents(List<InstanceSharingCompleteEvent> instanceSharingCompleteEvents) {
     log.info("Processing instance sharing complete events from Kafka [number of events: {}]",
-      instanceSharingCompleteEvents.size());
-    // wait until instance children processing is completed and the call number records are available for update
-    instanceSharingCompleteTaskScheduler.schedule(() -> {
-      var batch = instanceSharingCompleteEvents.stream()
+        instanceSharingCompleteEvents.size());
+
+    var batch = instanceSharingCompleteEvents.stream()
         .filter(event -> InstanceSharingCompleteEvent.Status.COMPLETE.equals(event.getStatus())
-          && StringUtils.isEmpty(event.getError()))
+            && StringUtils.isEmpty(event.getError()))
         .toList();
 
-      var batchByTenant = batch.stream()
+    var batchByTenant = batch.stream()
         .collect(Collectors.groupingBy(
-          InstanceSharingCompleteEvent::getTargetTenantId,
-          Collectors.mapping(InstanceSharingCompleteEvent::getInstanceIdentifier, Collectors.toList())
+            InstanceSharingCompleteEvent::getTargetTenantId,
+            Collectors.mapping(InstanceSharingCompleteEvent::getInstanceIdentifier, Collectors.toList())
         ));
-
-      updateTenantIdForCentralInstances(batch, batchByTenant);
-    }, Instant.now().plusMillis(taskSchedulerProperties.getDelayMs()));
+    updateCalNumbersLastUpdatedDate(batch, batchByTenant);
   }
 
-  private void updateTenantIdForCentralInstances(List<InstanceSharingCompleteEvent> batch,
-                                                 Map<String, List<String>> batchByTenant) {
+  private void updateCalNumbersLastUpdatedDate(List<InstanceSharingCompleteEvent> batch,
+                                               Map<String, List<String>> batchByTenant) {
     batchByTenant.forEach((tenant, instanceIdentifiers) -> executionService.executeSystemUserScoped(tenant, () -> {
       folioMessageBatchProcessor.consumeBatchWithFallback(batch, KAFKA_RETRY_TEMPLATE_NAME,
         event -> {
           if (consortiumTenantProvider.isCentralTenant(tenant)) {
-            log.info("updateTenantIdForCentralInstances: Updating call number tenant_id for {} instances in "
-              + "central tenant {}", instanceIdentifiers.size(), tenant);
-            callNumberRepository.updateTenantIdForCentralInstances(instanceIdentifiers, tenant);
+            log.info("updateCalNumbersLastUpdatedDate: Updating lastUpdatedDate for call numbers of {} instances "
+              + "in central tenant {}", instanceIdentifiers.size(), tenant);
+            callNumberRepository.updateLastUpdatedDate(instanceIdentifiers);
           }
         }, KafkaMessageListener::logFailedEvent);
       return null;

src/main/java/org/folio/search/configuration/properties/TaskSchedulerProperties.java

@@ -76,17 +76,12 @@ WHERE item_id IN (%2$s) %3$s
     WHERE id IN (SELECT * FROM deleted_ids);
     """;
 
-  private static final String UPDATE_TENANT_FOR_CENTRAL_QUERY = """
-    WITH updated_call_numbers AS (
-        UPDATE %1$s.instance_call_number
-        SET tenant_id = ?
-        WHERE instance_id IN (%2$s)
-          AND tenant_id != ?
-        RETURNING call_number_id
-    )
-    UPDATE %1$s.call_number
+  private static final String UPDATE_LAST_UPDATED_DATE_QUERY = """
+    UPDATE %1$s.call_number cn
     SET last_updated_date = CURRENT_TIMESTAMP
-    WHERE id IN (SELECT call_number_id FROM updated_call_numbers);
+    FROM %1$s.instance_call_number icn
+    WHERE icn.instance_id IN (%2$s)
+      AND cn.id = icn.call_number_id;
     """;
 
   private static final String SELECT_BY_UPDATED_QUERY = """
@@ -190,22 +185,15 @@ public void deleteByInstanceIds(List<String> itemIds, String tenantId) {
   }
 
   /**
-   * Updates tenant_id in instance_call_number records for the given instances to the central tenant.
-   * Used when an instance is shared to the central tenant - the background job processes the newly created
-   * central instance and this method updates any existing call number relations that still point to the member tenant.
-   * For the updated relations - also updates last_updated_date in call_number table
-   * to trigger reindexing of those call numbers.
+   * Updates last_updated_date in call_number records for the given instances to trigger reindexing
+   * of those call numbers when an instance is shared to the central tenant .
    *
-   * @param instanceIds     list of instance IDs whose call number relations should be updated
-   * @param centralTenantId the central tenant ID to set
+   * @param instanceIds list of instance IDs whose call number relations should be updated
    */
-  public void updateTenantIdForCentralInstances(List<String> instanceIds, String centralTenantId) {
-    var sql = UPDATE_TENANT_FOR_CENTRAL_QUERY.formatted(JdbcUtils.getSchemaName(context),
+  public void updateLastUpdatedDate(List<String> instanceIds) {
+    var sql = UPDATE_LAST_UPDATED_DATE_QUERY.formatted(JdbcUtils.getSchemaName(context),
       getParamPlaceholderForUuid(instanceIds.size()));
-    var params = Stream.of(List.of(centralTenantId), instanceIds, List.of(centralTenantId))
-      .flatMap(List::stream)
-      .toArray();
-    jdbcTemplate.update(sql, params);
+    jdbcTemplate.update(sql, instanceIds.toArray());
   }
 
   @Override

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

@@ -112,10 +112,6 @@ folio:
     max-pool-size: ${STREAM_ID_MAX_POOL_SIZE:2}
     queue-capacity: ${STREAM_ID_QUEUE_CAPACITY:500}
     job-expiration-days: ${STREAM_ID_JOB_EXPIRATION_DAYS:7}
-  task-scheduler:
-    pool-size: ${TASK_SCHEDULER_POOL_SIZE:2}
-    thread-name-prefix: ${folio.environment}-task-scheduler-
-    delay-ms: ${TASK_SCHEDULER_DELAY_MS:70000}
   kafka:
     retry-interval-ms: ${KAFKA_RETRY_INTERVAL_MS:2000}
     retry-delivery-attempts: ${KAFKA_RETRY_DELIVERY_ATTEMPTS:6}

src/main/java/org/folio/search/service/reindex/jdbc/CallNumberRepository.java

@@ -110,10 +110,6 @@ folio:
     max-pool-size: ${STREAM_ID_MAX_POOL_SIZE:2}
     queue-capacity: ${STREAM_ID_QUEUE_CAPACITY:500}
     job-expiration-days: ${STREAM_ID_JOB_EXPIRATION_DAYS:7}
-  task-scheduler:
-    pool-size: ${TASK_SCHEDULER_POOL_SIZE:2}
-    thread-name-prefix: ${folio.environment}-task-scheduler-
-    delay-ms: ${TASK_SCHEDULER_DELAY_MS:70000}
   kafka:
     retry-interval-ms: ${KAFKA_RETRY_INTERVAL_MS:2000}
     retry-delivery-attempts: ${KAFKA_RETRY_DELIVERY_ATTEMPTS:6}

src/main/resources/application-dev.yml

@@ -77,7 +77,7 @@ void tearDown() {
   void shouldUpdateInstanceCallNumber_onInstanceSharing() {
     // given
     createInstanceInMemberTenant(INSTANCE_ID, INSTANCE_TITLE, LOCATION_ID, CALL_NUMBER);
-    awaitAssertion(() -> assertInstanceCallNumberTenantId(MEMBER_TENANT_ID, false));
+    awaitAssertion(() -> assertInstanceCallNumberSharedState(false));
     // when - create instance in central tenant with the same instance id/title
     var centralInstance = new Instance().id(INSTANCE_ID).title(INSTANCE_TITLE).source("FOLIO");
     inventoryApi.createInstance(CENTRAL_TENANT_ID, centralInstance);
@@ -87,13 +87,13 @@ void shouldUpdateInstanceCallNumber_onInstanceSharing() {
     inventoryApi.updateInstance(MEMBER_TENANT_ID, memberInstance);
 
     // then - fetch call number documents for the instance and check if tenant field changed to member tenant id
-    awaitAssertion(() -> assertInstanceCallNumberTenantId(MEMBER_TENANT_ID, false));
+    awaitAssertion(() -> assertInstanceCallNumberSharedState(false));
 
     inventoryApi.shareInstance(CENTRAL_TENANT_ID, INSTANCE_ID, InstanceSharingCompleteEvent.Status.COMPLETE, "",
       MEMBER_TENANT_ID, CENTRAL_TENANT_ID);
 
-    // then - fetch call number documents and check if tenant field changed to central and shared is true
-    awaitAssertion(() -> assertInstanceCallNumberTenantId(CENTRAL_TENANT_ID, true));
+    // then - check that shared field is set to true
+    awaitAssertion(() -> assertInstanceCallNumberSharedState(true));
   }
 
   @ParameterizedTest(name = "{index} => status={0}, errorMessage={1}, targetTenant={2}")
@@ -107,25 +107,25 @@ void shouldNotUpdateInstanceCallNumber_onInvalidSharingEvent(
     // given
     createInstanceInMemberTenant(instanceId, title, locationId, callNumber);
 
-    awaitAssertion(() -> assertInstanceCallNumberTenantId(MEMBER_TENANT_ID, false));
+    awaitAssertion(() -> assertInstanceCallNumberSharedState(false));
 
     var centralInstance = new Instance().id(instanceId).title(title).source("FOLIO");
     inventoryApi.createInstance(CENTRAL_TENANT_ID, centralInstance);
 
     var memberInstance = new Instance().id(instanceId).title(title).source("CONSORTIUM-FOLIO");
     inventoryApi.updateInstance(MEMBER_TENANT_ID, memberInstance);
 
-    awaitAssertion(() -> assertInstanceCallNumberTenantId(MEMBER_TENANT_ID, false));
+    awaitAssertion(() -> assertInstanceCallNumberSharedState(false));
 
     // when
     inventoryApi.shareInstance(MEMBER_TENANT_ID, instanceId, status, errorMessage, MEMBER_TENANT_ID, targetTenantId);
 
-    // then check that call number document is not updated with central tenant id and shared is false
+    // then check that the shared field is not updated and remains false
     await()
       .pollDelay(Duration.ofSeconds(30))
       .atMost(ONE_MINUTE)
       .untilAsserted(() ->
-        assertInstanceCallNumberTenantId(MEMBER_TENANT_ID, false)
+        assertInstanceCallNumberSharedState(false)
       );
   }
 
@@ -142,7 +142,7 @@ private static Stream<Arguments> negativeSharingScenarios() {
         UUID.randomUUID().toString(), "title3", UUID.randomUUID().toString(), "call number3"));
   }
 
-  private static void assertInstanceCallNumberTenantId(String expectedTenantId, boolean shared) {
+  private static void assertInstanceCallNumberSharedState(boolean shared) {
     var hits = fetchAllDocuments(INSTANCE_CALL_NUMBER, CENTRAL_TENANT_ID);
     assertThat(hits).hasSize(1);
 
@@ -152,7 +152,7 @@ private static void assertInstanceCallNumberTenantId(String expectedTenantId, bo
     assertThat(instances)
       .hasSize(1)
       .allSatisfy(map -> assertThat(map)
-        .containsEntry("tenantId", expectedTenantId)
+        .containsEntry("tenantId", MEMBER_TENANT_ID)
         .containsEntry("shared", shared));
   }