MODDICORE-484. Update KafkaEventPublisher to send x-okapi-request header (#419)

SerhiiNosko · web-flow · commit 171212f9e767 · 2025-12-12T14:57:21.000+02:00
* MODDICORE-484. Update KafkaEventPublisher to send x-okapi-request header

* MODDICORE-421. Include Job Execution Identifiers In Logs

* MODDICORE-421. Increase unit tests coverage
diff --git a/src/main/java/org/folio/processing/events/EventManager.java b/src/main/java/org/folio/processing/events/EventManager.java
@@ -31,6 +31,7 @@
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.folio.DataImportEventTypes.DI_COMPLETED;
 import static org.folio.DataImportEventTypes.DI_ERROR;
+import static org.folio.processing.events.utils.EventUtils.extractRecordId;
 import static org.folio.rest.jaxrs.model.ProfileType.ACTION_PROFILE;
 import static org.folio.rest.jaxrs.model.ProfileType.JOB_PROFILE;
 import static org.folio.rest.jaxrs.model.ProfileType.MAPPING_PROFILE;
@@ -67,7 +68,9 @@ static List<EventPublisher> getEventPublishers() {
    * @return future with event payload after handling
    */
   public static CompletableFuture<DataImportEventPayload> handleEvent(DataImportEventPayload eventPayload, ProfileSnapshotWrapper jobProfileSnapshot) {
-    LOGGER.trace("handleEvent:: Event type: {}, event payload: {}", eventPayload.getEventType(), eventPayload);
+    LOGGER.trace("handleEvent:: Event type: {} jobExecutionId: {} recordId: {}, event payload: {}",
+      eventPayload.getEventType(), eventPayload.getJobExecutionId(),
+      extractRecordId(eventPayload), eventPayload);
     CompletableFuture<DataImportEventPayload> future = new CompletableFuture<>();
     try {
       setCurrentNodeIfRoot(eventPayload, jobProfileSnapshot);
@@ -78,12 +81,14 @@ public static CompletableFuture<DataImportEventPayload> handleEvent(DataImportEv
               if (publishThrowable == null) {
                 future.complete(eventPayload);
               } else {
-                LOGGER.warn("handleEvent:: Can`t publish event", publishThrowable);
+                LOGGER.warn("handleEvent:: Can`t publish event jobExecutionId: {} recordId: {}",
+                  eventPayload.getJobExecutionId(), extractRecordId(eventPayload), publishThrowable);
                 future.completeExceptionally(publishThrowable);
               }
             }));
     } catch (Exception e) {
-      LOGGER.warn("handleEvent:: Can`t handle event", e);
+      LOGGER.warn("handleEvent:: Can`t handle event jobExecutionId: {} recordId: {}",
+        eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);
       future.completeExceptionally(e);
     }
     return future;
diff --git a/src/main/java/org/folio/processing/events/services/processor/EventProcessorImpl.java b/src/main/java/org/folio/processing/events/services/processor/EventProcessorImpl.java
@@ -16,6 +16,7 @@
 import static org.folio.processing.events.EventManager.OL_ACCUMULATIVE_RESULTS;
 import static org.folio.processing.events.EventManager.POST_PROCESSING_INDICATOR;
 import static org.folio.processing.events.EventManager.POST_PROCESSING_RESULT_EVENT_KEY;
+import static org.folio.processing.events.utils.EventUtils.extractRecordId;
 
 public class EventProcessorImpl implements EventProcessor {
 
@@ -25,7 +26,8 @@ public class EventProcessorImpl implements EventProcessor {
 
   @Override
   public CompletableFuture<DataImportEventPayload> process(DataImportEventPayload eventPayload) {
-    LOG.debug("process:: Processing event payload {}", eventPayload);
+    LOG.debug("process:: Processing event payload jobExecutionId: {} recordId: {}",
+      eventPayload.getJobExecutionId(), extractRecordId(eventPayload));
     CompletableFuture<DataImportEventPayload> future = new CompletableFuture<>();
     try {
       Optional<EventHandler> optionalEventHandler = eventHandlers.stream()
@@ -41,18 +43,22 @@ public CompletableFuture<DataImportEventPayload> process(DataImportEventPayload
           .whenComplete((payload, throwable) -> {
             logEventProcessingTime(eventType, startTime, eventPayload);
             if (throwable != null) {
-              LOG.warn("process:: Failed to process event payload", throwable);
+              LOG.warn("process:: Failed to process event payload jobExecutionId: {} recordId: {}",
+                eventPayload.getJobExecutionId(), extractRecordId(eventPayload), throwable);
               future.completeExceptionally(throwable);
             } else {
               future.complete(payload);
             }
           });
       } else {
-        LOG.info("process:: No suitable handler found for {} event type and current profile {}", eventPayload.getEventType(), eventPayload.getCurrentNode().getContentType());
+        LOG.info("process:: No suitable handler found for {} event type and current profile {} jobExecutionId: {} recordId: {}",
+          eventPayload.getEventType(), eventPayload.getCurrentNode().getContentType(),
+          eventPayload.getJobExecutionId(), extractRecordId(eventPayload));
         future.completeExceptionally(new EventHandlerNotFoundException(format("No suitable handler found for %s event type", eventPayload.getEventType())));
       }
     } catch (Exception e) {
-      LOG.warn("process:: Failed to process event payload", e);
+      LOG.warn("process:: Failed to process event payload jobExecutionId: {} recordId: {}",
+        eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);
       future.completeExceptionally(e);
     }
     return future;
@@ -75,14 +81,17 @@ private void logEventProcessingTime(String eventType, long startTime, DataImport
       var endTime = System.nanoTime();
       final String lastEvent = getLastEvent(eventPayload);
       if (DI_SRS_MARC_AUTHORITY_RECORD_CREATED.value().equals(lastEvent)) {
-        LOG.debug("logEventProcessingTime:: Event '{}' has been processed for {} ms", lastEvent, (endTime - startTime) / 1000000L);
+        LOG.debug("logEventProcessingTime:: Event '{}' has been processed for {} ms jobExecutionId: {} recordId: {}",
+          lastEvent, (endTime - startTime) / 1000000L, eventPayload.getJobExecutionId(), extractRecordId(eventPayload));
       } else {
         String profileType = eventPayload.getCurrentNode().getContentType().toString();
         String profileId = eventPayload.getCurrentNode().getProfileId();
-        LOG.debug("logEventProcessingTime:: Event '{}' has been processed using {} with id '{}' for {} ms", eventType, profileType, profileId, (endTime - startTime) / 1000000L);
+        LOG.debug("logEventProcessingTime:: Event '{}' has been processed using {} with id '{}' for {} ms jobExecutionId: {} recordId: {}",
+          eventType, profileType, profileId, (endTime - startTime) / 1000000L, eventPayload.getJobExecutionId(), extractRecordId(eventPayload));
       }
     } catch (Exception e) {
-      LOG.warn("logEventProcessingTime:: An Exception occurred {}", e.getMessage());
+      LOG.warn("logEventProcessingTime:: An Exception occurred {} jobExecutionId: {} recordId: {}",
+        e.getMessage(), eventPayload.getJobExecutionId(), extractRecordId(eventPayload));
     }
   }
 
diff --git a/src/main/java/org/folio/processing/events/services/publisher/KafkaEventPublisher.java b/src/main/java/org/folio/processing/events/services/publisher/KafkaEventPublisher.java
@@ -24,6 +24,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
 
+import static org.folio.rest.util.OkapiConnectionParams.OKAPI_REQUEST_ID_HEADER;
 import static org.folio.rest.util.OkapiConnectionParams.OKAPI_TENANT_HEADER;
 import static org.folio.rest.util.OkapiConnectionParams.OKAPI_TOKEN_HEADER;
 import static org.folio.rest.util.OkapiConnectionParams.OKAPI_URL_HEADER;
@@ -115,6 +116,9 @@ private List<KafkaHeader> getHeaders(DataImportEventPayload eventPayload, String
     Optional.ofNullable(eventPayload.getContext())
       .map(it -> it.get(USER_ID_HEADER))
       .ifPresent(userId -> headers.add(KafkaHeader.header(USER_ID_HEADER, userId)));
+    Optional.ofNullable(eventPayload.getContext())
+      .map(it -> it.get(OKAPI_REQUEST_ID_HEADER))
+      .ifPresent(requestId -> headers.add(KafkaHeader.header(OKAPI_REQUEST_ID_HEADER, requestId)));
 
     headers.add(KafkaHeader.header(OKAPI_URL_HEADER, eventPayload.getOkapiUrl()));
     headers.add(KafkaHeader.header(OKAPI_TENANT_HEADER, eventPayload.getTenant()));
diff --git a/src/main/java/org/folio/processing/events/utils/EventUtils.java b/src/main/java/org/folio/processing/events/utils/EventUtils.java
@@ -0,0 +1,27 @@
+package org.folio.processing.events.utils;
+
+import org.folio.DataImportEventPayload;
+import org.folio.processing.events.services.publisher.KafkaEventPublisher;
+
+/**
+ * Utility class for event-related operations
+ */
+public final class EventUtils {
+
+  private EventUtils() {
+  }
+
+  /**
+   * Safely extracts recordId from event payload context.
+   * Returns empty string if recordId is null or context is null.
+   *
+   * @param eventPayload the event payload
+   * @return recordId
+   */
+  public static String extractRecordId(DataImportEventPayload eventPayload) {
+    if (eventPayload == null || eventPayload.getContext() == null) {
+      return "";
+    }
+    return eventPayload.getContext().get(KafkaEventPublisher.RECORD_ID_HEADER);
+  }
+}
diff --git a/src/main/java/org/folio/processing/mapping/MappingManager.java b/src/main/java/org/folio/processing/mapping/MappingManager.java
@@ -18,6 +18,7 @@
 
 import java.util.Map;
 
+import static org.folio.processing.events.utils.EventUtils.extractRecordId;
 import static org.folio.rest.jaxrs.model.ProfileType.MAPPING_PROFILE;
 
 /**
@@ -47,7 +48,8 @@ private MappingManager() {
   public static DataImportEventPayload map(DataImportEventPayload eventPayload, MappingContext mappingContext) {
     try {
       if (eventPayload.getCurrentNode().getContentType() != MAPPING_PROFILE) {
-        LOGGER.info("map:: Current node is not of {} content type", MAPPING_PROFILE);
+        LOGGER.info("map:: Current node is not of {} content type jobExecutionId: {} recordId: {}",
+          MAPPING_PROFILE, eventPayload.getJobExecutionId(), extractRecordId(eventPayload));
         return eventPayload;
       }
       ProfileSnapshotWrapper mappingProfileWrapper = eventPayload.getCurrentNode();
@@ -65,7 +67,8 @@ public static DataImportEventPayload map(DataImportEventPayload eventPayload, Ma
       Mapper mapper = FACTORY_REGISTRY.createMapper(eventPayload, reader, writer);
       return mapper.map(mappingProfile, eventPayload, mappingContext);
     } catch (Exception e) {
-      LOGGER.warn("map:: Failed to perform mapping", e);
+      LOGGER.warn("map:: Failed to perform mapping jobExecutionId: {} recordId: {}",
+        eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);
       throw new MappingException(e);
     }
   }
diff --git a/src/main/java/org/folio/processing/mapping/mapper/mappers/AbstractMapper.java b/src/main/java/org/folio/processing/mapping/mapper/mappers/AbstractMapper.java
@@ -15,6 +15,8 @@
 import java.io.IOException;
 import java.util.List;
 
+import static org.folio.processing.events.utils.EventUtils.extractRecordId;
+
 public class AbstractMapper implements Mapper {
   private static final Logger LOGGER = LogManager.getLogger(AbstractMapper.class);
 
@@ -42,7 +44,8 @@ public DataImportEventPayload map(MappingProfile profile, DataImportEventPayload
       }
       return writer.getResult(eventPayload);
     } catch (IOException e) {
-      LOGGER.warn("map:: Failed to perform Abstract mapping", e);
+      LOGGER.warn("map:: Failed to perform Abstract mapping jobExecutionId: {} recordId: {}",
+        eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);
       throw new MappingException(e);
     }
   }
diff --git a/src/main/java/org/folio/processing/mapping/mapper/mappers/HoldingsMapper.java b/src/main/java/org/folio/processing/mapping/mapper/mappers/HoldingsMapper.java
@@ -22,6 +22,7 @@
 import java.util.Objects;
 import java.util.Optional;
 
+import static org.folio.processing.events.utils.EventUtils.extractRecordId;
 import static org.folio.processing.mapping.mapper.reader.record.marc.MarcRecordReader.EXPRESSIONS_DIVIDER;
 import static org.folio.processing.mapping.mapper.reader.record.marc.MarcRecordReader.MARC_PATTERN;
 
@@ -48,7 +49,8 @@ public DataImportEventPayload map(MappingProfile profile, DataImportEventPayload
       }
       return executeMultipleHoldingsLogic(eventPayload, profile, mappingContext);
     } catch (IOException e) {
-      LOGGER.warn("map:: Failed to perform Holdings mapping", e);
+      LOGGER.warn("map:: Failed to perform Holdings mapping jobExecutionId: {} recordId: {}",
+        eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);
       throw new MappingException(e);
     }
   }
diff --git a/src/main/java/org/folio/processing/mapping/mapper/mappers/ItemMapper.java b/src/main/java/org/folio/processing/mapping/mapper/mappers/ItemMapper.java
@@ -18,6 +18,7 @@
 import java.util.HashMap;
 import java.util.List;
 
+import static org.folio.processing.events.utils.EventUtils.extractRecordId;
 import static org.folio.processing.mapping.mapper.mappers.HoldingsMapper.MULTIPLE_HOLDINGS_FIELD;
 import static org.folio.rest.jaxrs.model.EntityType.ITEM;
 
@@ -40,7 +41,8 @@ public DataImportEventPayload map(MappingProfile profile, DataImportEventPayload
       }
       return executeMultipleItemsLogic(eventPayload, profile, mappingContext);
     } catch (IOException e) {
-      LOGGER.warn("map:: Failed to perform Items mapping", e);
+      LOGGER.warn("map:: Failed to perform Items mapping jobExecutionId: {} recordId: {}",
+        eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);
       throw new MappingException(e);
     }
   }
diff --git a/src/main/java/org/folio/processing/matching/MatchingManager.java b/src/main/java/org/folio/processing/matching/MatchingManager.java
@@ -1,5 +1,6 @@
 package org.folio.processing.matching;
 
+import static org.folio.processing.events.utils.EventUtils.extractRecordId;
 import static org.folio.rest.jaxrs.model.ProfileType.MATCH_PROFILE;
 
 import java.util.Map;
@@ -36,7 +37,8 @@ public static CompletableFuture<Boolean> match(DataImportEventPayload eventPaylo
     CompletableFuture<Boolean> future = new CompletableFuture<>();
     try {
       if (eventPayload.getCurrentNode().getContentType() != MATCH_PROFILE) {
-        LOGGER.info("match:: Current node is not of {} content type", MATCH_PROFILE);
+        LOGGER.info("match:: Current node is not of {} content type jobExecutionId: {} recordId: {}",
+          MATCH_PROFILE, eventPayload.getJobExecutionId(), extractRecordId(eventPayload));
         future.complete(false);
         return future;
       }
@@ -53,7 +55,8 @@ public static CompletableFuture<Boolean> match(DataImportEventPayload eventPaylo
 
       return matcher.match(eventPayload);
     } catch (Exception e) {
-      LOGGER.warn("match:: Failed to perform matching", e);
+      LOGGER.warn("match:: Failed to perform matching jobExecutionId: {} recordId: {}",
+        eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);
       future.completeExceptionally(new MatchingException(e));
     }
     return future;
diff --git a/src/main/java/org/folio/rest/util/OkapiConnectionParams.java b/src/main/java/org/folio/rest/util/OkapiConnectionParams.java
@@ -9,6 +9,7 @@ public final class OkapiConnectionParams {
   public static final String OKAPI_TENANT_HEADER = "x-okapi-tenant";
   public static final String OKAPI_TOKEN_HEADER = "x-okapi-token";
   public static final String USER_ID_HEADER = "userId";
+  public static final String OKAPI_REQUEST_ID_HEADER = "x-okapi-request-id";
   private String okapiUrl;
   private String tenantId;
   private String token;
diff --git a/src/test/java/org/folio/processing/events/EventManagerTest.java b/src/test/java/org/folio/processing/events/EventManagerTest.java
@@ -1,18 +1,30 @@
 package org.folio.processing.events;
 
 import io.vertx.core.Vertx;
+import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.RunTestOnContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
+import org.folio.DataImportEventPayload;
 import org.folio.kafka.KafkaConfig;
 import org.folio.processing.TestUtil;
+import org.folio.rest.jaxrs.model.ProfileSnapshotWrapper;
+import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.testcontainers.kafka.KafkaContainer;
 
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+
+import static org.folio.rest.jaxrs.model.ProfileType.ACTION_PROFILE;
+import static org.folio.rest.jaxrs.model.ProfileType.JOB_PROFILE;
+
 @RunWith(VertxUnitRunner.class)
 public class EventManagerTest {
   private static final String KAFKA_ENV = "folio";
@@ -33,6 +45,11 @@ public static void setUpClass() {
         .build();
   }
 
+  @Before
+  public void setUp() {
+    EventManager.clearEventHandlers();
+  }
+
   @Test
   public void registerKafkaEventPublisher(TestContext context) {
     Vertx vertx = rule.vertx();
@@ -41,4 +58,38 @@ public void registerKafkaEventPublisher(TestContext context) {
     EventManager.registerKafkaEventPublisher(kafkaConfig, vertx, 100);
     context.assertEquals(1, EventManager.getEventPublishers().size());
   }
+
+  @Test
+  public void shouldCompleteSuccessfullyIfNoEventHandlersFound(TestContext context) {
+    Async async = context.async();
+    // given
+    DataImportEventPayload eventPayload = new DataImportEventPayload()
+      .withEventType("DI_SRS_MARC_BIB_RECORD_CREATED")
+      .withTenant("diku")
+      .withOkapiUrl("http://localhost:9130")
+      .withToken("token")
+      .withJobExecutionId(UUID.randomUUID().toString())
+      .withContext(new HashMap<>());
+
+    ProfileSnapshotWrapper child = new ProfileSnapshotWrapper()
+      .withId(UUID.randomUUID().toString())
+      .withProfileId(UUID.randomUUID().toString())
+      .withContentType(ACTION_PROFILE);
+
+    ProfileSnapshotWrapper profileSnapshotWrapper = new ProfileSnapshotWrapper()
+      .withId(UUID.randomUUID().toString())
+      .withProfileId(UUID.randomUUID().toString())
+      .withContentType(JOB_PROFILE)
+      .withChildSnapshotWrappers(Collections.singletonList(child));
+
+    // when
+    CompletableFuture<DataImportEventPayload> future = EventManager.handleEvent(eventPayload, profileSnapshotWrapper);
+
+    // then
+    future.whenComplete((payload, throwable) -> {
+      context.assertNull(throwable);
+      context.assertNotNull(payload);
+      async.complete();
+    });
+  }
 }
diff --git a/src/test/java/org/folio/processing/events/services/publisher/KafkaEventPublisherTest.java b/src/test/java/org/folio/processing/events/services/publisher/KafkaEventPublisherTest.java
@@ -29,6 +29,7 @@
 import static org.folio.processing.events.services.publisher.KafkaEventPublisher.CHUNK_ID_HEADER;
 import static org.folio.processing.events.services.publisher.KafkaEventPublisher.PERMISSIONS_HEADER;
 import static org.folio.processing.events.services.publisher.KafkaEventPublisher.RECORD_ID_HEADER;
+import static org.folio.rest.util.OkapiConnectionParams.OKAPI_REQUEST_ID_HEADER;
 import static org.folio.rest.util.OkapiConnectionParams.USER_ID_HEADER;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -69,6 +70,7 @@ public void shouldPublishPayload() throws Exception {
     String expectedUserId = UUID.randomUUID().toString();
     String expectedRecordId = UUID.randomUUID().toString();
     String expectedChunkId = UUID.randomUUID().toString();
+    String expectedRequestId = UUID.randomUUID().toString();
     try(KafkaEventPublisher eventPublisher = new KafkaEventPublisher(kafkaConfig, vertx, 100)) {
       DataImportEventPayload eventPayload = new DataImportEventPayload()
         .withEventType(DI_COMPLETED.value())
@@ -80,6 +82,7 @@ public void shouldPublishPayload() throws Exception {
           put(CHUNK_ID_HEADER, expectedChunkId);
           put(PERMISSIONS_HEADER, expectedPermissionsHeader);
           put(USER_ID_HEADER, expectedUserId);
+          put(OKAPI_REQUEST_ID_HEADER, expectedRequestId);
         }});
 
       CompletableFuture<Event> future = eventPublisher.publish(eventPayload);
@@ -91,6 +94,7 @@ public void shouldPublishPayload() throws Exception {
       assertEquals(expectedUserId, actualPayload.getContext().get(USER_ID_HEADER));
       assertEquals(expectedRecordId, actualPayload.getContext().get(RECORD_ID_HEADER));
       assertEquals(expectedChunkId, actualPayload.getContext().get(CHUNK_ID_HEADER));
+      assertEquals(expectedRequestId, actualPayload.getContext().get(OKAPI_REQUEST_ID_HEADER));
 
       assertFalse(future.isCompletedExceptionally());
     }

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,8 @@`
`15`	`15`	`import java.io.IOException;`
`16`	`16`	`import java.util.List;`
`17`	`17`
	`18`	`+import static org.folio.processing.events.utils.EventUtils.extractRecordId;`
	`19`	`+`
`18`	`20`	`public class AbstractMapper implements Mapper {`
`19`	`21`	`private static final Logger LOGGER = LogManager.getLogger(AbstractMapper.class);`
`20`	`22`
`@@ -42,7 +44,8 @@ public DataImportEventPayload map(MappingProfile profile, DataImportEventPayload`
`42`	`44`	`}`
`43`	`45`	`return writer.getResult(eventPayload);`
`44`	`46`	`} catch (IOException e) {`
`45`		`- LOGGER.warn("map:: Failed to perform Abstract mapping", e);`
	`47`	`+ LOGGER.warn("map:: Failed to perform Abstract mapping jobExecutionId: {} recordId: {}",`
	`48`	`+ eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);`
`46`	`49`	`throw new MappingException(e);`
`47`	`50`	`}`
`48`	`51`	`}`
Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,7 @@`
`22`	`22`	`import java.util.Objects;`
`23`	`23`	`import java.util.Optional;`
`24`	`24`
	`25`	`+import static org.folio.processing.events.utils.EventUtils.extractRecordId;`
`25`	`26`	`import static org.folio.processing.mapping.mapper.reader.record.marc.MarcRecordReader.EXPRESSIONS_DIVIDER;`
`26`	`27`	`import static org.folio.processing.mapping.mapper.reader.record.marc.MarcRecordReader.MARC_PATTERN;`
`27`	`28`
`@@ -48,7 +49,8 @@ public DataImportEventPayload map(MappingProfile profile, DataImportEventPayload`
`48`	`49`	`}`
`49`	`50`	`return executeMultipleHoldingsLogic(eventPayload, profile, mappingContext);`
`50`	`51`	`} catch (IOException e) {`
`51`		`- LOGGER.warn("map:: Failed to perform Holdings mapping", e);`
	`52`	`+ LOGGER.warn("map:: Failed to perform Holdings mapping jobExecutionId: {} recordId: {}",`
	`53`	`+ eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);`
`52`	`54`	`throw new MappingException(e);`
`53`	`55`	`}`
`54`	`56`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`import java.util.HashMap;`
`19`	`19`	`import java.util.List;`
`20`	`20`
	`21`	`+import static org.folio.processing.events.utils.EventUtils.extractRecordId;`
`21`	`22`	`import static org.folio.processing.mapping.mapper.mappers.HoldingsMapper.MULTIPLE_HOLDINGS_FIELD;`
`22`	`23`	`import static org.folio.rest.jaxrs.model.EntityType.ITEM;`
`23`	`24`
`@@ -40,7 +41,8 @@ public DataImportEventPayload map(MappingProfile profile, DataImportEventPayload`
`40`	`41`	`}`
`41`	`42`	`return executeMultipleItemsLogic(eventPayload, profile, mappingContext);`
`42`	`43`	`} catch (IOException e) {`
`43`		`- LOGGER.warn("map:: Failed to perform Items mapping", e);`
	`44`	`+ LOGGER.warn("map:: Failed to perform Items mapping jobExecutionId: {} recordId: {}",`
	`45`	`+ eventPayload.getJobExecutionId(), extractRecordId(eventPayload), e);`
`44`	`46`	`throw new MappingException(e);`
`45`	`47`	`}`
`46`	`48`	`}`