MODDATAIMP-1145 Import of MARC records fails with "Invalid token" err…

…or. (#793)

* MODINV-1001 The sorting for Items on Instance details page is not worked

* add userId during create instance event handler

* add userId header in some handlers

* fix headers

* remove wildcards

src/main/java/org/folio/inventory/dataimport/consumers/MarcBibInstanceHridSetKafkaHandler.java

@@ -2,6 +2,8 @@
 
 import static java.lang.String.format;
 import static org.apache.commons.lang3.StringUtils.isEmpty;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_REQUEST_ID;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_USER_ID;
 import static org.folio.inventory.dataimport.util.MappingConstants.MARC_BIB_RECORD_FORMAT;
 import static org.folio.rest.util.OkapiConnectionParams.OKAPI_TENANT_HEADER;
 import static org.folio.rest.util.OkapiConnectionParams.OKAPI_TOKEN_HEADER;
@@ -106,7 +108,8 @@ public Future<String> handle(KafkaConsumerRecord<String, String> record) {
 
   private Future<Instance> updateInstance(HashMap<String, String> eventPayload, Record marcRecord, Map<String, String> headersMap) {
     String tenantId = eventPayload.getOrDefault(CENTRAL_TENANT_ID_KEY, eventPayload.get(OKAPI_TENANT_HEADER));
-    Context context = EventHandlingUtil.constructContext(tenantId, headersMap.get(OKAPI_TOKEN_HEADER), headersMap.get(OKAPI_URL_HEADER));
+    Context context = EventHandlingUtil.constructContext(tenantId, headersMap.get(OKAPI_TOKEN_HEADER), headersMap.get(OKAPI_URL_HEADER),
+      headersMap.get(OKAPI_USER_ID), headersMap.get(OKAPI_REQUEST_ID));
     return instanceUpdateDelegate.handle(eventPayload, marcRecord, context);
   }
 

src/main/java/org/folio/inventory/dataimport/consumers/MarcBibUpdateKafkaHandler.java

@@ -3,6 +3,8 @@
 import static java.lang.String.format;
 import static java.util.Objects.isNull;
 import static org.folio.inventory.EntityLinksKafkaTopic.LINKS_STATS;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_REQUEST_ID;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_USER_ID;
 import static org.folio.inventory.dataimport.util.AdditionalFieldsUtil.SUBFIELD_I;
 import static org.folio.inventory.dataimport.util.AdditionalFieldsUtil.TAG_999;
 import static org.folio.inventory.dataimport.util.MappingConstants.MARC_BIB_RECORD_TYPE;
@@ -55,8 +57,6 @@ public class MarcBibUpdateKafkaHandler implements AsyncRecordHandler<String, Str
   private static final String MAPPING_RULES_KEY = "MAPPING_RULES";
   private static final String MAPPING_PARAMS_KEY = "MAPPING_PARAMS";
   private static final String CURRENT_RETRY_NUMBER = "CURRENT_RETRY_NUMBER";
-  private static final String OKAPI_USER_ID = "x-okapi-user-id";
-  private static final String OKAPI_REQUEST_ID = "x-okapi-request-id";
   private static final int MAX_RETRIES_COUNT = Integer.parseInt(System.getenv().getOrDefault("inventory.di.ol.retry.number", "1"));
 
   private final InstanceUpdateDelegate instanceUpdateDelegate;
@@ -91,7 +91,8 @@ public Future<String> handle(KafkaConsumerRecord<String, String> consumerRecord)
         LOGGER.error(message);
         return Future.failedFuture(message);
       }
-      Context context = EventHandlingUtil.constructContext(instanceEvent.getTenant(), headersMap.get(OKAPI_TOKEN_HEADER), headersMap.get(OKAPI_URL_HEADER), headersMap.get(OKAPI_USER_ID), headersMap.get(OKAPI_REQUEST_ID));
+      Context context = EventHandlingUtil.constructContext(instanceEvent.getTenant(), headersMap.get(OKAPI_TOKEN_HEADER), headersMap.get(OKAPI_URL_HEADER),
+        headersMap.get(OKAPI_USER_ID), headersMap.get(OKAPI_REQUEST_ID));
       Record marcBibRecord = instanceEvent.getRecord();
 
       io.vertx.core.Context vertxContext = Vertx.currentContext();

src/main/java/org/folio/inventory/dataimport/consumers/MarcHoldingsRecordHridSetKafkaHandler.java

@@ -3,6 +3,8 @@
 import static java.lang.String.format;
 
 import static org.apache.commons.lang3.StringUtils.isEmpty;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_USER_ID;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_REQUEST_ID;
 import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.constructContext;
 import static org.folio.rest.util.OkapiConnectionParams.OKAPI_TENANT_HEADER;
 import static org.folio.rest.util.OkapiConnectionParams.OKAPI_TOKEN_HEADER;
@@ -89,7 +91,7 @@ public Future<String> handle(KafkaConsumerRecord<String, String> record) {
         }
 
         Context context = constructContext(headersMap.get(OKAPI_TENANT_HEADER), headersMap.get(OKAPI_TOKEN_HEADER),
-          headersMap.get(OKAPI_URL_HEADER));
+          headersMap.get(OKAPI_URL_HEADER), headersMap.get(OKAPI_USER_ID), headersMap.get(OKAPI_REQUEST_ID));
         Record marcRecord = Json.decodeValue(eventPayload.get(MARC_KEY), Record.class);
 
         mappingMetadataCache.get(jobExecutionId, context)

