MODLD-901: kafka listener saving ImportResultEvents

Author: PBobylev

pom.xml

@@ -405,7 +405,7 @@
               <configOptions>
                 <java8>true</java8>
                 <useTags>true</useTags>
-                <dateLibrary>java</dateLibrary>
+                <dateLibrary>java8</dateLibrary>
                 <interfaceOnly>true</interfaceOnly>
                 <useSpringBoot3>true</useSpringBoot3>
                 <requestMappingMode>api_interface</requestMappingMode>

src/main/java/org/folio/linked/data/imprt/batch/job/processor/Rdf2LdProcessor.java

@@ -9,8 +9,8 @@
 import java.util.Set;
 import lombok.extern.log4j.Log4j2;
 import org.folio.linked.data.imprt.domain.dto.ResourceWithLineNumber;
-import org.folio.linked.data.imprt.model.FailedRdfLine;
 import org.folio.linked.data.imprt.model.RdfLineWithNumber;
+import org.folio.linked.data.imprt.model.entity.FailedRdfLine;
 import org.folio.linked.data.imprt.repo.FailedRdfLineRepo;
 import org.folio.rdf4ld.service.Rdf4LdService;
 import org.springframework.batch.core.configuration.annotation.StepScope;

src/main/java/org/folio/linked/data/imprt/batch/job/writer/LdKafkaSender.java

@@ -4,7 +4,7 @@
 
 import java.util.Set;
 import lombok.extern.log4j.Log4j2;
-import org.folio.linked.data.imprt.domain.dto.ImportOutput;
+import org.folio.linked.data.imprt.domain.dto.ImportOutputEvent;
 import org.folio.linked.data.imprt.domain.dto.ResourceWithLineNumber;
 import org.folio.spring.tools.kafka.FolioMessageProducer;
 import org.springframework.batch.core.configuration.annotation.StepScope;
