Feat/MET-6035 Fixing depublication reason (#685)

* MET-6035 Fixed depublication reason while performing depublication

* MET-6035 Code review changes

metis-core/metis-core-service/src/main/java/eu/europeana/metis/core/dao/DepublishRecordIdDao.java

@@ -17,6 +17,7 @@
 import eu.europeana.metis.core.rest.RequestLimits;
 import eu.europeana.metis.core.util.DepublishRecordIdSortField;
 import eu.europeana.metis.core.util.SortDirection;
+import eu.europeana.metis.core.workflow.plugins.DepublicationReason;
 import eu.europeana.metis.exception.BadContentException;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -129,18 +130,19 @@ public int createRecordIdsToBeDepublished(String datasetId, Set<String> candidat
     }
 
     // Add the records and we're done.
-    addRecords(recordIdsToAdd, datasetId, DepublicationStatus.PENDING_DEPUBLICATION, null);
+    addRecords(recordIdsToAdd, datasetId, DepublicationStatus.PENDING_DEPUBLICATION, null, null);
     return recordIdsToAdd.size();
   }
 
   void addRecords(Set<String> recordIdsToAdd, String datasetId,
-      DepublicationStatus depublicationStatus, Instant depublicationDate) {
+      DepublicationStatus depublicationStatus, Instant depublicationDate, DepublicationReason depublicationReason) {
     final List<DepublishRecordId> objectsToAdd = recordIdsToAdd.stream().map(recordId -> {
       final DepublishRecordId depublishRecordId = new DepublishRecordId();
       depublishRecordId.setId(new ObjectId());
       depublishRecordId.setDatasetId(datasetId);
       depublishRecordId.setRecordId(recordId);
       depublishRecordId.setDepublicationStatus(depublicationStatus);
+      depublishRecordId.setDepublicationReason(depublicationReason);
       depublishRecordId.setDepublicationDate(depublicationDate);
       return depublishRecordId;
     }).toList();
@@ -328,7 +330,7 @@ private Query<DepublishRecordId> prepareQueryForDepublishRecordIds(String datase
    * {@link DepublicationStatus#PENDING_DEPUBLICATION}
    */
   public void markRecordIdsWithDepublicationStatus(String datasetId, Set<String> recordIds,
-      DepublicationStatus depublicationStatus, @Nullable Date depublicationDate) {
+      DepublicationStatus depublicationStatus, @Nullable Date depublicationDate, DepublicationReason depublicationReason) {
 
     // Check correctness of parameters
     if (Objects.isNull(depublicationStatus) || StringUtils.isBlank(datasetId)) {
@@ -339,6 +341,11 @@ public void markRecordIdsWithDepublicationStatus(String datasetId, Set<String> r
       throw new IllegalArgumentException(String
           .format("DepublicationDate cannot be null if depublicationStatus == %s ",
               DepublicationStatus.DEPUBLISHED.name()));
+    } else if(depublicationStatus == DepublicationStatus.DEPUBLISHED && Objects
+        .isNull(depublicationReason)){
+      throw new IllegalArgumentException(String
+          .format("DepublicationReason cannot be null if depublicationStatus == %s ",
+              DepublicationStatus.DEPUBLISHED.name()));
     }
 
     // If we have a specific record list, make sure that missing records are added.
@@ -353,7 +360,7 @@ public void markRecordIdsWithDepublicationStatus(String datasetId, Set<String> r
                                                    .filter(
                                                        date -> depublicationStatus != DepublicationStatus.PENDING_DEPUBLICATION)
                                                    .map(Date::toInstant).orElse(null);
-      addRecords(recordIdsToAdd, datasetId, depublicationStatus, depublicationInstant);
+      addRecords(recordIdsToAdd, datasetId, depublicationStatus, depublicationInstant, depublicationReason);
 
       // Compute the records to update - if there are none, we're done.
       recordIdsToUpdate = new HashSet<>(recordIds);

metis-core/metis-core-service/src/main/java/eu/europeana/metis/core/execution/WorkflowPostProcessor.java

@@ -123,13 +123,13 @@ private void reinstateDepublishRecordIdsStatus(IndexToPublishPlugin indexPlugin,
         if (!CollectionUtils.isEmpty(publishedRecordIds)) {
           depublishRecordIdDao.markRecordIdsWithDepublicationStatus(datasetId,
               publishedRecordIds.stream().map(depublishedRecordIdsByFullId::get)
-                                .collect(Collectors.toSet()), DepublicationStatus.PENDING_DEPUBLICATION, null);
+                                .collect(Collectors.toSet()), DepublicationStatus.PENDING_DEPUBLICATION, null, null);
         }
       }
     } else {
       // reset de-publish status, pass null, all records will be de-published
       depublishRecordIdDao.markRecordIdsWithDepublicationStatus(datasetId, null,
-          DepublicationStatus.PENDING_DEPUBLICATION, null);
+          DepublicationStatus.PENDING_DEPUBLICATION, null, null);
     }
   }
 
@@ -178,7 +178,7 @@ private void depublishRecordPostProcess(DepublishPlugin depublishPlugin, String
                                                                    Collectors.mapping(Pair::getRight, Collectors.toSet())));
     successfulRecords.forEach((dataset, records) ->
         depublishRecordIdDao.markRecordIdsWithDepublicationStatus(dataset, records,
-            DepublicationStatus.DEPUBLISHED, new Date()));
+            DepublicationStatus.DEPUBLISHED, new Date(), depublishPlugin.getPluginMetadata().getDepublicationReason()));
 
     // Set publication fitness to PARTIALLY FIT (if not set to the more severe UNFIT).
     final Dataset dataset = datasetDao.getDatasetByDatasetId(datasetId);
