MODLD-395: working implementation

folio-org · Jun 17, 2024 · 5c8cc83 · 5c8cc83
1 parent cfbc449
commit 5c8cc83
Show file tree

Hide file tree

Showing 22 changed files with 101 additions and 60 deletions.
diff --git a/.run/LinkedDataApplication [Local-Folio].run.xml b/.run/LinkedDataApplication [Local-Folio].run.xml
@@ -2,7 +2,7 @@
   <configuration default="false" name="LinkedDataApplication [Local-Folio]" type="Application" factoryName="Application">
     <option name="MAIN_CLASS_NAME" value="org.folio.linked.data.LinkedDataApplication" />
     <module name="mod-linked-data" />
-    <option name="VM_PARAMETERS" value="-Dspring.profiles.active=folio,search,local" />
+    <option name="VM_PARAMETERS" value="-Dspring.profiles.active=folio,local" />
     <extension name="coverage">
       <pattern>
         <option name="PATTERN" value="org.folio.linked.data.*" />

diff --git a/src/main/java/org/folio/linked/data/client/SearchClient.java b/src/main/java/org/folio/linked/data/client/SearchClient.java
@@ -1,5 +1,6 @@
 package org.folio.linked.data.client;
 
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
 
@@ -12,7 +13,7 @@
 import org.springframework.web.bind.annotation.RequestBody;
 
 @FeignClient(name = "search")