src/main/java/org/folio/inventory/dataimport/consumers/QuickMarcKafkaHandler.java

@@ -4,6 +4,8 @@
 import static org.folio.inventory.dataimport.handlers.QMEventTypes.QM_INVENTORY_AUTHORITY_UPDATED;
 import static org.folio.inventory.dataimport.handlers.QMEventTypes.QM_INVENTORY_HOLDINGS_UPDATED;
 import static org.folio.inventory.dataimport.handlers.QMEventTypes.QM_INVENTORY_INSTANCE_UPDATED;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_USER_ID;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.OKAPI_REQUEST_ID;
 import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.constructContext;
 import static org.folio.kafka.KafkaHeaderUtils.kafkaHeadersFromMap;
 import static org.folio.kafka.KafkaHeaderUtils.kafkaHeadersToMap;
@@ -79,8 +81,10 @@ public QuickMarcKafkaHandler(Vertx vertx, Storage storage, int maxDistributionNu
 
   @Override
   public Future<String> handle(KafkaConsumerRecord<String, String> record) {
-    var params = new OkapiConnectionParams(kafkaHeadersToMap(record.headers()), vertx);
-    var context = constructContext(params.getTenantId(), params.getToken(), params.getOkapiUrl());
+    var kafkaHeaders = kafkaHeadersToMap(record.headers());
+    var params = new OkapiConnectionParams(kafkaHeaders, vertx);
+    var context = constructContext(params.getTenantId(), params.getToken(), params.getOkapiUrl(),
+      kafkaHeaders.get(OKAPI_USER_ID), kafkaHeaders.get(OKAPI_REQUEST_ID));
     Event event = Json.decodeValue(record.value(), Event.class);
     LOGGER.info("Quick marc event payload has been received with event type: {}", event.getEventType());
     return getEventPayload(event)

src/main/java/org/folio/inventory/dataimport/handlers/actions/AbstractAuthorityEventHandler.java

@@ -3,6 +3,7 @@
 import static io.vertx.core.json.JsonObject.mapFrom;
 import static java.lang.String.format;
 
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.PAYLOAD_USER_ID;
 import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.constructContext;
 import static org.folio.inventory.dataimport.util.LoggerUtil.logParametersEventHandler;
 import static org.folio.rest.jaxrs.model.ProfileType.ACTION_PROFILE;
@@ -28,7 +29,6 @@
 import org.folio.MappingProfile;
 import org.folio.inventory.common.Context;
 import org.folio.inventory.dataimport.cache.MappingMetadataCache;
-import org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil;
 import org.folio.inventory.domain.AuthorityRecordCollection;
 import org.folio.inventory.storage.Storage;
 import org.folio.inventory.validation.exceptions.JsonMappingException;
@@ -79,7 +79,7 @@ public CompletableFuture<DataImportEventPayload> handle(DataImportEventPayload p
 
       prepareEvent(payload);
 
-      var context = constructContext(payload.getTenant(), payload.getToken(), payload.getOkapiUrl(), payload.getContext().get(EventHandlingUtil.USER_ID));
+      var context = constructContext(payload.getTenant(), payload.getToken(), payload.getOkapiUrl(), payload.getContext().get(PAYLOAD_USER_ID));
       var jobExecutionId = payload.getJobExecutionId();
       mappingMetadataCache.get(jobExecutionId, context)
         .map(mapMetadataOrFail())

src/main/java/org/folio/inventory/dataimport/handlers/actions/CreateHoldingEventHandler.java

@@ -16,7 +16,6 @@
 import org.folio.inventory.consortium.services.ConsortiumService;
 import org.folio.inventory.dataimport.cache.MappingMetadataCache;
 import org.folio.inventory.dataimport.entities.PartialError;
-import org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil;
 import org.folio.inventory.dataimport.services.OrderHelperService;
 import org.folio.inventory.consortium.util.ConsortiumUtil;
 import org.folio.inventory.dataimport.util.ParsedRecordUtil;
@@ -46,6 +45,7 @@
 import static org.folio.ActionProfile.FolioRecord.HOLDINGS;
 import static org.folio.ActionProfile.FolioRecord.MARC_BIBLIOGRAPHIC;
 import static org.folio.DataImportEventTypes.DI_INVENTORY_HOLDING_CREATED;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.PAYLOAD_USER_ID;
 import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.constructContext;
 import static org.folio.inventory.dataimport.util.DataImportConstants.UNIQUE_ID_ERROR_MESSAGE;
 import static org.folio.rest.jaxrs.model.ProfileType.ACTION_PROFILE;
@@ -97,7 +97,8 @@ public CompletableFuture<DataImportEventPayload> handle(DataImportEventPayload d
         return CompletableFuture.failedFuture(new EventProcessingException(ACTION_HAS_NO_MAPPING_MSG));
       }
 
-      Context context = constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(EventHandlingUtil.USER_ID));
+      Context context = constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(),
+        payloadContext.get(PAYLOAD_USER_ID));
       String jobExecutionId = dataImportEventPayload.getJobExecutionId();
       String recordId = payloadContext.get(RECORD_ID_HEADER);
       String chunkId = payloadContext.get(CHUNK_ID_HEADER);

src/main/java/org/folio/inventory/dataimport/handlers/actions/CreateInstanceEventHandler.java

@@ -8,6 +8,7 @@
 import static org.folio.ActionProfile.FolioRecord.MARC_BIBLIOGRAPHIC;
 import static org.folio.DataImportEventTypes.DI_INVENTORY_INSTANCE_CREATED;
 import static org.folio.DataImportEventTypes.DI_INVENTORY_INSTANCE_CREATED_READY_FOR_POST_PROCESSING;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.PAYLOAD_USER_ID;
 import static org.folio.inventory.dataimport.util.AdditionalFieldsUtil.SUBFIELD_I;
 import static org.folio.inventory.dataimport.util.AdditionalFieldsUtil.TAG_999;
 import static org.folio.inventory.dataimport.util.AdditionalFieldsUtil.reorderMarcRecordFields;
@@ -94,7 +95,8 @@ public CompletableFuture<DataImportEventPayload> handle(DataImportEventPayload d
         return CompletableFuture.failedFuture(new EventProcessingException(format(ACTION_HAS_NO_MAPPING_MSG, jobExecutionId, recordId)));
       }
 
-      Context context = EventHandlingUtil.constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(EventHandlingUtil.USER_ID));
+      Context context = EventHandlingUtil.constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(),
+        payloadContext.get(PAYLOAD_USER_ID));
       Record targetRecord = Json.decodeValue(payloadContext.get(EntityType.MARC_BIBLIOGRAPHIC.value()), Record.class);
       var sourceContent = targetRecord.getParsedRecord().getContent().toString();
 

