add taskScheduler for handleInstanceSharingCompleteEvents listener

SvitlanaKovalova1 · SvitlanaKovalova1 · commit cf6ecb649cde · 2026-03-31T03:05:13.000+03:00
diff --git a/README.md b/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_INSTANCE_SHARING_COMPLETE_CONCURRENCY        | 1                                                    | Custom number of kafka concurrent threads for 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.                                                                                                                            |
@@ -335,6 +336,8 @@ 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:
diff --git a/descriptors/ModuleDescriptor-template.json b/descriptors/ModuleDescriptor-template.json
@@ -1251,6 +1251,16 @@
         "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",
diff --git a/docker/kafka-init.sh b/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
diff --git a/src/main/java/org/folio/search/configuration/TaskSchedulerConfiguration.java b/src/main/java/org/folio/search/configuration/TaskSchedulerConfiguration.java
@@ -0,0 +1,21 @@
+package org.folio.search.configuration;
+
+import lombok.RequiredArgsConstructor;
+import org.folio.search.configuration.properties.TaskSchedulerProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+@Configuration
+@RequiredArgsConstructor
+public class TaskSchedulerConfiguration {
+
+  @Bean
+  public ThreadPoolTaskScheduler instanceSharingCompleteTaskScheduler(TaskSchedulerProperties properties) {
+    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+    scheduler.setPoolSize(properties.getPoolSize());
+    scheduler.setThreadNamePrefix(properties.getThreadNamePrefix());
+    scheduler.initialize();
+    return scheduler;
+  }
+}
diff --git a/src/main/java/org/folio/search/configuration/properties/TaskSchedulerProperties.java b/src/main/java/org/folio/search/configuration/properties/TaskSchedulerProperties.java
@@ -0,0 +1,26 @@
+package org.folio.search.configuration.properties;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Data
+@Configuration
+@ConfigurationProperties("folio.task-scheduler")
+public class TaskSchedulerProperties {
+
+  /**
+   * The delay in milliseconds before the task scheduler starts executing tasks.
+   */
+  private long delayMs = 70000L;
+
+  /**
+   * The size of the thread pool for task scheduling.
+   */
+  private int poolSize = 2;
+
+  /**
+   * The prefix for the names of threads in the task scheduler.
+   */
+  private String threadNamePrefix = "mod-search-task-scheduler-";
+}
diff --git a/src/main/java/org/folio/search/integration/message/KafkaMessageListener.java b/src/main/java/org/folio/search/integration/message/KafkaMessageListener.java
@@ -19,6 +19,7 @@
 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;
