Merge pull request #352 from UNC-Libraries/prevent-duplicate-jobs

Prevent duplicate deposit jobs from enqueuing when resuming

Author: daines

Diff for: deposit/src/main/java/edu/unc/lib/deposit/work/DepositSupervisor.java

@@ -3,6 +3,8 @@
 import java.io.File;
 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;
@@ -15,6 +17,8 @@
 
 import net.greghaines.jesque.Job;
 import net.greghaines.jesque.client.Client;
+import net.greghaines.jesque.meta.QueueInfo;
+import net.greghaines.jesque.meta.dao.QueueInfoDAO;
 import net.greghaines.jesque.worker.Worker;
 import net.greghaines.jesque.worker.WorkerEvent;
 import net.greghaines.jesque.worker.WorkerListener;
@@ -44,6 +48,7 @@
 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 +77,9 @@ public class DepositSupervisor implements WorkerListener {
 	@Autowired
 	private WorkerPool cdrMetsDepositWorkerPool;
 
+	@Autowired
+	private QueueInfoDAO queueDAO;
+	
 	@Autowired
 	private DepositEmailHandler depositEmailHandler;
 
@@ -137,6 +145,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 +236,37 @@ 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.DEPOSIT_PREPARE_QUEUE, depositMap);
+		addQueuedDeposits(RedisWorkerConstants.DEPOSIT_DELAYED_QUEUE, depositMap);
+		addQueuedDeposits(RedisWorkerConstants.DEPOSIT_CDRMETS_QUEUE, depositMap);
+		return depositMap;
+	}
+	
+	private void addQueuedDeposits(String queueName, Map<String, Set<String>> depositMap) {
+		QueueInfo info = queueDAO.getQueueInfo(queueName, 0, 0);
+		
+		for (Job job : info.getJobs()) {
+			String depositId = (String) job.getArgs()[1];
+			
+			Set<String> jobs = depositMap.get(depositId);
+			if (jobs == null) {
+				jobs = new HashSet<>();
+				depositMap.put(depositId, jobs);
+			}
+			jobs.add(job.getClassName());
+		}
 	}
 
 	/**
 	 * 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 +279,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 +299,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 +663,20 @@ 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);
+			
+			// since we already checked for queued jobs at startup, only check when resuming from a paused state
+			boolean enqueueNext = true;
+			if (DepositState.paused.name().equals(status.get(DepositField.state.name()))) {
+				Map<String, Set<String>> depositSet = getQueuedDepositsWithJobs();
+				enqueueNext = !depositSet.containsKey(uuid);
+			}
 
-			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);
+			} else {
+				LOG.info("Resuming {} from paused state.  A job is already queued so no new jobs will be enqueued", uuid);
+			}
 
 			depositStatusFactory.setState(uuid, DepositState.queued);
 		} catch (DepositFailedException e) {

Diff for: deposit/src/main/resources/service-context.xml

@@ -67,6 +67,11 @@
 		<constructor-arg ref="jedisPool" />
 	</bean>
 
+	<bean id="queueDAO" class="net.greghaines.jesque.meta.dao.impl.QueueInfoDAORedisImpl">
+		<constructor-arg ref="jesqueConfig" />
+		<constructor-arg ref="jedisPool" />
+	</bean>
+	
 	<bean id="dataSet" class="com.hp.hpl.jena.tdb.TDBFactory" factory-method="createDataset" destroy-method="close">
 		<constructor-arg value="${deposits.dir}/jena-tdb-dataset"/>
 	</bean>

Diff for: deposit/src/test/resources/service-context.xml

@@ -73,6 +73,11 @@
 		<constructor-arg ref="jedisPool" />
 	</bean>
 
+	<bean id="queueDAO" class="net.greghaines.jesque.meta.dao.impl.QueueInfoDAORedisImpl">
+		<constructor-arg ref="jesqueConfig" />
+		<constructor-arg ref="jedisPool" />
+	</bean>
+	
 	<bean id="dataSet" class="com.hp.hpl.jena.tdb.TDBFactory" factory-method="createDataset" destroy-method="close">
 		<constructor-arg value="#{depositsDirectory+'/jena-tdb-dataset'}"/>
 	</bean>

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

@@ -12,6 +12,11 @@ public class RedisWorkerConstants {
 	public static final String BULK_RESUME_PREFIX = "bulk-resume:";
 	public static final String BULK_UPDATE_QUEUE = "bulk-md-update";
 
+	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,