Author: kwright
Date: Tue Feb 11 11:15:57 2014
New Revision: 1567054
URL: http://svn.apache.org/r1567054
Log:
Fix for CONNECTORS-888.
Modified:
manifoldcf/trunk/CHANGES.txt
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
Modified: manifoldcf/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/CHANGES.txt?rev=1567054&r1=1567053&r2=1567054&view=diff
==============================================================================
--- manifoldcf/trunk/CHANGES.txt (original)
+++ manifoldcf/trunk/CHANGES.txt Tue Feb 11 11:15:57 2014
@@ -3,6 +3,10 @@ $Id$
======================= 1.6-dev =====================
+CONNECTORS-888: JobResetThread had issues when running on
+multiple agents processes.
+(Florian Schmedding, Karl Wright)
+
CONNECTORS-887: Upgrade issues with MySQL.
(Florian Schmedding, Karl Wright)
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java?rev=1567054&r1=1567053&r2=1567054&view=diff
==============================================================================
---
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
(original)
+++
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
Tue Feb 11 11:15:57 2014
@@ -36,6 +36,7 @@ public class JobManager implements IJobM
protected static final String deleteStufferLock = "_DELETESTUFFER_";
protected static final String expireStufferLock = "_EXPIRESTUFFER_";
protected static final String cleanStufferLock = "_CLEANSTUFFER_";
+ protected static final String jobStopLock = "_JOBSTOP_";
protected static final String hopLock = "_HOPLOCK_";
// Member variables
@@ -7573,73 +7574,81 @@ public class JobManager implements IJobM
public void finishJobStops(long timestamp, ArrayList modifiedJobs)
throws ManifoldCFException
{
- // The query I used to emit was:
- // SELECT jobid FROM jobs t0 WHERE t0.status='X' AND NOT EXISTS(SELECT 'x'
FROM jobqueue t1 WHERE
- // t0.id=t1.jobid AND t1.status IN ('A','F'))
- // Now the query is broken up so that Postgresql behaves more efficiently.
-
- // Do the first query, getting the candidate jobs to be considered
- StringBuilder sb = new StringBuilder("SELECT ");
- ArrayList list = new ArrayList();
-
- sb.append(jobs.idField)
- .append(" FROM ").append(jobs.getTableName()).append(" WHERE ")
- .append(database.buildConjunctionClause(list,new ClauseDescription[]{
- new MultiClause(jobs.statusField,new Object[]{
- jobs.statusToString(jobs.STATUS_ABORTING),
- jobs.statusToString(jobs.STATUS_ABORTINGFORRESTART),
- jobs.statusToString(jobs.STATUS_ABORTINGFORRESTARTMINIMAL),
- jobs.statusToString(jobs.STATUS_ABORTINGSHUTTINGDOWN),
- jobs.statusToString(jobs.STATUS_PAUSING),
- jobs.statusToString(jobs.STATUS_PAUSINGSEEDING),
- jobs.statusToString(jobs.STATUS_ACTIVEWAITING),
- jobs.statusToString(jobs.STATUS_ACTIVEWAITINGSEEDING),
- jobs.statusToString(jobs.STATUS_PAUSINGWAITING),
- jobs.statusToString(jobs.STATUS_PAUSINGWAITINGSEEDING)
- })}));
-
- IResultSet set = database.performQuery(sb.toString(),list,null,null);
-
- int i = 0;
- while (i < set.getRowCount())
+ lockManager.enterWriteLock(jobStopLock);
+ try
{
- IResultRow row = set.getRow(i++);
- Long jobID = (Long)row.getValue(jobs.idField);
-
- sb = new StringBuilder("SELECT ");
- list.clear();
+ // The query I used to emit was:
+ // SELECT jobid FROM jobs t0 WHERE t0.status='X' AND NOT EXISTS(SELECT
'x' FROM jobqueue t1 WHERE
+ // t0.id=t1.jobid AND t1.status IN ('A','F'))
+ // Now the query is broken up so that Postgresql behaves more
efficiently.
- sb.append(jobQueue.idField).append(" FROM
").append(jobQueue.getTableName()).append(" WHERE ")
+ // Do the first query, getting the candidate jobs to be considered
+ StringBuilder sb = new StringBuilder("SELECT ");
+ ArrayList list = new ArrayList();
+
+ sb.append(jobs.idField)
+ .append(" FROM ").append(jobs.getTableName()).append(" WHERE ")
.append(database.buildConjunctionClause(list,new ClauseDescription[]{
- new UnitaryClause(jobQueue.jobIDField,jobID),
- new MultiClause(jobQueue.statusField,new Object[]{
- jobQueue.statusToString(jobQueue.STATUS_ACTIVE),
- jobQueue.statusToString(jobQueue.STATUS_ACTIVEPURGATORY),
- jobQueue.statusToString(jobQueue.STATUS_ACTIVENEEDRESCAN),
- jobQueue.statusToString(jobQueue.STATUS_ACTIVENEEDRESCANPURGATORY),
- jobQueue.statusToString(jobQueue.STATUS_BEINGCLEANED)})}))
- .append(" ").append(database.constructOffsetLimitClause(0,1));
+ new MultiClause(jobs.statusField,new Object[]{
+ jobs.statusToString(jobs.STATUS_ABORTING),
+ jobs.statusToString(jobs.STATUS_ABORTINGFORRESTART),
+ jobs.statusToString(jobs.STATUS_ABORTINGFORRESTARTMINIMAL),
+ jobs.statusToString(jobs.STATUS_ABORTINGSHUTTINGDOWN),
+ jobs.statusToString(jobs.STATUS_PAUSING),
+ jobs.statusToString(jobs.STATUS_PAUSINGSEEDING),
+ jobs.statusToString(jobs.STATUS_ACTIVEWAITING),
+ jobs.statusToString(jobs.STATUS_ACTIVEWAITINGSEEDING),
+ jobs.statusToString(jobs.STATUS_PAUSINGWAITING),
+ jobs.statusToString(jobs.STATUS_PAUSINGWAITINGSEEDING)
+ })}));
+
+ IResultSet set = database.performQuery(sb.toString(),list,null,null);
- IResultSet confirmSet =
database.performQuery(sb.toString(),list,null,null,1,null);
+ int i = 0;
+ while (i < set.getRowCount())
+ {
+ IResultRow row = set.getRow(i++);
+ Long jobID = (Long)row.getValue(jobs.idField);
- if (confirmSet.getRowCount() > 0)
- continue;
+ sb = new StringBuilder("SELECT ");
+ list.clear();
- // All the job's documents need to have their docpriority set to null,
to clear dead wood out of the docpriority index.
- // See CONNECTORS-290.
- // We do this BEFORE updating the job state.
- jobQueue.clearDocPriorities(jobID);
-
- IJobDescription jobDesc = jobs.load(jobID,true);
- modifiedJobs.add(jobDesc);
+ sb.append(jobQueue.idField).append(" FROM
").append(jobQueue.getTableName()).append(" WHERE ")
+ .append(database.buildConjunctionClause(list,new ClauseDescription[]{
+ new UnitaryClause(jobQueue.jobIDField,jobID),
+ new MultiClause(jobQueue.statusField,new Object[]{
+ jobQueue.statusToString(jobQueue.STATUS_ACTIVE),
+ jobQueue.statusToString(jobQueue.STATUS_ACTIVEPURGATORY),
+ jobQueue.statusToString(jobQueue.STATUS_ACTIVENEEDRESCAN),
+
jobQueue.statusToString(jobQueue.STATUS_ACTIVENEEDRESCANPURGATORY),
+ jobQueue.statusToString(jobQueue.STATUS_BEINGCLEANED)})}))
+ .append(" ").append(database.constructOffsetLimitClause(0,1));
+
+ IResultSet confirmSet =
database.performQuery(sb.toString(),list,null,null,1,null);
+
+ if (confirmSet.getRowCount() > 0)
+ continue;
+
+ // All the job's documents need to have their docpriority set to null,
to clear dead wood out of the docpriority index.
+ // See CONNECTORS-290.
+ // We do this BEFORE updating the job state.
+ jobQueue.clearDocPriorities(jobID);
+
+ IJobDescription jobDesc = jobs.load(jobID,true);
+ modifiedJobs.add(jobDesc);
- jobs.finishStopJob(jobID,timestamp);
-
- if (Logging.jobs.isDebugEnabled())
- {
- Logging.jobs.debug("Stopped job "+jobID);
+ jobs.finishStopJob(jobID,timestamp);
+
+ if (Logging.jobs.isDebugEnabled())
+ {
+ Logging.jobs.debug("Stopped job "+jobID);
+ }
}
}
+ finally
+ {
+ lockManager.leaveWriteLock(jobStopLock);
+ }
}
/** Reset eligible jobs either back to the "inactive" state, or make them
active again. The