@@ -52,7 +53,8 @@ public class KafkaMessageListener {
   private final InstanceEventMapper instanceEventMapper;
   private final CallNumberRepository callNumberRepository;
   private final ConsortiumTenantProvider consortiumTenantProvider;
-  private TaskScheduler taskScheduler;
+  private final TaskSchedulerProperties taskSchedulerProperties;
+  private final TaskScheduler instanceSharingCompleteTaskScheduler;
 
   /**
    * Handles instance events and indexes them by id.
@@ -192,20 +194,20 @@ public void handleInstanceSharingCompleteEvents(List<InstanceSharingCompleteEven
     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
-    taskScheduler.schedule(() -> {
+    instanceSharingCompleteTaskScheduler.schedule(() -> {
       var batch = instanceSharingCompleteEvents.stream()
         .filter(event -> InstanceSharingCompleteEvent.Status.COMPLETE.equals(event.getStatus())
           && StringUtils.isEmpty(event.getError()))
         .toList();
+
       var batchByTenant = batch.stream()
         .collect(Collectors.groupingBy(
           InstanceSharingCompleteEvent::getTargetTenantId,
           Collectors.mapping(InstanceSharingCompleteEvent::getInstanceIdentifier, Collectors.toList())
         ));
-      log.info("handleInstanceSharingCompleteEvents: Grouped {} instance sharing complete events by tenant",
-        batch.size());
+
       updateTenantIdForCentralInstances(batch, batchByTenant);
-    }, Instant.now().plusSeconds(70));
+    }, Instant.now().plusMillis(taskSchedulerProperties.getDelayMs()));
   }
 
   private void updateTenantIdForCentralInstances(List<InstanceSharingCompleteEvent> batch,
@@ -214,11 +216,9 @@ private void updateTenantIdForCentralInstances(List<InstanceSharingCompleteEvent
       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);
+            log.info("updateTenantIdForCentralInstances: Updating call number tenant_id for {} instances in "
+              + "central tenant {}", instanceIdentifiers.size(), tenant);
             callNumberRepository.updateTenantIdForCentralInstances(instanceIdentifiers, tenant);
-          } else {
-            log.info("updateTenantIdForCentralInstances: Skipping tenant {} as it is not a central tenant", tenant);
           }
         }, KafkaMessageListener::logFailedEvent);
       return null;
diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml
@@ -112,6 +112,10 @@ 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}
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
@@ -110,6 +110,10 @@ 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}
diff --git a/src/test/java/org/folio/indexing/IndexingInstanceCallNumberConsortiumIT.java b/src/test/java/org/folio/indexing/IndexingInstanceCallNumberConsortiumIT.java
@@ -13,7 +13,6 @@
 import static org.folio.support.base.ApiEndpoints.instanceSearchPath;
 import static org.folio.support.utils.TestUtils.randomId;
 
-import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import org.folio.search.domain.dto.Holding;
@@ -81,7 +80,7 @@ void shouldUpdateInstanceCallNumber_onInstanceSharing() {
     inventoryApi.shareInstance(CENTRAL_TENANT_ID, INSTANCE_ID);
 
     // then - fetch call number documents and check if tenant field changed to central and shared is true
-    awaitAssertion(() -> assertInstanceCallNumberTenantId(CENTRAL_TENANT_ID, true), Duration.ofSeconds(80));
+    awaitAssertion(() -> assertInstanceCallNumberTenantId(CENTRAL_TENANT_ID, true));
   }
 
   private static void assertInstanceCallNumberTenantId(String expectedTenantId, boolean shared) {
diff --git a/src/test/java/org/folio/search/integration/KafkaMessageListenerIT.java b/src/test/java/org/folio/search/integration/KafkaMessageListenerIT.java
@@ -31,8 +31,11 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.folio.search.configuration.RetryTemplateConfiguration;
+import org.folio.search.configuration.TaskSchedulerConfiguration;
 import org.folio.search.configuration.kafka.InstanceResourceEventKafkaConfiguration;
+import org.folio.search.configuration.kafka.InstanceSharingCompleteEventKafkaConfiguration;
 import org.folio.search.configuration.kafka.ResourceEventKafkaConfiguration;
+import org.folio.search.configuration.properties.TaskSchedulerProperties;
 import org.folio.search.domain.dto.ResourceEvent;
 import org.folio.search.integration.KafkaMessageListenerIT.KafkaListenerTestConfiguration;
 import org.folio.search.integration.message.FolioMessageBatchProcessor;
@@ -71,6 +74,7 @@
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.resilience.annotation.EnableResilientMethods;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.test.context.bean.override.mockito.MockitoBean;
 
 @Log4j2
@@ -112,6 +116,8 @@ class KafkaMessageListenerIT {
   private CallNumberRepository callNumberRepository;
   @MockitoBean
   private ConsortiumTenantProvider consortiumTenantProvider;
+  @MockitoBean
+  private TaskScheduler instanceSharingCompleteTaskScheduler;
   @Captor
   private ArgumentCaptor<ProducerRecord<String, IndexInstanceEvent>> producerRecordCaptor;
 
@@ -207,7 +213,9 @@ private static ResourceEvent instanceEvent() {
   @Import({
     InstanceResourceEventKafkaConfiguration.class, ResourceEventKafkaConfiguration.class,
     KafkaAutoConfiguration.class, FolioMessageBatchProcessor.class,
-    RetryTemplateConfiguration.class, ResourceEventBatchInterceptor.class
+    RetryTemplateConfiguration.class, ResourceEventBatchInterceptor.class,
+    InstanceSharingCompleteEventKafkaConfiguration.class, TaskSchedulerConfiguration.class,
+    TaskSchedulerProperties.class
   })
   static class KafkaListenerTestConfiguration {
 
diff --git a/src/test/java/org/folio/support/base/BaseIntegrationTest.java b/src/test/java/org/folio/support/base/BaseIntegrationTest.java
@@ -44,7 +44,6 @@
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 import java.io.IOException;
-import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -196,13 +195,6 @@ public static void awaitAssertion(ThrowingRunnable runnable) {
       .untilAsserted(runnable);
   }
 
-  public static void awaitAssertion(ThrowingRunnable runnable, Duration timeout) {
-    Awaitility.await()
-      .atMost(timeout)
-      .pollInterval(ONE_HUNDRED_MILLISECONDS)
-      .untilAsserted(runnable);
-  }
-
   @SneakyThrows
   protected static ResultActions attemptPost(String uri, Object body) {
     return mockMvc.perform(post(uri)
diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml
@@ -77,6 +77,10 @@ folio:
     core-pool-size: 2
     max-pool-size: 2
     queue-capacity: 500
+  task-scheduler:
+    pool-size: 2
+    thread-name-prefix: ${folio.environment}-task-scheduler-
+    delay-ms: 2000
   kafka:
     retry-interval-ms: 100
     retry-delivery-attempts: 2

Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,7 @@ TOPICS=(`
`26`	`26`	`"${ENV}.Default.linked-data.hub"`
`27`	`27`	`"${ENV}.Default.search.reindex.range-index"`
`28`	`28`	`"${ENV}.Default.inventory.reindex-records"`
	`29`	`+"${ENV}.Default.ALL.CONSORTIUM_INSTANCE_SHARING_COMPLETE"`
`29`	`30`	`)`
`30`	`31`
`31`	`32`	`# Updated to use the full path to kafka-topics.sh`