@@ -20,11 +20,11 @@
 public class LdKafkaSender implements ItemWriter<Set<ResourceWithLineNumber>> {
 
   private final Long jobInstanceId;
-  private final FolioMessageProducer<ImportOutput> importOutputFolioMessageProducer;
+  private final FolioMessageProducer<ImportOutputEvent> importOutputFolioMessageProducer;
   private final Integer chunkSize;
 
   public LdKafkaSender(@Value("#{jobInstanceId}") Long jobInstanceId,
-                       @Qualifier("importOutputMessageProducer") FolioMessageProducer<ImportOutput> producer,
+                       @Qualifier("importOutputMessageProducer") FolioMessageProducer<ImportOutputEvent> producer,
                        @Value("${mod-linked-data-import.output-chunk-size}") Integer chunkSize) {
     this.jobInstanceId = jobInstanceId;
     this.importOutputFolioMessageProducer = producer;
@@ -36,7 +36,7 @@ public LdKafkaSender(@Value("#{jobInstanceId}") Long jobInstanceId,
   public void write(Chunk<? extends Set<ResourceWithLineNumber>> chunk) {
     var messages = chunked(chunk.getItems().stream().flatMap(Set::stream), chunkSize)
       .map(resourcesWithLineNumbers ->
-        new ImportOutput(resourcesWithLineNumbers).jobInstanceId(jobInstanceId))
+        new ImportOutputEvent(resourcesWithLineNumbers).jobInstanceId(jobInstanceId))
       .toList();
     importOutputFolioMessageProducer.sendMessages(messages);
   }

src/main/java/org/folio/linked/data/imprt/config/ObjectMapperConfig.java

@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
@@ -14,11 +15,14 @@ public class ObjectMapperConfig {
   @Bean
   @Primary
   public ObjectMapper objectMapper() {
-    return new ObjectMapper()
+    var objectMapper = new ObjectMapper()
       .setSerializationInclusion(JsonInclude.Include.NON_EMPTY)
       .configure(SerializationFeature.USE_EQUALITY_FOR_OBJECT_ID, true)
+      .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
       .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
       .configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true);
+    objectMapper.registerModule(new JavaTimeModule());
+    return objectMapper;
   }
 
 }

src/main/java/org/folio/linked/data/imprt/config/kafka/ListenerConfiguration.java

@@ -0,0 +1,61 @@
+package org.folio.linked.data.imprt.config.kafka;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.HashMap;
+import java.util.function.Supplier;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.folio.linked.data.imprt.domain.dto.ImportResultEvent;
+import org.folio.spring.tools.kafka.FolioKafkaProperties;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+@Configuration
+public class ListenerConfiguration {
+
+  @Bean
+  @ConfigurationProperties("folio.kafka")
+  public FolioKafkaProperties folioKafkaProperties() {
+    return new FolioKafkaProperties();
+  }
+
+  @Bean
+  public ConcurrentKafkaListenerContainerFactory<String, ImportResultEvent> importResultEventListenerContainerFactory(
+    ConsumerFactory<String, ImportResultEvent> importResultEventConsumerFactory
+  ) {
+    return concurrentKafkaBatchListenerContainerFactory(importResultEventConsumerFactory, false);
+  }
+
+  @Bean
+  public ConsumerFactory<String, ImportResultEvent> importResultEventConsumerFactory(ObjectMapper mapper,
+                                                                                     KafkaProperties properties) {
+    return errorHandlingConsumerFactory(ImportResultEvent.class, mapper, properties);
+  }
+
+
+  private <V> ConcurrentKafkaListenerContainerFactory<String, V> concurrentKafkaBatchListenerContainerFactory(
+    ConsumerFactory<String, V> consumerFactory, boolean batch) {
+    var factory = new ConcurrentKafkaListenerContainerFactory<String, V>();
+    factory.setBatchListener(batch);
+    factory.setConsumerFactory(consumerFactory);
+    return factory;
+  }
+
+
+  private <V> ConsumerFactory<String, V> errorHandlingConsumerFactory(Class<V> clazz,
+                                                                      ObjectMapper mapper,
+                                                                      KafkaProperties kafkaProperties) {
+    var properties = new HashMap<>(kafkaProperties.buildConsumerProperties(null));
+    Supplier<Deserializer<String>> keyDeserializer = StringDeserializer::new;
+    Supplier<Deserializer<V>> valueDeserializer = () ->
+      new ErrorHandlingDeserializer<>(new JsonDeserializer<>(clazz, mapper));
+    return new DefaultKafkaConsumerFactory<>(properties, keyDeserializer, valueDeserializer);
+  }
+}

src/main/java/org/folio/linked/data/imprt/config/kafka/ProducerConfiguration.java

@@ -6,7 +6,7 @@
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.serialization.StringSerializer;
-import org.folio.linked.data.imprt.domain.dto.ImportOutput;
+import org.folio.linked.data.imprt.domain.dto.ImportOutputEvent;
 import org.folio.spring.tools.kafka.FolioMessageProducer;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.context.annotation.Bean;
@@ -23,24 +23,24 @@ public class ProducerConfiguration {
   private final TopicProperties topicProperties;
 
   @Bean
-  public FolioMessageProducer<ImportOutput> importOutputMessageProducer(
-    KafkaTemplate<String, ImportOutput> importOutputMessageTemplate) {
+  public FolioMessageProducer<ImportOutputEvent> importOutputMessageProducer(
+    KafkaTemplate<String, ImportOutputEvent> importOutputMessageTemplate) {
     var producer = new FolioMessageProducer<>(importOutputMessageTemplate, topicProperties::getLinkedDataImportOutput);
-    producer.setKeyMapper(ImportOutput::getTs);
+    producer.setKeyMapper(ImportOutputEvent::getTs);
     return producer;
   }
 
   @Bean
-  public KafkaTemplate<String, ImportOutput> importOutputMessageTemplate(
-    ProducerFactory<String, ImportOutput> importOutputMessageProducerFactory) {
+  public KafkaTemplate<String, ImportOutputEvent> importOutputMessageTemplate(
+    ProducerFactory<String, ImportOutputEvent> importOutputMessageProducerFactory) {
     return new KafkaTemplate<>(importOutputMessageProducerFactory);
   }
 
   @Bean
-  public ProducerFactory<String, ImportOutput> importOutputMessageProducerFactory(ObjectMapper objectMapper) {
+  public ProducerFactory<String, ImportOutputEvent> importOutputMessageProducerFactory(ObjectMapper objectMapper) {
     var properties = new HashMap<>(kafkaProperties.buildProducerProperties(null));
     Supplier<Serializer<String>> keySerializer = StringSerializer::new;
-    Supplier<Serializer<ImportOutput>> valueSerializer = () -> new JsonSerializer<>(objectMapper);
+    Supplier<Serializer<ImportOutputEvent>> valueSerializer = () -> new JsonSerializer<>(objectMapper);
     return new DefaultKafkaProducerFactory<>(properties, keySerializer, valueSerializer);
   }
 

src/main/java/org/folio/linked/data/imprt/integration/kafka/ImportResultEventListener.java

@@ -0,0 +1,62 @@
+package org.folio.linked.data.imprt.integration.kafka;
+
+import static java.util.Optional.ofNullable;
+import static org.folio.linked.data.imprt.util.KafkaUtils.handleForExistedTenant;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.logging.log4j.Level;
+import org.folio.linked.data.imprt.domain.dto.ImportResultEvent;
+import org.folio.linked.data.imprt.integration.kafka.handler.KafkaMessageHandler;
+import org.folio.linked.data.imprt.service.tenant.LinkedDataImportTenantService;
+import org.folio.linked.data.imprt.service.tenant.TenantScopedExecutionService;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.retry.RetryContext;
+import org.springframework.stereotype.Component;
+
+@Log4j2
+@Component
+@RequiredArgsConstructor
+public class ImportResultEventListener {
+
+  private static final String LISTENER_ID = "mod-linked-data-import-import-result-event-listener";
+  private static final String CONTAINER_FACTORY = "importResultEventListenerContainerFactory";
+  private final LinkedDataImportTenantService linkedDataTenantService;
+  private final TenantScopedExecutionService tenantScopedExecutionService;
+  private final KafkaMessageHandler<ImportResultEvent> importResultEventHandler;
+
+  @KafkaListener(
+    id = LISTENER_ID,
+    containerFactory = CONTAINER_FACTORY,
+    groupId = "#{folioKafkaProperties.listener['import-result-event'].groupId}",
+    concurrency = "#{folioKafkaProperties.listener['import-result-event'].concurrency}",
+    topicPattern = "#{folioKafkaProperties.listener['import-result-event'].topicPattern}")
+  public void handleImportOutputEvent(ConsumerRecord<String, ImportResultEvent> consumerRecord) {
+    var event = consumerRecord.value();
+    handleForExistedTenant(consumerRecord, event.getTs(), linkedDataTenantService, log, this::handleRecord);
+  }
+
+  private void handleRecord(ConsumerRecord<String, ImportResultEvent> consumerRecord) {
+    log.info("Processing import result event with Job ID {} and ts {}",
+      consumerRecord.value().getJobInstanceId(), consumerRecord.value().getTs());
+    var event = consumerRecord.value();
+    tenantScopedExecutionService.executeAsyncWithRetry(
+      consumerRecord.headers(),
+      retryContext -> runRetryableJob(event, retryContext),
+      ex -> logFailedEvent(event, ex, false)
+    );
+  }
+
+  private void runRetryableJob(ImportResultEvent event, RetryContext retryContext) {
+    ofNullable(retryContext.getLastThrowable())
+      .ifPresent(ex -> logFailedEvent(event, ex, true));
+    importResultEventHandler.handle(event);
+  }
+
+  private void logFailedEvent(ImportResultEvent event, Throwable ex, boolean isRetrying) {
+    var logLevel = isRetrying ? Level.INFO : Level.ERROR;
+    log.log(logLevel, "Failed to handle import result event with id {}. Retrying: {}",
+      event.getTs(), isRetrying, ex);
+  }
+}

src/main/java/org/folio/linked/data/imprt/integration/kafka/handler/ImportResultEventHandler.java

@@ -0,0 +1,47 @@
+package org.folio.linked.data.imprt.integration.kafka.handler;
+
+import static org.folio.linked.data.imprt.batch.job.Parameters.FILE_URL;
+
+import java.util.Optional;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.folio.linked.data.imprt.domain.dto.ImportResultEvent;
+import org.folio.linked.data.imprt.model.entity.BatchJobExecutionParams;
+import org.folio.linked.data.imprt.model.mapper.ImportResultEventMapper;
+import org.folio.linked.data.imprt.repo.BatchJobExecutionParamsRepo;
+import org.folio.linked.data.imprt.repo.ImportResultEventRepo;
+import org.folio.linked.data.imprt.service.file.FileService;
+import org.springframework.stereotype.Component;
+
+@Log4j2
+@Component
+@RequiredArgsConstructor
+public class ImportResultEventHandler implements KafkaMessageHandler<ImportResultEvent> {
+  private static final String FILE_URL_NOT_FOUND_MESSAGE =
+    "Job parameter [fileUrl] for jobInstanceId [%s] is not found. RDF line reading failed.";
+
+  private final FileService fileService;
+  private final ImportResultEventRepo importResultEventRepo;
+  private final ImportResultEventMapper importResultEventMapper;
+  private final BatchJobExecutionParamsRepo batchJobExecutionParamsRepo;
+
+  @Override
+  public void handle(ImportResultEvent importResultEvent) {
+    var entity = importResultEventMapper.toEntity(importResultEvent);
+    if (!entity.getFailedRdfLines().isEmpty()) {
+      var fileUrl = getFileUrl(importResultEvent.getJobInstanceId());
+      entity.getFailedRdfLines().forEach(failedLine ->
+        failedLine.setFailedRdfLine(fileUrl.map(url -> fileService.readLineFromFile(url, failedLine.getLineNumber()))
+          .orElse(FILE_URL_NOT_FOUND_MESSAGE.formatted(importResultEvent.getJobInstanceId()))
+        )
+      );
+    }
+    importResultEventRepo.save(entity);
+  }
+
+  private Optional<String> getFileUrl(Long jobInstanceId) {
+    return batchJobExecutionParamsRepo.findByJobInstanceIdAndParameterName(jobInstanceId, FILE_URL)
+      .map(BatchJobExecutionParams::getParameterValue);
+  }
+
+}

src/main/java/org/folio/linked/data/imprt/integration/kafka/handler/KafkaMessageHandler.java

@@ -0,0 +1,6 @@
+package org.folio.linked.data.imprt.integration.kafka.handler;
+
+public interface KafkaMessageHandler<T> {
+
+  void handle(T message);
+}

src/main/java/org/folio/linked/data/imprt/model/entity/BatchJobExecutionParams.java

@@ -0,0 +1,28 @@
+package org.folio.linked.data.imprt.model.entity;
+
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.IdClass;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@Entity
+@IdClass(BatchJobExecutionParams.BatchJobExecutionParamsId.class)
+@EqualsAndHashCode(of = {"jobExecutionId", "parameterName"})
+public class BatchJobExecutionParams {
+
+  @Id
+  private Long jobExecutionId;
+  @Id
+  private String parameterName;
+  private String parameterValue;
+
+  @Data
+  public static class BatchJobExecutionParamsId implements Serializable {
+    private Long jobExecutionId;
+    private String parameterName;
+  }
+}
+