src/main/java/org/folio/inventory/dataimport/handlers/actions/CreateItemEventHandler.java

@@ -56,6 +56,7 @@
 import static org.folio.ActionProfile.Action.CREATE;
 import static org.folio.ActionProfile.FolioRecord.ITEM;
 import static org.folio.DataImportEventTypes.DI_INVENTORY_ITEM_CREATED;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.PAYLOAD_USER_ID;
 import static org.folio.inventory.dataimport.util.DataImportConstants.UNIQUE_ID_ERROR_MESSAGE;
 import static org.folio.inventory.dataimport.util.LoggerUtil.logParametersEventHandler;
 import static org.folio.rest.jaxrs.model.ProfileType.ACTION_PROFILE;
@@ -134,7 +135,8 @@ public CompletableFuture<DataImportEventPayload> handle(DataImportEventPayload d
       Future<RecordToEntity> recordToItemFuture = idStorageService.store(recordId, UUID.randomUUID().toString(), dataImportEventPayload.getTenant());
       recordToItemFuture.onSuccess(res -> {
         String deduplicationItemId = res.getEntityId();
-        Context context = EventHandlingUtil.constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(EventHandlingUtil.USER_ID));
+        Context context = EventHandlingUtil.constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(),
+          payloadContext.get(PAYLOAD_USER_ID));
         ItemCollection itemCollection = storage.getItemCollection(context);
 
         mappingMetadataCache.get(jobExecutionId, context)

src/main/java/org/folio/inventory/dataimport/handlers/actions/CreateMarcHoldingsEventHandler.java

@@ -8,6 +8,7 @@
 import static org.folio.ActionProfile.FolioRecord.MARC_HOLDINGS;
 import static org.folio.DataImportEventTypes.DI_INVENTORY_HOLDINGS_CREATED_READY_FOR_POST_PROCESSING;
 import static org.folio.DataImportEventTypes.DI_INVENTORY_HOLDING_CREATED;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.PAYLOAD_USER_ID;
 import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.constructContext;
 import static org.folio.inventory.dataimport.util.DataImportConstants.UNIQUE_ID_ERROR_MESSAGE;
 import static org.folio.inventory.dataimport.util.LoggerUtil.logParametersEventHandler;
@@ -28,7 +29,6 @@
 import org.folio.MappingMetadataDto;
 import org.folio.inventory.common.Context;
 import org.folio.inventory.dataimport.cache.MappingMetadataCache;
-import org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil;
 import org.folio.inventory.domain.HoldingsRecordCollection;
 import org.folio.inventory.domain.relationship.RecordToEntity;
 import org.folio.inventory.services.HoldingsCollectionService;
@@ -100,7 +100,8 @@ public CompletableFuture<DataImportEventPayload> handle(DataImportEventPayload d
         return CompletableFuture.failedFuture(new EventProcessingException(ACTION_HAS_NO_MAPPING_MSG));
       }
 
-      Context context = constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(EventHandlingUtil.USER_ID));
+      Context context = constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(),
+        payloadContext.get(PAYLOAD_USER_ID));
       Record targetRecord = new JsonObject(payloadContext.get(EntityType.MARC_HOLDINGS.value())).mapTo(Record.class);
       prepareEvent(dataImportEventPayload);
 