@@ -195,7 +195,7 @@ private void depublishDatasetPostProcess(String datasetId) {
 
     // Set all depublished records back to PENDING.
     depublishRecordIdDao.markRecordIdsWithDepublicationStatus(datasetId, null,
-        DepublicationStatus.PENDING_DEPUBLICATION, null);
+        DepublicationStatus.PENDING_DEPUBLICATION, null, null);
     // Find latest PUBLISH Type Plugin and set dataStatus to DELETED.
     final PluginWithExecutionId<MetisPlugin> latestSuccessfulPlugin = workflowExecutionDao
         .getLatestSuccessfulPlugin(datasetId, OrchestratorService.PUBLISH_TYPES);

metis-core/metis-core-service/src/test/java/eu/europeana/metis/core/dao/TestDepublishRecordIdDao.java

@@ -19,6 +19,7 @@
 import eu.europeana.metis.core.util.DepublishRecordIdSortField;
 import eu.europeana.metis.core.util.SortDirection;
 import eu.europeana.metis.core.utils.TestObjectFactory;
+import eu.europeana.metis.core.workflow.plugins.DepublicationReason;
 import eu.europeana.metis.exception.BadContentException;
 import eu.europeana.metis.mongo.embedded.EmbeddedLocalhostMongo;
 import java.time.Instant;
@@ -118,7 +119,7 @@ void deletePendingRecordIdsTest() throws BadContentException {
         () -> depublishRecordIdDao.deletePendingRecordIds(datasetId, biggerThanAllowedSet));
 
     depublishRecordIdDao
-        .addRecords(setTest, datasetId, DepublicationStatus.PENDING_DEPUBLICATION, Instant.now());
+        .addRecords(setTest, datasetId, DepublicationStatus.PENDING_DEPUBLICATION, Instant.now(), DepublicationReason.GENERIC);
     assertEquals(1, provider.getDatastore().find(DepublishRecordId.class).count());
 
     depublishRecordIdDao.deletePendingRecordIds(datasetId, setTest);
@@ -132,7 +133,7 @@ void countSuccessfullyDepublishedRecordIdsForDatasetTest() {
     final Set<String> setTest = Set.of("1003");
 
     depublishRecordIdDao
-        .addRecords(setTest, datasetId, DepublicationStatus.DEPUBLISHED, Instant.now());
+        .addRecords(setTest, datasetId, DepublicationStatus.DEPUBLISHED, Instant.now(), DepublicationReason.GENERIC);
     long result = depublishRecordIdDao.countSuccessfullyDepublishedRecordIdsForDataset(datasetId);
     assertEquals(1L, result);
   }
@@ -143,7 +144,7 @@ void getDepublishRecordIdsTest() {
     final Set<String> setTest = Set.of("1004", "1005");
 
     depublishRecordIdDao
-        .addRecords(setTest, datasetId, DepublicationStatus.PENDING_DEPUBLICATION, Instant.now());
+        .addRecords(setTest, datasetId, DepublicationStatus.PENDING_DEPUBLICATION, Instant.now(), DepublicationReason.GENERIC);
     List<DepublishRecordIdView> find1004 = depublishRecordIdDao
         .getDepublishRecordIds(datasetId, 0, DepublishRecordIdSortField.DEPUBLICATION_STATE,
             SortDirection.ASCENDING, "1004");
@@ -172,7 +173,7 @@ void getAllDepublishRecordIdsWithStatusTest() throws BadContentException {
             DepublicationStatus.DEPUBLISHED, biggerThanAllowedSet));
 
     depublishRecordIdDao
-        .addRecords(setTest, datasetId, DepublicationStatus.DEPUBLISHED, Instant.now());
+        .addRecords(setTest, datasetId, DepublicationStatus.DEPUBLISHED, Instant.now(), DepublicationReason.GENERIC);
     Set<String> result = depublishRecordIdDao.getAllDepublishRecordIdsWithStatus(datasetId,
         DepublishRecordIdSortField.DEPUBLICATION_STATE, SortDirection.ASCENDING,
         DepublicationStatus.DEPUBLISHED, setTest);