-@Profile(SEARCH_PROFILE)
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 public interface SearchClient {
 
   @PostMapping(value = "/index/indices", consumes = APPLICATION_JSON_VALUE, produces = APPLICATION_JSON_VALUE)

diff --git a/src/main/java/org/folio/linked/data/configuration/FolioKafkaConfig.java b/src/main/java/org/folio/linked/data/configuration/FolioKafkaConfig.java
@@ -22,17 +22,18 @@
 
 @Configuration
 @RequiredArgsConstructor
-@Profile(FOLIO_PROFILE)
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 public class FolioKafkaConfig {
   private final KafkaProperties kafkaProperties;
 
   @Bean
-  @Profile(SEARCH_PROFILE)
+  @Profile({FOLIO_PROFILE, SEARCH_PROFILE})
   public ProducerFactory<String, ResourceEvent> searchIndexProducerFactory() {
     return new DefaultKafkaProducerFactory<>(getProducerFactoryProperties());
   }
 
   @Bean
+  @Profile(FOLIO_PROFILE)
   public ProducerFactory<String, InstanceIngressEvent> instanceIngressProducerFactory() {
     return new DefaultKafkaProducerFactory<>(getProducerFactoryProperties());
   }
@@ -45,12 +46,13 @@ private Map<String, Object> getProducerFactoryProperties() {
   }
 
   @Bean
-  @Profile(SEARCH_PROFILE)
+  @Profile({FOLIO_PROFILE, SEARCH_PROFILE})
   public KafkaTemplate<String, ResourceEvent> searchTemplate(ProducerFactory<String, ResourceEvent> factory) {
     return new KafkaTemplate<>(factory);
   }
 
   @Bean
+  @Profile(FOLIO_PROFILE)
   public KafkaTemplate<String, InstanceIngressEvent> iiTemplate(ProducerFactory<String, InstanceIngressEvent> factory) {
     return new KafkaTemplate<>(factory);
   }

diff --git a/src/main/java/org/folio/linked/data/configuration/standalone/StandaloneExcludeProcessor.java b/src/main/java/org/folio/linked/data/configuration/standalone/StandaloneExcludeProcessor.java
@@ -22,6 +22,7 @@ public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) t
 
   private Collection<String> getExcludeBeans() {
     return List.of(
+      "TenantService",
       "defaultTenantController",
       "defaultTenantOkapiHeaderValidationFilter"
     );

diff --git a/src/main/java/org/folio/linked/data/controller/ReindexingController.java b/src/main/java/org/folio/linked/data/controller/ReindexingController.java
@@ -1,5 +1,6 @@
 package org.folio.linked.data.controller;
 
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 
 import lombok.RequiredArgsConstructor;
@@ -11,7 +12,7 @@
 import org.springframework.web.bind.annotation.RestController;
 
 @Validated
-@Profile(SEARCH_PROFILE)
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 @RestController
 @RequiredArgsConstructor
 public class ReindexingController implements ReindexApi {

diff --git a/...va/org/folio/linked/data/integration/kafka/sender/inventory/KafkaInventorySenderImpl.java b/...va/org/folio/linked/data/integration/kafka/sender/inventory/KafkaInventorySenderImpl.java
@@ -1,18 +1,22 @@
 package org.folio.linked.data.integration.kafka.sender.inventory;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
+import static org.folio.spring.tools.config.properties.FolioEnvironment.getFolioEnvName;
 import static org.folio.spring.tools.kafka.KafkaUtils.getTenantTopicName;
 
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.log4j.Log4j2;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.folio.linked.data.mapper.kafka.KafkaInventoryMessageMapper;
 import org.folio.linked.data.model.entity.Resource;
 import org.folio.search.domain.dto.InstanceIngressEvent;
 import org.folio.search.domain.dto.InventoryInstanceIngressEventEventMetadata;
 import org.folio.spring.FolioExecutionContext;
+import org.folio.spring.integration.XOkapiHeaders;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Profile;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -35,19 +39,21 @@ public class KafkaInventorySenderImpl implements KafkaInventorySender {
   @SneakyThrows
   public void sendInstanceCreated(Resource resource) {
     var tenant = folioExecutionContext.getTenantId();
-    var tenantTopicName = getTenantTopicName(initialInventoryInstanceIngressTopicName, tenant);
+    var tenantTopicName = getTenantTopicName(initialInventoryInstanceIngressTopicName, getFolioEnvName(), tenant);
     kafkaInventoryMessageMapper.toInstanceIngressPayload(resource)
       .map(p -> {
-          String id = UUID.randomUUID().toString();
-          return inventoryKafkaTemplate.send(tenantTopicName, id,
-            new InstanceIngressEvent()
-              .id(id)
-              .eventType(InstanceIngressEvent.EventTypeEnum.CREATE_INSTANCE)
-              .eventPayload(p)
-              .eventMetadata(new InventoryInstanceIngressEventEventMetadata()
-                .tenantId(tenant)
-              )
-          );
+          var id = UUID.randomUUID().toString();
+          var event = new InstanceIngressEvent()
+            .id(id)
+            .eventType(InstanceIngressEvent.EventTypeEnum.CREATE_INSTANCE)
+            .eventPayload(p)
+            .eventMetadata(new InventoryInstanceIngressEventEventMetadata()
+              .tenantId(tenant)
+            );
+          var pr = new ProducerRecord<>(tenantTopicName, id, event);
+          pr.headers().add(XOkapiHeaders.URL, folioExecutionContext.getOkapiUrl().getBytes(UTF_8));
+          pr.headers().add(XOkapiHeaders.TOKEN, folioExecutionContext.getToken().getBytes(UTF_8));
+          return inventoryKafkaTemplate.send(pr);
         }
       ).ifPresent(f -> logSending(tenantTopicName, f));
   }

diff --git a/...in/java/org/folio/linked/data/integration/kafka/sender/search/KafkaSearchSenderDummy.java b/...in/java/org/folio/linked/data/integration/kafka/sender/search/KafkaSearchSenderDummy.java
@@ -1,5 +1,6 @@
 package org.folio.linked.data.integration.kafka.sender.search;
 
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 
 import lombok.RequiredArgsConstructor;
@@ -10,7 +11,7 @@
 
 @Log4j2
 @Service
-@Profile("!" + SEARCH_PROFILE)
+@Profile({"!" + FOLIO_PROFILE + "& !" + SEARCH_PROFILE})
 @RequiredArgsConstructor
 public class KafkaSearchSenderDummy implements KafkaSearchSender {
 

diff --git a/...ain/java/org/folio/linked/data/integration/kafka/sender/search/KafkaSearchSenderImpl.java b/...ain/java/org/folio/linked/data/integration/kafka/sender/search/KafkaSearchSenderImpl.java
@@ -2,6 +2,7 @@
 
 import static java.lang.Long.parseLong;
 import static org.folio.linked.data.util.BibframeUtils.isSameResource;
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_RESOURCE_NAME;
 import static org.folio.search.domain.dto.ResourceEventType.CREATE;
@@ -26,15 +27,15 @@
 
 @Log4j2
 @Service
-@Profile(SEARCH_PROFILE)
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 @RequiredArgsConstructor
 public class KafkaSearchSenderImpl implements KafkaSearchSender {
 
   private final KafkaTemplate<String, ResourceEvent> kafkaTemplate;
   private final FolioExecutionContext folioExecutionContext;
   private final ApplicationEventPublisher eventPublisher;
   private final KafkaSearchMessageMapper kafkaSearchMessageMapper;
-  @Value("${mod-linked-data.kafka.topic.bibframe-index}")
+  @Value("${mod-linked-data.kafka.topic.search.bibframe-index}")
   private String initialBibframeIndexTopicName;
 
   @SneakyThrows

diff --git a/src/main/java/org/folio/linked/data/mapper/kafka/impl/KafkaInventoryMessageMapperImpl.java b/src/main/java/org/folio/linked/data/mapper/kafka/impl/KafkaInventoryMessageMapperImpl.java
@@ -3,6 +3,7 @@
 import static java.util.Objects.isNull;
 
 import java.util.Optional;
+import java.util.UUID;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.folio.linked.data.mapper.ResourceModelMapper;
@@ -17,6 +18,7 @@
 @RequiredArgsConstructor
 public class KafkaInventoryMessageMapperImpl implements KafkaInventoryMessageMapper {
 
+  private static final String LINKED_DATA_ID = "linkedDataId";
   private final Bibframe2MarcMapper bibframe2MarcMapper;
   private final ResourceModelMapper resourceModelMapper;
 
@@ -31,9 +33,10 @@ public Optional<InstanceIngressPayload> toInstanceIngressPayload(Resource instan
       return Optional.empty();
     }
     var payload = new InstanceIngressPayload()
-      .sourceRecordIdentifier(String.valueOf(instance.getId()))
-      .sourceType(InstanceIngressPayload.SourceTypeEnum.BIBFRAME)
-      .sourceRecordObject(marcJson);
+      .sourceRecordIdentifier(UUID.randomUUID().toString())
+      .sourceType(InstanceIngressPayload.SourceTypeEnum.LINKED_DATA)
+      .sourceRecordObject(marcJson)
+      .putAdditionalProperty(LINKED_DATA_ID, instance.getId());
     return Optional.of(payload);
   }
 

diff --git a/src/main/java/org/folio/linked/data/service/impl/BatchIndexServiceImpl.java b/src/main/java/org/folio/linked/data/service/impl/BatchIndexServiceImpl.java
@@ -1,5 +1,6 @@
 package org.folio.linked.data.service.impl;
 
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 
 import jakarta.persistence.EntityManager;
@@ -19,7 +20,7 @@
 @Log4j2
 @Service
 @Transactional(readOnly = true)
-@Profile(SEARCH_PROFILE)
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 @RequiredArgsConstructor
 public class BatchIndexServiceImpl implements BatchIndexService {
 

diff --git a/src/main/java/org/folio/linked/data/service/impl/ReindexServiceImpl.java b/src/main/java/org/folio/linked/data/service/impl/ReindexServiceImpl.java
@@ -2,6 +2,7 @@
 
 import static java.lang.Boolean.TRUE;
 import static org.folio.ld.dictionary.ResourceTypeDictionary.WORK;
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 
 import java.util.Set;
@@ -24,7 +25,7 @@
 @Log4j2
 @Service
 @Transactional
-@Profile(SEARCH_PROFILE)
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 @RequiredArgsConstructor
 public class ReindexServiceImpl implements ReindexService {
 

diff --git a/src/main/java/org/folio/linked/data/service/impl/tenant/LinkedTenantService.java b/src/main/java/org/folio/linked/data/service/impl/tenant/LinkedTenantService.java
@@ -1,5 +1,8 @@
 package org.folio.linked.data.service.impl.tenant;
 
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
+import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
+
 import java.util.Collection;
 import lombok.extern.log4j.Log4j2;
 import org.folio.spring.FolioExecutionContext;
@@ -8,12 +11,14 @@
 import org.folio.tenant.domain.dto.TenantAttributes;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Primary;
+import org.springframework.context.annotation.Profile;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Service;
 
 @Log4j2
 @Primary
 @Service
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 public class LinkedTenantService extends TenantService {
 
   private final Collection<TenantServiceWorker> workers;

diff --git a/src/main/java/org/folio/linked/data/service/impl/tenant/workers/KafkaAdminWorker.java b/src/main/java/org/folio/linked/data/service/impl/tenant/workers/KafkaAdminWorker.java
@@ -1,11 +1,16 @@
 package org.folio.linked.data.service.impl.tenant.workers;
 
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
+import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
+
 import lombok.RequiredArgsConstructor;
 import org.folio.linked.data.service.impl.tenant.TenantServiceWorker;
 import org.folio.spring.tools.kafka.KafkaAdminService;
 import org.folio.tenant.domain.dto.TenantAttributes;
+import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Service;
 
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 @Service
 @RequiredArgsConstructor
 public class KafkaAdminWorker implements TenantServiceWorker {

diff --git a/src/main/java/org/folio/linked/data/service/impl/tenant/workers/SearchWorker.java b/src/main/java/org/folio/linked/data/service/impl/tenant/workers/SearchWorker.java
@@ -1,5 +1,6 @@
 package org.folio.linked.data.service.impl.tenant.workers;
 
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 import static org.folio.linked.data.util.Constants.SEARCH_RESOURCE_NAME;
 
@@ -14,10 +15,10 @@
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 
-@Profile(SEARCH_PROFILE)
 @Log4j2
 @Service
 @RequiredArgsConstructor
+@Profile({FOLIO_PROFILE, SEARCH_PROFILE})
 public class SearchWorker implements TenantServiceWorker {
 
   private final SearchClient searchClient;

diff --git a/src/main/resources/application-folio.yml b/src/main/resources/application-folio.yml
@@ -1,3 +1,13 @@
+mod-linked-data:
+  reindex:
+    page-size: 1000
+  kafka:
+    topic:
+      search:
+        bibframe-index: search.bibframe
+      inventory:
+        instance-ingress: inventory.instance_ingress
+
 folio:
   environment: ${ENV:folio}
   okapiUrl: ${okapi.url}
@@ -17,6 +27,15 @@ folio:
         concurrency: ${KAFKA_DATA_IMPORT_INSTANCE_CREATE_CONCURRENCY:1}
         topic-pattern: ${KAFKA_DI_INSTANCE_CREATED_TOPIC_PATTERN:(${folio.environment}\.)(.*\.)DI_COMPLETED}
         group-id: ${folio.environment}-linked-data-data-import-instances-created-group
+    retry-interval-ms: ${KAFKA_RETRY_INTERVAL_MS:2000}
+    retry-delivery-attempts: ${KAFKA_RETRY_DELIVERY_ATTEMPTS:6}
+    topics:
+      - name: ${mod-linked-data.kafka.topic.search.bibframe-index:search.bibframe}
+        numPartitions: ${KAFKA_BIBFRAME_TOPIC_PARTITIONS:3}
+        replicationFactor: ${KAFKA_BIBFRAME_TOPIC_REPLICATION_FACTOR:}
+      - name: ${mod-linked-data.kafka.topic.inventory.instance-ingress:inventory.instance_ingress}
+        numPartitions: ${KAFKA_BIBFRAME_TOPIC_PARTITIONS:3}
+        replicationFactor: ${KAFKA_BIBFRAME_TOPIC_REPLICATION_FACTOR:}
 
 spring:
   kafka:
@@ -37,9 +56,3 @@ spring:
         max.in.flight.requests.per.connection: 5
         retries: 5
         spring.json.add.type.headers: false
-
-mod-linked-data:
-  kafka:
-    topic:
-      inventory:
-        instance-ingress: inventory.instance_ingress
diff --git a/src/main/resources/application-local.yml b/src/main/resources/application-local.yml
@@ -1,14 +1,16 @@
 spring:
   datasource:
     url: jdbc:postgresql://localhost:5432/okapi_modules
+    username: folio_admin
+    password: folio_admin
   kafka:
-    bootstrap-servers: localhost:9092
+    bootstrap-servers: localhost:29092
   cloud:
     openfeign:
       client:
         config:
           search:
-            url: http://localhost:8085/search
+            url: http://localhost:9130/search
 
 management:
   endpoints: