MODLD-901: preserve initial RDF line number (#46)

PBobylev · web-flow · commit 6ef3e19d7220 · 2025-12-08T17:42:18.000+05:00
diff --git a/src/main/java/org/folio/linked/data/imprt/batch/job/mapper/LineNumberCapturingMapper.java b/src/main/java/org/folio/linked/data/imprt/batch/job/mapper/LineNumberCapturingMapper.java
@@ -0,0 +1,14 @@
+package org.folio.linked.data.imprt.batch.job.mapper;
+
+import org.folio.linked.data.imprt.model.RdfLineWithNumber;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.batch.item.file.LineMapper;
+
+public class LineNumberCapturingMapper implements LineMapper<RdfLineWithNumber> {
+
+  @Override
+  public @NotNull RdfLineWithNumber mapLine(@NotNull String line, int lineNumber) {
+    return new RdfLineWithNumber(lineNumber, line);
+  }
+}
+
diff --git a/src/main/java/org/folio/linked/data/imprt/batch/job/processor/Rdf2LdProcessor.java b/src/main/java/org/folio/linked/data/imprt/batch/job/processor/Rdf2LdProcessor.java
@@ -1,25 +1,29 @@
 package org.folio.linked.data.imprt.batch.job.processor;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.stream.Collectors.toCollection;
 import static org.folio.linked.data.imprt.batch.job.Parameters.CONTENT_TYPE;
 
 import java.io.ByteArrayInputStream;
+import java.util.LinkedHashSet;
 import java.util.Set;
 import lombok.extern.log4j.Log4j2;
-import org.folio.ld.dictionary.model.Resource;
+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.repo.FailedRdfLineRepo;
 import org.folio.rdf4ld.service.Rdf4LdService;
-import org.jetbrains.annotations.NotNull;
 import org.springframework.batch.core.configuration.annotation.StepScope;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
 import org.springframework.stereotype.Component;
 
 @Log4j2
 @Component
 @StepScope
-public class Rdf2LdProcessor implements ItemProcessor<String, Set<Resource>> {
+public class Rdf2LdProcessor implements ItemProcessor<RdfLineWithNumber, Set<ResourceWithLineNumber>> {
 
   private static final String EMPTY_RESULT = "Empty result returned by rdf4ld library";
   private final Long jobInstanceId;
@@ -37,29 +41,36 @@ public Rdf2LdProcessor(@Value("#{jobInstanceId}") Long jobInstanceId,
     this.failedRdfLineRepo = failedRdfLineRepo;
   }
 
-
   @Override
-  public Set<Resource> process(@NotNull String rdfLine) {
-    log.trace("Processing RDF line of contentType[{}]: {}", contentType, rdfLine);
+  @Nullable
+  @SuppressWarnings("java:S2638")
+  public Set<ResourceWithLineNumber> process(@NonNull RdfLineWithNumber rdfLineWithNumber) {
+    var rdfLine = rdfLineWithNumber.getContent();
+    var lineNumber = rdfLineWithNumber.getLineNumber();
+    log.trace("Processing RDF line #{} of contentType[{}]: {}", lineNumber, contentType, rdfLine);
     try {
       var is = new ByteArrayInputStream(rdfLine.getBytes(UTF_8));
       var result = rdf4LdService.mapBibframe2RdfToLd(is, contentType);
       if (result.isEmpty()) {
-        log.debug(EMPTY_RESULT + ", saving FailedRdfLine. JobInstanceId [{}]", jobInstanceId);
-        saveFailedLine(rdfLine, EMPTY_RESULT);
+        log.debug(EMPTY_RESULT + ", saving FailedRdfLine. JobInstanceId [{}], line #{}", jobInstanceId, lineNumber);
+        saveFailedLine(lineNumber, rdfLine, EMPTY_RESULT);
         return null;
       }
-      return result;
+      return result.stream()
+        .map(resource -> new ResourceWithLineNumber(lineNumber, resource))
+        .collect(toCollection(LinkedHashSet::new));
     } catch (Exception e) {
-      log.warn("Exception during processing RDF line, saving FailedRdfLine. JobInstanceId [{}]", jobInstanceId);
-      saveFailedLine(rdfLine, e.getMessage());
+      log.warn("Exception during processing RDF line #{}, saving FailedRdfLine. JobInstanceId [{}]", lineNumber,
+        jobInstanceId);
+      saveFailedLine(lineNumber, rdfLine, e.getMessage());
       return null;
     }
   }
 
-  private void saveFailedLine(String rdfLine, String message) {
+  private void saveFailedLine(long lineNumber, String rdfLine, String message) {
     var frl = new FailedRdfLine()
       .setJobInstanceId(jobInstanceId)
+      .setLineNumber(lineNumber)
       .setDescription(message)
       .setFailedRdfLine(rdfLine);
     failedRdfLineRepo.save(frl);
diff --git a/src/main/java/org/folio/linked/data/imprt/batch/job/writer/LdKafkaSender.java b/src/main/java/org/folio/linked/data/imprt/batch/job/writer/LdKafkaSender.java
@@ -4,8 +4,8 @@
 
 import java.util.Set;
 import lombok.extern.log4j.Log4j2;
-import org.folio.ld.dictionary.model.Resource;
 import org.folio.linked.data.imprt.domain.dto.ImportOutput;
+import org.folio.linked.data.imprt.domain.dto.ResourceWithLineNumber;
 import org.folio.spring.tools.kafka.FolioMessageProducer;
 import org.springframework.batch.core.configuration.annotation.StepScope;
 import org.springframework.batch.item.Chunk;
@@ -17,7 +17,7 @@
 @Log4j2
 @Component
 @StepScope
-public class LdKafkaSender implements ItemWriter<Set<Resource>> {
+public class LdKafkaSender implements ItemWriter<Set<ResourceWithLineNumber>> {
 
   private final Long jobInstanceId;
   private final FolioMessageProducer<ImportOutput> importOutputFolioMessageProducer;
@@ -33,9 +33,10 @@ public LdKafkaSender(@Value("#{jobInstanceId}") Long jobInstanceId,
 
 
   @Override
-  public void write(Chunk<? extends Set<Resource>> chunk) {
+  public void write(Chunk<? extends Set<ResourceWithLineNumber>> chunk) {
     var messages = chunked(chunk.getItems().stream().flatMap(Set::stream), chunkSize)
-      .map(resources -> new ImportOutput(resources).jobInstanceId(jobInstanceId))
+      .map(resourcesWithLineNumbers ->
+        new ImportOutput(resourcesWithLineNumbers).jobInstanceId(jobInstanceId))
       .toList();
     importOutputFolioMessageProducer.sendMessages(messages);
   }
diff --git a/src/main/java/org/folio/linked/data/imprt/config/BatchConfig.java b/src/main/java/org/folio/linked/data/imprt/config/BatchConfig.java
@@ -8,9 +8,11 @@
 import java.io.File;
 import java.util.Set;
 import javax.sql.DataSource;
-import org.folio.ld.dictionary.model.Resource;
+import org.folio.linked.data.imprt.batch.job.mapper.LineNumberCapturingMapper;
 import org.folio.linked.data.imprt.batch.job.processor.Rdf2LdProcessor;
 import org.folio.linked.data.imprt.batch.job.writer.LdKafkaSender;
+import org.folio.linked.data.imprt.domain.dto.ResourceWithLineNumber;
+import org.folio.linked.data.imprt.model.RdfLineWithNumber;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.configuration.annotation.StepScope;
@@ -21,9 +23,8 @@
 import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
 import org.springframework.batch.core.step.builder.StepBuilder;
 import org.springframework.batch.core.step.tasklet.Tasklet;
-import org.springframework.batch.item.file.FlatFileItemReader;
 import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
-import org.springframework.batch.item.file.mapping.PassThroughLineMapper;
+import org.springframework.batch.item.support.SynchronizedItemStreamReader;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
@@ -121,13 +122,13 @@ public TaskExecutor processFileTaskExecutor(
   @Bean
   public Step processFileStep(JobRepository jobRepository,
                               PlatformTransactionManager transactionManager,
-                              FlatFileItemReader<String> rdfLineItemReader,
+                              SynchronizedItemStreamReader<RdfLineWithNumber> rdfLineItemReader,
                               Rdf2LdProcessor rdf2LdProcessor,
                               LdKafkaSender ldKafkaSender,
                               @Value("${mod-linked-data-import.chunk-size}") int chunkSize,
                               TaskExecutor processFileTaskExecutor) {
     return new StepBuilder("processFileStep", jobRepository)
-      .<String, Set<Resource>>chunk(chunkSize, transactionManager)
+      .<RdfLineWithNumber, Set<ResourceWithLineNumber>>chunk(chunkSize, transactionManager)
       .reader(rdfLineItemReader)
       .processor(rdf2LdProcessor)
       .writer(ldKafkaSender)
@@ -137,13 +138,18 @@ public Step processFileStep(JobRepository jobRepository,
 
   @Bean
   @StepScope
-  public FlatFileItemReader<String> rdfLineItemReader(@Value("#{jobParameters['" + FILE_URL + "']}") String fileUrl) {
+  public SynchronizedItemStreamReader<RdfLineWithNumber> rdfLineItemReader(
+    @Value("#{jobParameters['" + FILE_URL + "']}") String fileUrl
+  ) {
     var file = new File(TMP_DIR, extractFileName(fileUrl));
-    return new FlatFileItemReaderBuilder<String>()
+    var flatFileReader = new FlatFileItemReaderBuilder<RdfLineWithNumber>()
       .name("lineItemReader")
       .resource(new org.springframework.core.io.FileSystemResource(file))
-      .lineMapper(new PassThroughLineMapper())
+      .lineMapper(new LineNumberCapturingMapper())
       .build();
+    var synchronizedReader = new SynchronizedItemStreamReader<RdfLineWithNumber>();
+    synchronizedReader.setDelegate(flatFileReader);
+    return synchronizedReader;
   }
 
   @Bean
diff --git a/src/main/java/org/folio/linked/data/imprt/model/FailedRdfLine.java b/src/main/java/org/folio/linked/data/imprt/model/FailedRdfLine.java
@@ -24,6 +24,7 @@ public class FailedRdfLine {
   @GeneratedValue(strategy = SEQUENCE, generator = FAILED_RDF_LINE_SEQ_GEN)
   private Long id;
   private Long jobInstanceId;
+  private Long lineNumber;
   private String failedRdfLine;
   private String description;
 
diff --git a/src/main/java/org/folio/linked/data/imprt/model/RdfLineWithNumber.java b/src/main/java/org/folio/linked/data/imprt/model/RdfLineWithNumber.java
@@ -0,0 +1,13 @@
+package org.folio.linked.data.imprt.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class RdfLineWithNumber {
+
+  private long lineNumber;
+  private String content;
+
+}
diff --git a/src/main/java/org/folio/linked/data/imprt/util/CollectionUtil.java b/src/main/java/org/folio/linked/data/imprt/util/CollectionUtil.java
@@ -1,6 +1,6 @@
 package org.folio.linked.data.imprt.util;
 
-import static java.util.HashSet.newHashSet;
+import static java.util.LinkedHashSet.newLinkedHashSet;
 
 import java.util.Iterator;
 import java.util.Set;
@@ -21,7 +21,7 @@ public boolean hasNext() {
 
       @Override
       public Set<T> next() {
-        Set<T> chunk = newHashSet(chunkSize);
+        Set<T> chunk = newLinkedHashSet(chunkSize);
         int count = 0;
         while (count < chunkSize && sourceIterator.hasNext()) {
           chunk.add(sourceIterator.next());
diff --git a/src/main/resources/changelog/scripts/v-1.0.0/metadata/tables/2_failed_rdf_line.xml b/src/main/resources/changelog/scripts/v-1.0.0/metadata/tables/2_failed_rdf_line.xml
@@ -31,6 +31,7 @@
         <constraints nullable="false" primaryKey="true" primaryKeyName="pk_failed_rdf_line"/>
       </column>
       <column name="job_instance_id" type="BIGINT"/>
+      <column name="line_number" type="BIGINT"/>
       <column name="failed_rdf_line" type="TEXT"/>
       <column name="description" type="TEXT"/>
     </createTable>
diff --git a/src/main/resources/swagger.api/kafka/importOutput.json b/src/main/resources/swagger.api/kafka/importOutput.json
@@ -3,13 +3,13 @@
   "$schema": "http://json-schema.org/draft-04/schema#",
   "description": "Linked Data Import result DTO",
   "properties": {
-    "resources": {
-      "description": "Array of imported resources",
+    "resourcesWithLineNumbers": {
+      "description": "Array of imported resources with their line numbers",
       "type": "array",
       "uniqueItems": true,
       "items": {
         "type": "object",
-        "$ref": "resourcePlaceholder.json"
+        "$ref": "resourceWithLineNumber.json"
       }
     },
     "jobInstanceId": {
@@ -27,5 +27,5 @@
     }
   },
   "x-implements": "org.folio.spring.tools.kafka.BaseKafkaMessage",
-  "required": ["resources"]
+  "required": ["resourcesWithLineNumbers"]
 }
diff --git a/src/main/resources/swagger.api/kafka/resourceWithLineNumber.json b/src/main/resources/swagger.api/kafka/resourceWithLineNumber.json
@@ -0,0 +1,19 @@
+{
+  "type": "object",
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "description": "Resource with its line number from the RDF file",
+  "properties": {
+    "lineNumber": {
+      "description": "Line number from the RDF file",
+      "type": "integer",
+      "format": "int64"
+    },
+    "resource": {
+      "description": "Mapped resource",
+      "type": "object",
+      "$ref": "resourcePlaceholder.json"
+    }
+  },
+  "required": ["lineNumber", "resource"]
+}
+
diff --git a/src/test/java/org/folio/linked/data/imprt/batch/job/processor/Rdf2LdProcessorTest.java b/src/test/java/org/folio/linked/data/imprt/batch/job/processor/Rdf2LdProcessorTest.java
@@ -10,6 +10,7 @@
 import java.util.Set;
 import org.folio.ld.dictionary.model.Resource;
 import org.folio.linked.data.imprt.model.FailedRdfLine;
+import org.folio.linked.data.imprt.model.RdfLineWithNumber;
 import org.folio.linked.data.imprt.repo.FailedRdfLineRepo;
 import org.folio.rdf4ld.service.Rdf4LdService;
 import org.folio.spring.testing.type.UnitTest;
@@ -39,26 +40,32 @@ void setUp() {
   @Test
   void process_shouldReturnSuccessfulRdf4LdServiceResult() {
     // given
-    var rdfLine = "rdfLine";
-    var expectedResult = Set.of(new Resource().setId(JOB_INSTANCE_ID));
-    doReturn(expectedResult).when(rdf4LdService).mapBibframe2RdfToLd(any(), eq(CONTENT_TYPE));
+    var lineNumber = 1L;
+    var rdfLine = new RdfLineWithNumber(lineNumber, "rdfLine");
+    var resource = new Resource().setId(JOB_INSTANCE_ID);
+    var rdf4LdResult = Set.of(resource);
+    doReturn(rdf4LdResult).when(rdf4LdService).mapBibframe2RdfToLd(any(), eq(CONTENT_TYPE));
 
     // when
     var result = rdf2LdProcessor.process(rdfLine);
 
     // then
-    assertThat(result).isEqualTo(expectedResult);
+    assertThat(result).hasSize(1);
+    var resourceWithLineNumber = result.iterator().next();
+    assertThat(resourceWithLineNumber.getLineNumber()).isEqualTo(lineNumber);
+    assertThat(resourceWithLineNumber.getResource()).isEqualTo(resource);
   }
 
   @Test
   void process_shouldSaveFailedRdfLineAndReturnNull_ifAnyExceptionInRdf4LdService() {
     // given
-    var rdfLine = "rdfLine";
+    var rdfLine = new RdfLineWithNumber(2L, "rdfLine");
     var rdf4LdException = "rdf4LdService exception";
     doThrow(new RuntimeException(rdf4LdException)).when(rdf4LdService).mapBibframe2RdfToLd(any(), eq(CONTENT_TYPE));
     var expectedFailedRdfLine = new FailedRdfLine()
       .setJobInstanceId(JOB_INSTANCE_ID)
-      .setFailedRdfLine(rdfLine)
+      .setLineNumber(rdfLine.getLineNumber())
+      .setFailedRdfLine(rdfLine.getContent())
       .setDescription(rdf4LdException);
 
     // when
@@ -72,12 +79,13 @@ void process_shouldSaveFailedRdfLineAndReturnNull_ifAnyExceptionInRdf4LdService(
   @Test
   void process_shouldSaveFailedRdfLineAndReturnNull_ifEmptyResultFromRdf4LdService() {
     // given
-    var rdfLine = "rdfLine";
+    var rdfLine = new RdfLineWithNumber(3L, "rdfLine");
     var expectedResult = Set.of();
     doReturn(expectedResult).when(rdf4LdService).mapBibframe2RdfToLd(any(), eq(CONTENT_TYPE));
     var expectedFailedRdfLine = new FailedRdfLine()
       .setJobInstanceId(JOB_INSTANCE_ID)
-      .setFailedRdfLine(rdfLine)
+      .setLineNumber(rdfLine.getLineNumber())
+      .setFailedRdfLine(rdfLine.getContent())
       .setDescription("Empty result returned by rdf4ld library");
 
     // when
diff --git a/src/test/java/org/folio/linked/data/imprt/batch/job/writer/LdKafkaSenderTest.java b/src/test/java/org/folio/linked/data/imprt/batch/job/writer/LdKafkaSenderTest.java
diff --git a/src/test/java/org/folio/linked/data/imprt/e2e/ImportIT.java b/src/test/java/org/folio/linked/data/imprt/e2e/ImportIT.java
diff --git a/src/test/java/org/folio/linked/data/imprt/test/KafkaOutputTopicTestListener.java b/src/test/java/org/folio/linked/data/imprt/test/KafkaOutputTopicTestListener.java
diff --git a/src/test/java/org/folio/linked/data/imprt/util/CollectionUtilTest.java b/src/test/java/org/folio/linked/data/imprt/util/CollectionUtilTest.java