@@ -195,17 +196,17 @@ void markRecordIdsWithDepublicationStatus_wrong_parametersTest() {
 
     //Null depublication status
     assertThrows(IllegalArgumentException.class, () -> depublishRecordIdDao
-        .markRecordIdsWithDepublicationStatus(datasetId, recordIdsSet, null, date));
+        .markRecordIdsWithDepublicationStatus(datasetId, recordIdsSet, null, date, DepublicationReason.GENERIC));
 
     //Blank dataset id
     assertThrows(IllegalArgumentException.class, () -> depublishRecordIdDao
         .markRecordIdsWithDepublicationStatus(null, recordIdsSet,
-            DepublicationStatus.PENDING_DEPUBLICATION, date));
+            DepublicationStatus.PENDING_DEPUBLICATION, date, DepublicationReason.GENERIC));
 
     //Depublished status but date null
     assertThrows(IllegalArgumentException.class, () -> depublishRecordIdDao
         .markRecordIdsWithDepublicationStatus(datasetId, recordIdsSet,
-            DepublicationStatus.DEPUBLISHED, null));
+            DepublicationStatus.DEPUBLISHED, null, null));
   }
 
   @Test
@@ -227,7 +228,7 @@ void markRecordIdsWithDepublicationStatus_all_recordIds_set_depublished_and_then
     //Set to DEPUBLISHED
     depublishRecordIdDao
         .markRecordIdsWithDepublicationStatus(datasetId, null, DepublicationStatus.DEPUBLISHED,
-            date);
+            date, DepublicationReason.GENERIC);
     //Check stored recordIds
     findAll = depublishRecordIdDao
         .getDepublishRecordIds(datasetId, 0, DepublishRecordIdSortField.DEPUBLICATION_STATE,
@@ -238,7 +239,7 @@ void markRecordIdsWithDepublicationStatus_all_recordIds_set_depublished_and_then
             .equals(Date.from(depublishRecordIdView.getDepublicationDate()))));
     //Set to PENDING_DEPUBLICATION
     depublishRecordIdDao.markRecordIdsWithDepublicationStatus(datasetId, null,
-        DepublicationStatus.PENDING_DEPUBLICATION, date);
+        DepublicationStatus.PENDING_DEPUBLICATION, date, DepublicationReason.GENERIC);
     //Check stored recordIds
     findAll = depublishRecordIdDao
         .getDepublishRecordIds(datasetId, 0, DepublishRecordIdSortField.DEPUBLICATION_STATE,
@@ -267,7 +268,7 @@ void markRecordIdsWithDepublicationStatus_specified_recordIds_set_depublished_an
             .getDepublicationStatus() && null == depublishRecordIdView.getDepublicationDate()));
     //Set to DEPUBLISHED
     depublishRecordIdDao.markRecordIdsWithDepublicationStatus(datasetId, recordIdsToUpdate,
-        DepublicationStatus.DEPUBLISHED, date);
+        DepublicationStatus.DEPUBLISHED, date, DepublicationReason.GENERIC);
     //Check stored recordIds
     findAll = depublishRecordIdDao
         .getDepublishRecordIds(datasetId, 0, DepublishRecordIdSortField.DEPUBLICATION_STATE,
@@ -278,7 +279,7 @@ void markRecordIdsWithDepublicationStatus_specified_recordIds_set_depublished_an
             .equals(Date.from(depublishRecordIdView.getDepublicationDate()))).count());
     //Set to PENDING_DEPUBLICATION
     depublishRecordIdDao.markRecordIdsWithDepublicationStatus(datasetId, recordIdsToUpdate,
-        DepublicationStatus.PENDING_DEPUBLICATION, date);
+        DepublicationStatus.PENDING_DEPUBLICATION, date, DepublicationReason.GENERIC);
     //Check stored recordIds
     findAll = depublishRecordIdDao
         .getDepublishRecordIds(datasetId, 0, DepublishRecordIdSortField.DEPUBLICATION_STATE,
@@ -301,7 +302,7 @@ void markRecordIdsWithDepublicationStatus_depublish_non_already_existing_recordI
     depublishRecordIdDao.createRecordIdsToBeDepublished(datasetId, recordIdsToCreate);
     //Set to DEPUBLISHED
     depublishRecordIdDao.markRecordIdsWithDepublicationStatus(datasetId, recordIdsToUpdate,
-        DepublicationStatus.DEPUBLISHED, date);
+        DepublicationStatus.DEPUBLISHED, date, DepublicationReason.GENERIC);
 
     //Check stored recordIds
     List<DepublishRecordIdView> findAll = depublishRecordIdDao