Verifying that jobs are not already queued for a deposit before resuming or re-enqueuing during service startup to prevent duplicate jobs being created

Author: bbpennel

deposit/src/main/java/edu/unc/lib/deposit/work/DepositSupervisor.java

@@ -1,8 +1,11 @@
 package edu.unc.lib.deposit.work;
 
 import java.io.File;
+import java.io.IOException;
 import java.text.MessageFormat;
 import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -20,10 +23,14 @@
 import net.greghaines.jesque.worker.WorkerListener;
 import net.greghaines.jesque.worker.WorkerPool;
 
+import org.codehaus.jackson.JsonNode;
+import org.codehaus.jackson.map.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
 import edu.unc.lib.deposit.CleanupDepositJob;
 import edu.unc.lib.deposit.PrepareResubmitJob;
 import edu.unc.lib.deposit.fcrepo3.IngestDeposit;
@@ -36,14 +43,15 @@
 import edu.unc.lib.deposit.normalize.UnpackDepositJob;
 import edu.unc.lib.deposit.normalize.VocabularyEnforcementJob;
 import edu.unc.lib.deposit.validate.PackageIntegrityCheckJob;
+import edu.unc.lib.deposit.validate.ValidateFileAvailabilityJob;
 import edu.unc.lib.deposit.validate.ValidateMODS;
 import edu.unc.lib.deposit.validate.VirusScanJob;
-import edu.unc.lib.deposit.validate.ValidateFileAvailabilityJob;
 import edu.unc.lib.dl.fedora.FedoraTimeoutException;
 import edu.unc.lib.dl.util.DepositConstants;
 import edu.unc.lib.dl.util.DepositStatusFactory;
 import edu.unc.lib.dl.util.JobStatusFactory;
 import edu.unc.lib.dl.util.PackagingType;
+import edu.unc.lib.dl.util.RedisWorkerConstants;
 import edu.unc.lib.dl.util.RedisWorkerConstants.DepositAction;
 import edu.unc.lib.dl.util.RedisWorkerConstants.DepositField;
 import edu.unc.lib.dl.util.RedisWorkerConstants.DepositState;
@@ -72,6 +80,9 @@ public class DepositSupervisor implements WorkerListener {
 	@Autowired
 	private WorkerPool cdrMetsDepositWorkerPool;
 
+	@Autowired
+	private JedisPool jedisPool;
+	
 	@Autowired
 	private DepositEmailHandler depositEmailHandler;
 
@@ -137,6 +148,9 @@ public void init() {
 	}
 
 	public void start() {
+		// Repopulate the queue
+		requeueAll();
+		
 		LOG.info("Starting deposit checks and worker pool");
 		if (timer != null)
 			return;
@@ -225,15 +239,53 @@ public void run() {
 			LOG.info("Starting deposit workers");
 			cdrMetsDepositWorkerPool.run();
 		}
-
-		// Repopulate the queue
-		requeueAll();
+	}
+	
+	private Map<String, Set<String>> getQueuedDepositsWithJobs() {
+		Map<String, Set<String>> depositMap = new HashMap<>();
+		addQueuedDeposits(
+				RedisWorkerConstants.RESQUE_QUEUE_PREFIX + RedisWorkerConstants.DEPOSIT_PREPARE_QUEUE, depositMap);
+		addQueuedDeposits(
+				RedisWorkerConstants.RESQUE_QUEUE_PREFIX + RedisWorkerConstants.DEPOSIT_DELAYED_QUEUE, depositMap);
+		addQueuedDeposits(
+				RedisWorkerConstants.RESQUE_QUEUE_PREFIX + RedisWorkerConstants.DEPOSIT_CDRMETS_QUEUE, depositMap);
+		return depositMap;
+	}
+	
+	private void addQueuedDeposits(String queueName, Map<String, Set<String>> depositMap) {
+		Jedis jedis = jedisPool.getResource();
+		Set<String> queue;
+		try {
+			queue = jedis.zrange(queueName, 0, -1);
+		} catch (Exception e) {
+			// Resque seems to sometimes switch the type of the queue
+			LOG.warn("Redis did not return a zset for {}, trying to retrieve as a list", queueName, e);
+			queue = new HashSet<>(jedis.lrange(queueName, 0, -1));
+		}
+		
+		ObjectMapper mapper = new ObjectMapper();
+		for (String entry : queue) {
+			try {
+				JsonNode node = mapper.readTree(entry);
+				String depositId = node.get("args").get(1).asText();
+				Set<String> jobs = depositMap.get(depositId);
+				if (jobs == null) {
+					jobs = new HashSet<>();
+					depositMap.put(depositId, jobs);
+				}
+				jobs.add(node.get("class").asText());
+			} catch (IOException e) {
+				LOG.error("Failed to parse deposit job from resque", e);
+			}
+		}
 	}
 
 	/**
 	 * Add jobs previously running or queued back to the queue
 	 */
 	private void requeueAll() {
+		
+		Map<String, Set<String>> depositSet = getQueuedDepositsWithJobs();
 		Set<Map<String, String>> depositStatuses = depositStatusFactory.getAll();
 
 		LOG.info("Repopulating the deposit queue, {} items in backlog", depositStatuses.size());
@@ -246,7 +298,16 @@ private void requeueAll() {
 				// Job may have been locked to a particular supervisor depend on when it was interrupted
 				depositStatusFactory.removeSupervisorLock(uuid);
 				// Inform supervisor to resume this deposit from where it left off
-				depositStatusFactory.setActionRequest(uuid, DepositAction.resume);
+				if (depositSet.containsKey(uuid)) {
+					// If the job is queued but the job it is waiting on is a cleanup, then it is finished
+					if (depositSet.get(uuid).contains(CleanupDepositJob.class.getName())) {
+						depositStatusFactory.setState(uuid, DepositState.finished);
+					} else {
+						LOG.debug("Skipping resumption of deposit {} because it already is in the queue", uuid);
+					}
+				} else {
+					depositStatusFactory.setActionRequest(uuid, DepositAction.resume);
+				}
 			}
 		}
 
@@ -257,7 +318,21 @@ private void requeueAll() {
 
 				depositStatusFactory.removeSupervisorLock(uuid);
 				// Re-register as a new deposit
-				depositStatusFactory.setActionRequest(uuid, DepositAction.register);
+				if (depositSet.containsKey(uuid)) {
+					if (depositSet.get(uuid).contains(CleanupDepositJob.class.getName())) {
+						depositStatusFactory.setState(uuid, DepositState.finished);
+					} else {
+						LOG.debug("Skipping resumption of queued deposit {} because it already is in the queue", uuid);
+					}
+				} else {
+					List<String> successfulJobs = jobStatusFactory.getSuccessfulJobNames(uuid);
+					if (successfulJobs != null && successfulJobs.size() > 0) {
+						// Queued but had already performed some jobs, so this is a resumption rather than new deposit
+						depositStatusFactory.setActionRequest(uuid, DepositAction.resume);
+					} else {
+						depositStatusFactory.setActionRequest(uuid, DepositAction.register);
+					}
+				}
 			}
 		}
 	}
