MSEARCH-617 Ignore hard-delete domain authority events

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

@@ -13,6 +13,7 @@
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.folio.search.domain.dto.ResourceEvent;
+import org.folio.search.integration.ResourceChangeFilterStrategy;
 import org.folio.search.model.event.ConsortiumInstanceEvent;
 import org.folio.spring.config.properties.FolioEnvironment;
 import org.folio.spring.tools.kafka.FolioKafkaTopic;
@@ -51,6 +52,7 @@ public ConcurrentKafkaListenerContainerFactory<String, ResourceEvent> standardLi
     var factory = new ConcurrentKafkaListenerContainerFactory<String, ResourceEvent>();
     factory.setBatchListener(true);
     factory.setConsumerFactory(resourceEventConsumerFactory());
+    factory.setRecordFilterStrategy(new ResourceChangeFilterStrategy());
     return factory;
   }
 

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

@@ -0,0 +1,30 @@
+package org.folio.search.integration;
+
+import static org.folio.search.utils.SearchUtils.AUTHORITY_RESOURCE;
+
+import lombok.extern.log4j.Log4j2;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.folio.search.domain.dto.ResourceEvent;
+import org.folio.search.domain.dto.ResourceEventSubType;
+import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
+
+@Log4j2
+public class ResourceChangeFilterStrategy implements RecordFilterStrategy<String, ResourceEvent> {
+
+  @Override
+  public boolean filter(ConsumerRecord<String, ResourceEvent> consumerRecord) {
+    log.info("Processing resource event [id: {}]", consumerRecord.value().getId());
+    var resourceEvent = consumerRecord.value();
+    var resourceName = resourceEvent.getResourceName();
+    if (resourceName != null) {
+      log.info("Processing resource event [resourceName: {}]", resourceName);
+      if (resourceName.equals(AUTHORITY_RESOURCE)) {
+        if (resourceEvent.getSubType() == ResourceEventSubType.HARD_DELETE) {
+          log.info("Skip event. No need to process hard-delete event for authority resource");
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+}

src/main/resources/swagger.api/schemas/resourceEvent.json

@@ -11,6 +11,10 @@
       "description": "Resource event operation type",
       "$ref": "resourceEventType.json"
     },
+    "subType": {
+      "description": "Resource event delete operation type",
+      "$ref": "resourceEventSubType.json"
+    },
     "tenant": {
       "description": "Tenant id",
       "type": "string"

src/main/resources/swagger.api/schemas/resourceEventSubType.json

@@ -0,0 +1,6 @@
+{
+  "type": "string",
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "description": "Resource event delete operation type - one of [soft-delete, hard-delete]",
+  "enum": [ "SOFT_DELETE", "HARD_DELETE"]
+}

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

@@ -0,0 +1,79 @@
+package org.folio.search.integration;
+
+import static org.folio.search.utils.SearchUtils.AUTHORITY_RESOURCE;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.folio.search.domain.dto.ResourceEvent;
+import org.folio.search.domain.dto.ResourceEventSubType;
+import org.folio.spring.test.type.UnitTest;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@UnitTest
+@ExtendWith(MockitoExtension.class)
+public class ResourceChangeFilterStrategyTest {
+
+  private final ResourceChangeFilterStrategy filterStrategy = new ResourceChangeFilterStrategy();
+
+  @Mock
+  private ConsumerRecord<String, ResourceEvent> consumerRecord;
+
+  @Test
+  void shouldNotFilterResourceEventWithoutName() {
+    var event = createResourceEvent(ResourceEventSubType.HARD_DELETE, null);
+    mockConsumerRecord(event);
+
+    var actual = filterStrategy.filter(consumerRecord);
+
+    assertFalse(actual);
+  }
+
+  @ValueSource(strings = {"instance", "contributor", "instance_subject"})
+  @ParameterizedTest
+  void shouldNotFilterNonAuthResourceEvent(String resourceName) {
+    var event = createResourceEvent(ResourceEventSubType.HARD_DELETE, resourceName);
+    mockConsumerRecord(event);
+
+    var actual = filterStrategy.filter(consumerRecord);
+
+    assertFalse(actual);
+  }
+
+  @Test
+  void shouldFilterHardDeleteAuthResourceEvent() {
+    var event = createResourceEvent(ResourceEventSubType.HARD_DELETE, AUTHORITY_RESOURCE);
+    mockConsumerRecord(event);
+    var actual = filterStrategy.filter(consumerRecord);
+
+    assertTrue(actual);
+  }
+
+  @Test
+  void shouldNotFilterSoftDeleteAuthResourceEvent() {
+    var event = createResourceEvent(ResourceEventSubType.SOFT_DELETE, AUTHORITY_RESOURCE);
+    mockConsumerRecord(event);
+
+    var actual = filterStrategy.filter(consumerRecord);
+
+    assertFalse(actual);
+  }
+
+  private void mockConsumerRecord(ResourceEvent event) {
+    when(consumerRecord.value()).thenReturn(event);
+  }
+
+  private ResourceEvent createResourceEvent(ResourceEventSubType subType, String resourceName) {
+    var event = new ResourceEvent();
+    event.setId("1");
+    event.setSubType(subType);
+    event.setResourceName(resourceName);
+    return event;
+  }
+}