MODINVSTOR-1283 Modify endpoint for bulk instances upsert with publish events flag

NEWS.md

@@ -1,4 +1,28 @@
-## v28.0.0 In progress
+## v28.1.0 YYYY-mm-DD
+### Breaking changes
+* Description ([ISSUE](https://folio-org.atlassian.net/browse/ISSUE))
+
+### New APIs versions
+* Provides `API_NAME vX.Y`
+* Requires `API_NAME vX.Y`
+
+### Features
+* Modify endpoint for bulk instances upsert with publish events flag ([MODINVSTOR-1283](https://folio-org.atlassian.net/browse/MODINVSTOR-1283))
+
+### Bug fixes
+* Description ([ISSUE](https://folio-org.atlassian.net/browse/ISSUE))
+
+### Tech Dept
+* Description ([ISSUE](https://folio-org.atlassian.net/browse/ISSUE))
+
+### Dependencies
+* Bump `LIB_NAME` from `OLD_VERSION` to `NEW_VERSION`
+* Add `LIB_NAME VERSION`
+* Remove `LIB_NAME`
+
+---
+
+## v28.0.0 2024-11-01
 ### Breaking changes
 * Migrate "publicationPeriod" data to the Dates object and remove it from the Instance schema ([MODINVSTOR-1232](https://folio-org.atlassian.net/browse/MODINVSTOR-1232))
 * Delete deprecated `shelf-locations` API ([MODINVSTOR-1183](https://folio-org.atlassian.net/browse/MODINVSTOR-1183))

ramls/bulkUpsertRequest.json

@@ -7,6 +7,11 @@
     "recordsFileName": {
       "description": "File name of entities records",
       "type": "string"
+    },
+    "publishEvents": {
+      "description": "A flag that indicates whether domain events should be published.",
+      "type": "boolean",
+      "default": true
     }
   },
   "required": [

src/main/java/org/folio/rest/impl/InstanceBatchSyncApi.java

@@ -25,7 +25,7 @@ public void postInstanceStorageBatchSynchronous(boolean upsert, InstancesPost en
     var instances = ObjectConverterUtils.convertObject(entity, Instances.class);
 
     new InstanceService(vertxContext, okapiHeaders)
-      .createInstances(instances.getInstances(), upsert, true)
+      .createInstances(instances.getInstances(), upsert, true, true)
       .otherwise(cause -> respond500WithTextPlain(cause.getMessage()))
       .onComplete(asyncResultHandler);
   }

src/main/java/org/folio/rest/impl/InstanceBatchSyncUnsafeApi.java

@@ -24,7 +24,7 @@ public void postInstanceStorageBatchSynchronousUnsafe(InstancesPost entity, Map<
     var instances = ObjectConverterUtils.convertObject(entity, Instances.class);
 
     new InstanceService(vertxContext, okapiHeaders)
-      .createInstances(instances.getInstances(), true, false)
+      .createInstances(instances.getInstances(), true, false, true)
       .otherwise(cause -> respond500WithTextPlain(cause.getMessage()))
       .onComplete(asyncResultHandler);
   }

src/main/java/org/folio/services/BulkProcessingContext.java

@@ -1,6 +1,7 @@
 package org.folio.services;
 
 import org.apache.commons.lang3.StringUtils;
+import org.folio.rest.jaxrs.model.BulkUpsertRequest;
 
 /**
  * Encapsulates the context for bulk processing of entities from external file.
@@ -17,13 +18,15 @@ public class BulkProcessingContext {
   private final String errorsFilePath;
   private final String errorEntitiesFileLocalPath;
   private final String errorsFileLocalPath;
+  private final boolean publishEvents;
 
-  public BulkProcessingContext(String entitiesFilePath) {
-    this.initialFilePath = StringUtils.removeStart(entitiesFilePath, '/');
+  public BulkProcessingContext(BulkUpsertRequest request) {
+    this.initialFilePath = StringUtils.removeStart(request.getRecordsFileName(), '/');
     this.errorEntitiesFilePath = initialFilePath + FAILED_ENTITIES_FILE_SUFFIX;
     this.errorsFilePath = initialFilePath + ERRORS_FILE_SUFFIX;
     this.errorEntitiesFileLocalPath = ROOT_FOLDER + errorEntitiesFilePath;
     this.errorsFileLocalPath = ROOT_FOLDER + errorsFilePath;
+    this.publishEvents = request.getPublishEvents();
   }
 
   public String getErrorEntitiesFilePath() {
@@ -42,4 +45,7 @@ public String getErrorsFileLocalPath() {
     return errorsFileLocalPath;
   }
 
+  public boolean isPublishEvents() {
+    return publishEvents;
+  }
 }

src/main/java/org/folio/services/batch/BatchOperationContext.java

@@ -11,9 +11,12 @@ public final class BatchOperationContext<T> {
    */
   private final Collection<T> existingRecords;
 
-  public BatchOperationContext(Collection<T> recordsToBeCreated, Collection<T> existingRecords) {
+  private final boolean publishEvents;
+
+  public BatchOperationContext(Collection<T> recordsToBeCreated, Collection<T> existingRecords, boolean publishEvents) {
     this.recordsToBeCreated = unmodifiableCollection(recordsToBeCreated);
     this.existingRecords = unmodifiableCollection(existingRecords);
+    this.publishEvents = publishEvents;
   }
 
   public Collection<T> getRecordsToBeCreated() {
@@ -23,4 +26,8 @@ public Collection<T> getRecordsToBeCreated() {
   public Collection<T> getExistingRecords() {
     return existingRecords;
   }
+
+  public boolean isPublishEvents() {
+    return publishEvents;
+  }
 }

src/main/java/org/folio/services/batch/BatchOperationContextFactory.java

@@ -12,19 +12,21 @@
 public final class BatchOperationContextFactory {
   private BatchOperationContextFactory() { }
 
-  public static <T> Future<BatchOperationContext<T>> buildBatchOperationContext(
-    boolean upsert, List<T> all, AbstractRepository<T> repository, Function<T, String> idGetter) {
+  public static <T> Future<BatchOperationContext<T>> buildBatchOperationContext(boolean upsert, List<T> all,
+                                                                                AbstractRepository<T> repository,
+                                                                                Function<T, String> idGetter,
+                                                                                boolean publishEvents) {
 
     if (!upsert) {
-      return succeededFuture(new BatchOperationContext<>(all, emptyList()));
+      return succeededFuture(new BatchOperationContext<>(all, emptyList(), publishEvents));
     }
 
     return repository.getById(all, idGetter).map(found -> {
       final var toBeCreated = all.stream()
         .filter(entity -> !found.containsKey(idGetter.apply(entity)))
         .collect(toList());
 
-      return new BatchOperationContext<>(toBeCreated, found.values());
+      return new BatchOperationContext<>(toBeCreated, found.values(), publishEvents);
     });
   }
 }

src/main/java/org/folio/services/bulkprocessing/AbstractEntityS3Service.java

@@ -77,9 +77,9 @@ private Future<List<T>> loadEntities(BulkUpsertRequest bulkRequest) {
   }
 
   private Future<BulkUpsertResponse> upsertEntities(List<T> entities, BulkUpsertRequest bulkRequest) {
-    BulkProcessingContext bulkProcessingContext = new BulkProcessingContext(bulkRequest.getRecordsFileName());
+    BulkProcessingContext bulkProcessingContext = new BulkProcessingContext(bulkRequest);
 
-    return upsert(entities)
+    return upsert(entities, bulkProcessingContext.isPublishEvents())
       .map(v -> new BulkUpsertResponse().withErrorsNumber(0))
       .recover(e -> processSequentially(entities, bulkProcessingContext));
   }
@@ -89,7 +89,7 @@ private Future<BulkUpsertResponse> processSequentially(List<T> entities, BulkPro
     BulkProcessingErrorFileWriter errorsWriter = new BulkProcessingErrorFileWriter(vertx, bulkContext);
 
     return errorsWriter.initialize()
-      .compose(v -> processInBatches(entities, entity -> upsert(List.of(entity))
+      .compose(v -> processInBatches(entities, entity -> upsert(List.of(entity), bulkContext.isPublishEvents())
         .recover(e -> handleUpsertFailure(errorsCounter, errorsWriter, entity, e))))
       .eventually(errorsWriter::close)
       .eventually(() -> uploadErrorsFiles(bulkContext))
@@ -166,10 +166,11 @@ private Future<Void> uploadErrorsFiles(BulkProcessingContext bulkContext) {
    * Performs an upsert operation on specified list of {@code entities}.
    * The implementation of the upsert operation depends on the specifics of the {@code <T>} type of entity.
    *
-   * @param entities - a list of entities to be updated or created
+   * @param entities      - a list of entities to be updated or created
+   * @param publishEvents - a flag that indicates whether domain events should be published
    * @return Future of Void, succeeded if the upsert operation is successful, otherwise failed
    */
-  protected abstract Future<Void> upsert(List<T> entities);
+  protected abstract Future<Void> upsert(List<T> entities, boolean publishEvents);
 
   /**
    * Provides a representation of the given {@code entity} to be written to error file containing entities

src/main/java/org/folio/services/bulkprocessing/InstanceS3Service.java

@@ -99,10 +99,10 @@ protected Future<Void> ensureEntitiesWithNonMarcControlledFieldsData(List<Instan
   }
 
   @Override
-  protected Future<Void> upsert(List<InstanceWrapper> instanceWrappers) {
+  protected Future<Void> upsert(List<InstanceWrapper> instanceWrappers, boolean publishEvents) {
     List<Instance> instances = instanceWrappers.stream().map(InstanceWrapper::instance).toList();
 
-    return instanceService.createInstances(instances, APPLY_UPSERT, APPLY_OPTIMISTIC_LOCKING)
+    return instanceService.createInstances(instances, APPLY_UPSERT, APPLY_OPTIMISTIC_LOCKING, publishEvents)
       .compose(response -> {
         if (!isCreateSuccessResponse(response)) {
           String msg = String.format("Failed to update instances, status: '%s', message: '%s'",

src/main/java/org/folio/services/domainevent/AbstractDomainEventPublisher.java

@@ -67,8 +67,10 @@ public Handler<Response> publishCreatedOrUpdated(BatchOperationContext<D> batchO
       log.info("Records created {}, records updated {}", batchOperation.getRecordsToBeCreated().size(),
         batchOperation.getExistingRecords().size());
 
-      publishRecordsCreated(batchOperation.getRecordsToBeCreated()).compose(
-        notUsed -> publishUpdated(batchOperation.getExistingRecords()));
+      if (batchOperation.isPublishEvents()) {
+        publishRecordsCreated(batchOperation.getRecordsToBeCreated()).compose(
+          notUsed -> publishUpdated(batchOperation.getExistingRecords()));
+      }
     };
   }
 

src/main/java/org/folio/services/holding/HoldingsService.java

@@ -168,7 +168,7 @@ public Future<Response> createHoldings(List<HoldingsRecord> holdings, boolean up
       .compose(ar -> hridManager.populateHridForHoldings(holdings)
         .compose(NotesValidators::refuseHoldingLongNotes)
         .compose(result -> buildBatchOperationContext(upsert, holdings,
-          holdingsRepository, HoldingsRecord::getId))
+          holdingsRepository, HoldingsRecord::getId, true))
         .compose(batchOperation -> postSync(HOLDINGS_RECORD_TABLE, holdings, MAX_ENTITIES,
           upsert, optimisticLocking, okapiHeaders, vertxContext, PostHoldingsStorageBatchSynchronousResponse.class)
           .onSuccess(domainEventPublisher.publishCreatedOrUpdated(batchOperation))))

src/main/java/org/folio/services/instance/InstanceService.java

@@ -120,14 +120,15 @@ public Future<Response> createInstance(Instance entity) {
       .map(ResponseHandlerUtil::handleHridError);
   }
 
-  public Future<Response> createInstances(List<Instance> instances, boolean upsert, boolean optimisticLocking) {
+  public Future<Response> createInstances(List<Instance> instances, boolean upsert, boolean optimisticLocking,
+                                          boolean publishEvents) {
     final String statusUpdatedDate = generateStatusUpdatedDate();
     instances.forEach(instance -> instance.setStatusUpdatedDate(statusUpdatedDate));
 
     return hridManager.populateHridForInstances(instances)
       .compose(NotesValidators::refuseInstanceLongNotes)
       .compose(notUsed -> buildBatchOperationContext(upsert, instances,
-        instanceRepository, Instance::getId))
+        instanceRepository, Instance::getId, publishEvents))
       .compose(batchOperation ->
         // Can use instances list here directly because the class is stateful
         postSync(INSTANCE_TABLE, instances, MAX_ENTITIES, upsert, optimisticLocking, okapiHeaders,

src/main/java/org/folio/services/item/ItemService.java

@@ -144,7 +144,7 @@ public Future<Response> createItems(List<Item> items, boolean upsert, boolean op
       .compose(NotesValidators::refuseItemLongNotes)
       .compose(result -> effectiveValuesService.populateEffectiveValues(items))
       .compose(this::populateCirculationNoteId)
-      .compose(result -> buildBatchOperationContext(upsert, items, itemRepository, Item::getId))
+      .compose(result -> buildBatchOperationContext(upsert, items, itemRepository, Item::getId, true))
       .compose(batchOperation -> postSync(ITEM_TABLE, items, MAX_ENTITIES, upsert, optimisticLocking,
         okapiHeaders, vertxContext, PostItemStorageBatchSynchronousResponse.class)
         .onSuccess(domainEventService.publishCreatedOrUpdated(batchOperation)))