src/main/java/org/folio/inventory/dataimport/handlers/actions/DeleteAuthorityEventHandler.java

@@ -10,7 +10,6 @@
 import org.folio.Authority;
 import org.folio.DataImportEventPayload;
 import org.folio.inventory.dataimport.exceptions.DataImportException;
-import org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil;
 import org.folio.inventory.domain.AuthorityRecordCollection;
 import org.folio.inventory.storage.Storage;
 import org.folio.processing.events.services.handler.EventHandler;
@@ -23,6 +22,7 @@
 import static java.lang.String.format;
 import static org.folio.ActionProfile.Action.DELETE;
 import static org.folio.ActionProfile.FolioRecord.MARC_AUTHORITY;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.PAYLOAD_USER_ID;
 import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.constructContext;
 import static org.folio.inventory.dataimport.util.LoggerUtil.logParametersEventHandler;
 import static org.folio.rest.jaxrs.model.ProfileType.ACTION_PROFILE;
@@ -55,7 +55,8 @@ public CompletableFuture<DataImportEventPayload> handle(DataImportEventPayload p
         throw new EventProcessingException(UNEXPECTED_PAYLOAD_MSG);
       }
 
-      var context = constructContext(payload.getTenant(), payload.getToken(), payload.getOkapiUrl(), payload.getContext().get(EventHandlingUtil.USER_ID));
+      var context = constructContext(payload.getTenant(), payload.getToken(), payload.getOkapiUrl(),
+        payload.getContext().get(PAYLOAD_USER_ID));
       AuthorityRecordCollection authorityRecordCollection = storage.getAuthorityRecordCollection(context);
       String id = payload.getContext().get(AUTHORITY_RECORD_ID);
       LOGGER.info("Delete authority with id: {}", id);

src/main/java/org/folio/inventory/dataimport/handlers/actions/MarcBibModifiedPostProcessingEventHandler.java

@@ -32,6 +32,7 @@
 import static org.apache.commons.lang3.StringUtils.isBlank;
 import static org.folio.ActionProfile.FolioRecord.INSTANCE;
 import static org.folio.DataImportEventTypes.DI_SRS_MARC_BIB_RECORD_MODIFIED_READY_FOR_POST_PROCESSING;
+import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.PAYLOAD_USER_ID;
 import static org.folio.inventory.dataimport.handlers.matching.util.EventHandlingUtil.getTenant;
 import static org.folio.inventory.dataimport.util.LoggerUtil.logParametersEventHandler;
 import static org.folio.rest.jaxrs.model.EntityType.MARC_BIBLIOGRAPHIC;
@@ -79,8 +80,8 @@ public CompletableFuture<DataImportEventPayload> handle(DataImportEventPayload d
       }
 
       record.setExternalIdsHolder(new ExternalIdsHolder().withInstanceId(instanceId));
-      Context localTenantContext = EventHandlingUtil.constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(EventHandlingUtil.USER_ID));
-      Context targetInstanceContext = EventHandlingUtil.constructContext(getTenant(dataImportEventPayload), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(EventHandlingUtil.USER_ID));
+      Context localTenantContext = EventHandlingUtil.constructContext(dataImportEventPayload.getTenant(), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(PAYLOAD_USER_ID));
+      Context targetInstanceContext = EventHandlingUtil.constructContext(getTenant(dataImportEventPayload), dataImportEventPayload.getToken(), dataImportEventPayload.getOkapiUrl(), payloadContext.get(PAYLOAD_USER_ID));
       Promise<Instance> instanceUpdatePromise = Promise.promise();
 
       mappingMetadataCache.get(dataImportEventPayload.getJobExecutionId(), localTenantContext)