fix(consortium): ignore authority shadow copies while indexing

NEWS.md

@@ -12,6 +12,7 @@
 ### Bug fixes
 * Fix secure setup of system users by default ([MSEARCH-608](https://issues.folio.org/browse/MSEARCH-608))
 * Fix result filtering for items.effectiveLocationId ([MSEARCH-615](https://issues.folio.org/browse/MSEARCH-615))
+* Ignore authority shadow copies while indexing ([MSEARCH-638](https://issues.folio.org/browse/MSEARCH-638))
 
 ### Tech Dept
 * Description ([ISSUE_NUMBER](https://issues.folio.org/browse/ISSUE_NUMBER))

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

@@ -7,19 +7,22 @@
 import static org.folio.search.domain.dto.ResourceEventType.REINDEX;
 import static org.folio.search.utils.SearchConverterUtils.getEventPayload;
 import static org.folio.search.utils.SearchConverterUtils.getResourceEventId;
+import static org.folio.search.utils.SearchConverterUtils.getResourceSource;
 import static org.folio.search.utils.SearchUtils.AUTHORITY_RESOURCE;
 import static org.folio.search.utils.SearchUtils.CONTRIBUTOR_RESOURCE;
 import static org.folio.search.utils.SearchUtils.ID_FIELD;
 import static org.folio.search.utils.SearchUtils.INSTANCE_ID_FIELD;
 import static org.folio.search.utils.SearchUtils.INSTANCE_RESOURCE;
 import static org.folio.search.utils.SearchUtils.INSTANCE_SUBJECT_RESOURCE;
+import static org.folio.search.utils.SearchUtils.SOURCE_CONSORTIUM_PREFIX;
 
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.logging.log4j.message.FormattedMessage;
 import org.folio.search.domain.dto.ResourceEvent;
@@ -80,6 +83,7 @@ public void handleAuthorityEvents(List<ConsumerRecord<String, ResourceEvent>> co
     log.info("Processing authority events from Kafka [number of events: {}]", consumerRecords.size());
     var batch = consumerRecords.stream()
       .map(ConsumerRecord::value)
+      .filter(authority -> !StringUtils.startsWith(getResourceSource(authority), SOURCE_CONSORTIUM_PREFIX))
       .map(authority -> authority.resourceName(AUTHORITY_RESOURCE).id(getResourceEventId(authority)))
       .toList();
 

src/main/java/org/folio/search/utils/SearchConverterUtils.java

@@ -147,6 +147,16 @@ public static String getResourceEventId(Map<String, Object> eventPayload) {
     return getString(eventPayload, ID_FIELD);
   }
 
+  /**
+   * Returns resource event source field value from {@link ResourceEvent} object.
+   *
+   * @param event - resource event body to analyze
+   * @return event source field value as {@link String} object
+   */
+  public static String getResourceSource(ResourceEvent event) {
+    return getResourceSource(getEventPayload(event));
+  }
+
   /**
    * Returns resource event source field value from event payload {@link Map} object.
    *

src/test/java/org/folio/search/integration/KafkaMessageListenerTest.java

@@ -26,9 +26,11 @@
 import static org.folio.search.utils.TestUtils.resourceEvent;
 import static org.folio.search.utils.TestUtils.toMap;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -180,6 +182,17 @@ void handleAuthorityEvent_positive() {
     verify(batchProcessor).consumeBatchWithFallback(eq(expectedEvents), eq(KAFKA_RETRY_TEMPLATE_NAME), any(), any());
   }
 
+  @Test
+  void handleAuthorityEvent_positive_shouldSkipAuthorityShadowCopies() {
+    var payload = toMap(new Authority().id(RESOURCE_ID).source("CONSORTIUM-MARC"));
+
+    messageListener.handleAuthorityEvents(List.of(new ConsumerRecord<>(
+      inventoryAuthorityTopic(), 0, 0, RESOURCE_ID, resourceEvent(null, null, REINDEX, payload, null))));
+
+    verify(resourceService, never()).indexResources(anyList());
+    verify(batchProcessor, never()).consumeBatchWithFallback(any(), any(), any(), any());
+  }
+
   @Test
   void handleAuthorityEvent_negative_logFailedEvent() {
     var payload = toMap(new Authority().id(RESOURCE_ID).personalName("test"));