@@ -607,9 +682,18 @@ private void resumeDeposit(String uuid, Map<String, String> status, long delay)
 			// Clear out the previous failed job if there was one
 			jobStatusFactory.clearStale(uuid);
 			depositStatusFactory.deleteField(uuid, DepositField.errorMessage);
+			
+			boolean enqueueNext = true;
+			if (DepositState.paused.name().equals(status.get(DepositField.state.name()))) {
+				Map<String, Set<String>> depositSet = getQueuedDepositsWithJobs();
+				enqueueNext = !depositSet.containsKey(uuid);
+				LOG.info("Resuming from paused state.  {} will enqueue a new job {}", uuid, enqueueNext);
+			}
 
-			List<String> successfulJobs = jobStatusFactory.getSuccessfulJobNames(uuid);
-			queueNextJob(null, uuid, status, successfulJobs, delay);
+			if (enqueueNext) {
+				List<String> successfulJobs = jobStatusFactory.getSuccessfulJobNames(uuid);
+				queueNextJob(null, uuid, status, successfulJobs, delay);
+			}
 
 			depositStatusFactory.setState(uuid, DepositState.queued);
 		} catch (DepositFailedException e) {

metadata/src/main/java/edu/unc/lib/dl/util/RedisWorkerConstants.java

@@ -9,6 +9,11 @@ public class RedisWorkerConstants {
 	public static final String DEPOSIT_TO_JOBS_PREFIX = "deposit-to-jobs:";
 	public static final String JOB_STATUS_PREFIX = "job-status:";
 
+	public static final String DEPOSIT_PREPARE_QUEUE = "PREPARE";
+	public static final String DEPOSIT_DELAYED_QUEUE = "DELAYED_PREPARE";
+	public static final String DEPOSIT_CDRMETS_QUEUE = "CDRMETSCONVERT";
+	public static final String RESQUE_QUEUE_PREFIX = "resque:queue:";
+
 	public static enum DepositField {
 		uuid, state, actionRequest, contactName, depositorName, intSenderIdentifier, intSenderDescription,
 		fileName, resubmitDirName, resubmitFileName, isResubmit, depositMethod, containerId, payLoadOctets,