Author: kwright
Date: Mon Nov 25 00:29:40 2013
New Revision: 1545117
URL: http://svn.apache.org/r1545117
Log:
Finish processID propagation
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java?rev=1545117&r1=1545116&r2=1545117&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
Mon Nov 25 00:29:40 2013
@@ -772,11 +772,12 @@ public interface IJobManager
throws ManifoldCFException;
/** Get the list of jobs that are ready for seeding.
+ *@param processID is the current process ID.
*@param currentTime is the current time in milliseconds since epoch.
*@return jobs that are active and are running in adaptive mode. These will
be seeded
* based on what the connector says should be added to the queue.
*/
- public JobSeedingRecord[] getJobsReadyForSeeding(long currentTime)
+ public JobSeedingRecord[] getJobsReadyForSeeding(String processID, long
currentTime)
throws ManifoldCFException;
/** Reset a seeding job back to "active" state.
@@ -786,21 +787,24 @@ public interface IJobManager
throws ManifoldCFException;
/** Get the list of jobs that are ready for deletion.
+ *@param processID is the current process ID.
*@return jobs that were in the "readyfordelete" state.
*/
- public JobDeleteRecord[] getJobsReadyForDelete()
+ public JobDeleteRecord[] getJobsReadyForDelete(String processID)
throws ManifoldCFException;
/** Get the list of jobs that are ready for startup.
+ *@param processID is the current process ID.
*@return jobs that were in the "readyforstartup" state. These will be
marked as being in the "starting up" state.
*/
- public JobStartRecord[] getJobsReadyForStartup()
+ public JobStartRecord[] getJobsReadyForStartup(String processID)
throws ManifoldCFException;
/** Find the list of jobs that need to have their connectors notified of job
completion.
+ *@param processID is the current process ID.
*@return the ID's of jobs that need their output connectors notified in
order to become inactive.
*/
- public JobNotifyRecord[] getJobsReadyForInactivity()
+ public JobNotifyRecord[] getJobsReadyForInactivity(String processID)
throws ManifoldCFException;
/** Inactivate a job, from the notification state.
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java?rev=1545117&r1=1545116&r2=1545117&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
Mon Nov 25 00:29:40 2013
@@ -534,7 +534,7 @@ public class JobManager implements IJobM
throw new ManifoldCFException("Job "+id+" is active; you must shut it
down before deleting it");
if (status != jobs.STATUS_INACTIVE)
throw new ManifoldCFException("Job "+id+" is busy; you must wait
and/or shut it down before deleting it");
- jobs.writeStatus(id,jobs.STATUS_READYFORDELETE);
+ jobs.writePermanentStatus(id,jobs.STATUS_READYFORDELETE);
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Job "+id+" marked for deletion");
}
@@ -798,7 +798,7 @@ public class JobManager implements IJobM
database.beginTransaction();
try
{
- jobs.resetSeedingWorkerStatus();
+ jobs.resetSeedingWorkerStatus(processID);
TrackerClass.notePrecommit();
database.performCommit();
TrackerClass.noteCommit();
@@ -945,7 +945,7 @@ public class JobManager implements IJobM
database.beginTransaction();
try
{
- jobs.resetDeleteStartupWorkerStatus();
+ jobs.resetDeleteStartupWorkerStatus(processID);
TrackerClass.notePrecommit();
database.performCommit();
TrackerClass.noteCommit();
@@ -993,7 +993,7 @@ public class JobManager implements IJobM
database.beginTransaction();
try
{
- jobs.resetNotificationWorkerStatus();
+ jobs.resetNotificationWorkerStatus(processID);
TrackerClass.notePrecommit();
database.performCommit();
TrackerClass.noteCommit();
@@ -1041,7 +1041,7 @@ public class JobManager implements IJobM
database.beginTransaction();
try
{
- jobs.resetStartupWorkerStatus();
+ jobs.resetStartupWorkerStatus(processID);
TrackerClass.notePrecommit();
database.performCommit();
TrackerClass.noteCommit();
@@ -6249,10 +6249,12 @@ public class JobManager implements IJobM
}
/** Get the list of jobs that are ready for seeding.
+ *@param processID is the current process ID.
*@return jobs that are active and are running in adaptive mode. These will
be seeded
* based on what the connector says should be added to the queue.
*/
- public JobSeedingRecord[] getJobsReadyForSeeding(long currentTime)
+ @Override
+ public JobSeedingRecord[] getJobsReadyForSeeding(String processID, long
currentTime)
throws ManifoldCFException
{
while (true)
@@ -6300,7 +6302,7 @@ public class JobManager implements IJobM
// Mark status of job as "active/seeding". Special status is needed
so that abort
// will not complete until seeding is completed.
- jobs.writeStatus(jobID,jobs.STATUS_ACTIVESEEDING,reseedTime);
+
jobs.writeTransientStatus(jobID,jobs.STATUS_ACTIVESEEDING,reseedTime,processID);
if (Logging.jobs.isDebugEnabled())
{
Logging.jobs.debug("Marked job "+jobID+" for seeding");
@@ -6338,9 +6340,11 @@ public class JobManager implements IJobM
}
/** Get the list of jobs that are ready for deletion.
+ *@param processID is the current process ID.
*@return jobs that were in the "readyfordelete" state.
*/
- public JobDeleteRecord[] getJobsReadyForDelete()
+ @Override
+ public JobDeleteRecord[] getJobsReadyForDelete(String processID)
throws ManifoldCFException
{
while (true)
@@ -6368,7 +6372,7 @@ public class JobManager implements IJobM
Long jobID = (Long)row.getValue(jobs.idField);
// Mark status of job as "starting delete"
- jobs.writeStatus(jobID,jobs.STATUS_DELETESTARTINGUP);
+
jobs.writeTransientStatus(jobID,jobs.STATUS_DELETESTARTINGUP,processID);
if (Logging.jobs.isDebugEnabled())
{
Logging.jobs.debug("Marked job "+jobID+" for delete startup");
@@ -6406,9 +6410,11 @@ public class JobManager implements IJobM
}
/** Get the list of jobs that are ready for startup.
+ *@param processID is the current process ID.
*@return jobs that were in the "readyforstartup" state. These will be
marked as being in the "starting up" state.
*/
- public JobStartRecord[] getJobsReadyForStartup()
+ @Override
+ public JobStartRecord[] getJobsReadyForStartup(String processID)
throws ManifoldCFException
{
while (true)
@@ -6449,7 +6455,7 @@ public class JobManager implements IJobM
synchTime = x.longValue();
// Mark status of job as "starting"
-
jobs.writeStatus(jobID,requestMinimum?jobs.STATUS_STARTINGUPMINIMAL:jobs.STATUS_STARTINGUP);
+
jobs.writeTransientStatus(jobID,requestMinimum?jobs.STATUS_STARTINGUPMINIMAL:jobs.STATUS_STARTINGUP,processID);
if (Logging.jobs.isDebugEnabled())
{
Logging.jobs.debug("Marked job "+jobID+" for startup");
@@ -6586,7 +6592,7 @@ public class JobManager implements IJobM
Logging.jobs.debug("Setting job "+jobID+" back to 'ReadyForDelete'
state");
// Set the state of the job back to "ReadyForStartup"
- jobs.writeStatus(jobID,jobs.STATUS_READYFORDELETE);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_READYFORDELETE);
break;
default:
throw new ManifoldCFException("Unexpected job status:
"+Integer.toString(status));
@@ -6654,7 +6660,7 @@ public class JobManager implements IJobM
Logging.jobs.debug("Setting job "+jobID+" back to 'ReadyForNotify'
state");
// Set the state of the job back to "ReadyForNotify"
- jobs.writeStatus(jobID,jobs.STATUS_READYFORNOTIFY);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_READYFORNOTIFY);
break;
default:
throw new ManifoldCFException("Unexpected job status:
"+Integer.toString(status));
@@ -6690,6 +6696,7 @@ public class JobManager implements IJobM
/** Reset a starting job back to "ready for startup" state.
*@param jobID is the job id.
*/
+ @Override
public void resetStartupJob(Long jobID)
throws ManifoldCFException
{
@@ -6721,30 +6728,30 @@ public class JobManager implements IJobM
Logging.jobs.debug("Setting job "+jobID+" back to
'ReadyForStartup' state");
// Set the state of the job back to "ReadyForStartup"
- jobs.writeStatus(jobID,jobs.STATUS_READYFORSTARTUP);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_READYFORSTARTUP);
break;
case Jobs.STATUS_STARTINGUPMINIMAL:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to
'ReadyForStartupMinimal' state");
// Set the state of the job back to "ReadyForStartupMinimal"
- jobs.writeStatus(jobID,jobs.STATUS_READYFORSTARTUPMINIMAL);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_READYFORSTARTUPMINIMAL);
break;
case Jobs.STATUS_ABORTINGSTARTINGUP:
case Jobs.STATUS_ABORTINGSTARTINGUPMINIMAL:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" to 'Aborting' state");
- jobs.writeStatus(jobID,jobs.STATUS_ABORTING);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ABORTING);
break;
case Jobs.STATUS_ABORTINGSTARTINGUPFORRESTART:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" to 'AbortingForRestart'
state");
- jobs.writeStatus(jobID,jobs.STATUS_ABORTINGFORRESTART);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ABORTINGFORRESTART);
break;
case Jobs.STATUS_ABORTINGSTARTINGUPFORRESTARTMINIMAL:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" to
'AbortingForRestartMinimal' state");
- jobs.writeStatus(jobID,jobs.STATUS_ABORTINGFORRESTARTMINIMAL);
+
jobs.writePermanentStatus(jobID,jobs.STATUS_ABORTINGFORRESTARTMINIMAL);
break;
case Jobs.STATUS_READYFORSTARTUP:
@@ -6818,56 +6825,56 @@ public class JobManager implements IJobM
Logging.jobs.debug("Setting job "+jobID+" back to
'Active_Uninstalled' state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ACTIVE_UNINSTALLED);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ACTIVE_UNINSTALLED);
break;
case Jobs.STATUS_ACTIVESEEDING_NOOUTPUT:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to
'Active_NoOutput' state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ACTIVE_NOOUTPUT);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ACTIVE_NOOUTPUT);
break;
case Jobs.STATUS_ACTIVESEEDING_NEITHER:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to 'Active_Neither'
state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ACTIVE_NEITHER);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ACTIVE_NEITHER);
break;
case Jobs.STATUS_ACTIVESEEDING:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to 'Active' state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ACTIVE);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ACTIVE);
break;
case Jobs.STATUS_ACTIVEWAITSEEDING:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to 'ActiveWait'
state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ACTIVEWAIT);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ACTIVEWAIT);
break;
case Jobs.STATUS_PAUSEDSEEDING:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to 'Paused' state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_PAUSED);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_PAUSED);
break;
case Jobs.STATUS_PAUSEDWAITSEEDING:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to 'PausedWait'
state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_PAUSEDWAIT);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_PAUSEDWAIT);
break;
case Jobs.STATUS_ABORTINGSEEDING:
if (Logging.jobs.isDebugEnabled())
Logging.jobs.debug("Setting job "+jobID+" back to 'Aborting'
state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ABORTING);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ABORTING);
break;
case Jobs.STATUS_ABORTINGFORRESTARTSEEDING:
@@ -6875,7 +6882,7 @@ public class JobManager implements IJobM
Logging.jobs.debug("Setting job "+jobID+" back to
'AbortingForRestart' state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ABORTINGFORRESTART);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_ABORTINGFORRESTART);
break;
case Jobs.STATUS_ABORTINGFORRESTARTSEEDINGMINIMAL:
@@ -6883,7 +6890,7 @@ public class JobManager implements IJobM
Logging.jobs.debug("Setting job "+jobID+" back to
'AbortingForRestartMinimal' state");
// Set the state of the job back to "Active"
- jobs.writeStatus(jobID,jobs.STATUS_ABORTINGFORRESTARTMINIMAL);
+
jobs.writePermanentStatus(jobID,jobs.STATUS_ABORTINGFORRESTARTMINIMAL);
break;
case Jobs.STATUS_ABORTING:
@@ -7102,7 +7109,7 @@ public class JobManager implements IJobM
continue;
// Mark status of job as "finishing"
- jobs.writeStatus(jobID,jobs.STATUS_SHUTTINGDOWN);
+ jobs.writePermanentStatus(jobID,jobs.STATUS_SHUTTINGDOWN);
if (Logging.jobs.isDebugEnabled())
{
Logging.jobs.debug("Marked job "+jobID+" for shutdown");
@@ -7138,9 +7145,11 @@ public class JobManager implements IJobM
}
/** Find the list of jobs that need to have their connectors notified of job
completion.
+ *@param processID is the process ID.
*@return the ID's of jobs that need their output connectors notified in
order to become inactive.
*/
- public JobNotifyRecord[] getJobsReadyForInactivity()
+ @Override
+ public JobNotifyRecord[] getJobsReadyForInactivity(String processID)
throws ManifoldCFException
{
while (true)
@@ -7167,7 +7176,7 @@ public class JobManager implements IJobM
IResultRow row = set.getRow(i);
Long jobID = (Long)row.getValue(jobs.idField);
// Mark status of job as "starting delete"
- jobs.writeStatus(jobID,jobs.STATUS_NOTIFYINGOFCOMPLETION);
+
jobs.writeTransientStatus(jobID,jobs.STATUS_NOTIFYINGOFCOMPLETION,processID);
if (Logging.jobs.isDebugEnabled())
{
Logging.jobs.debug("Found job "+jobID+" in need of notification");
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java?rev=1545117&r1=1545116&r2=1545117&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
Mon Nov 25 00:29:40 2013
@@ -257,6 +257,7 @@ public class Jobs extends org.apache.man
hopmodeMap.put("V",new Integer(HOPCOUNT_NEVERDELETE));
}
+ /*
protected static Set<Integer> transientStates;
static
{
@@ -284,6 +285,7 @@ public class Jobs extends org.apache.man
transientStates.add(new Integer(STATUS_ACTIVESEEDING_NOOUTPUT));
transientStates.add(new Integer(STATUS_ACTIVESEEDING_NEITHER));
}
+ */
// Local variables
protected ICacheManager cacheManager;
@@ -1441,7 +1443,7 @@ public class Jobs extends org.apache.man
/** Reset delete startup worker thread status.
*/
- public void resetDeleteStartupWorkerStatus()
+ public void resetDeleteStartupWorkerStatus(String processID)
throws ManifoldCFException
{
// This handles everything that the delete startup thread would resolve.
@@ -1450,7 +1452,7 @@ public class Jobs extends org.apache.man
HashMap map = new HashMap();
String query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_DELETESTARTINGUP)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_READYFORDELETE));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -1459,7 +1461,7 @@ public class Jobs extends org.apache.man
/** Reset notification worker thread status.
*/
- public void resetNotificationWorkerStatus()
+ public void resetNotificationWorkerStatus(String processID)
throws ManifoldCFException
{
// This resets everything that the job notification thread would resolve.
@@ -1468,7 +1470,7 @@ public class Jobs extends org.apache.man
HashMap map = new HashMap();
String query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_NOTIFYINGOFCOMPLETION)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_READYFORNOTIFY));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -1477,7 +1479,7 @@ public class Jobs extends org.apache.man
/** Reset startup worker thread status.
*/
- public void resetStartupWorkerStatus()
+ public void resetStartupWorkerStatus(String processID)
throws ManifoldCFException
{
// We have to handle all states that the startup thread would resolve, and
change them to something appropriate.
@@ -1489,7 +1491,7 @@ public class Jobs extends org.apache.man
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_STARTINGUP)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_READYFORSTARTUP));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -1497,7 +1499,7 @@ public class Jobs extends org.apache.man
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_STARTINGUPMINIMAL)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_READYFORSTARTUPMINIMAL));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -1507,7 +1509,7 @@ public class Jobs extends org.apache.man
new MultiClause(statusField,new Object[]{
statusToString(STATUS_ABORTINGSTARTINGUP),
statusToString(STATUS_ABORTINGSTARTINGUPMINIMAL)}),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ABORTING));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -1515,7 +1517,7 @@ public class Jobs extends org.apache.man
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ABORTINGSTARTINGUPFORRESTART)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ABORTINGFORRESTART));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -1523,7 +1525,7 @@ public class Jobs extends org.apache.man
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ABORTINGSTARTINGUPFORRESTARTMINIMAL)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ABORTINGFORRESTARTMINIMAL));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -1532,7 +1534,7 @@ public class Jobs extends org.apache.man
/** Reset as part of restoring seeding worker threads.
*/
- public void resetSeedingWorkerStatus()
+ public void resetSeedingWorkerStatus(String processID)
throws ManifoldCFException
{
StringSet invKey = new StringSet(getJobStatusKey());
@@ -1543,98 +1545,98 @@ public class Jobs extends org.apache.man
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_ACTIVESEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ACTIVE));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_PAUSINGSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_PAUSING));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ACTIVEWAITINGSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ACTIVEWAITING));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_PAUSINGWAITINGSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_PAUSINGWAITING));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_RESUMINGSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_RESUMING));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_ABORTINGSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ABORTING));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ABORTINGFORRESTARTSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ABORTINGFORRESTART));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ABORTINGFORRESTARTSEEDINGMINIMAL)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ABORTINGFORRESTARTMINIMAL));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_PAUSEDSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_PAUSED));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_ACTIVEWAITSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ACTIVEWAIT));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_PAUSEDWAITSEEDING)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_PAUSEDWAIT));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ACTIVESEEDING_UNINSTALLED)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ACTIVE_UNINSTALLED));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ACTIVESEEDING_NOOUTPUT)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ACTIVE_NOOUTPUT));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
list.clear();
query = buildConjunctionClause(list,new ClauseDescription[]{
new
UnitaryClause(statusField,statusToString(STATUS_ACTIVESEEDING_NEITHER)),
- new UnitaryClause(processIDField,ManifoldCF.getProcessID())});
+ new UnitaryClause(processIDField,processID)});
map.put(statusField,statusToString(STATUS_ACTIVE_NEITHER));
map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,invKey);
@@ -2222,7 +2224,29 @@ public class Jobs extends org.apache.man
*@param status is the desired status.
*@param reseedTime is the reseed time.
*/
- public void writeStatus(Long jobID, int status, Long reseedTime)
+ public void writeTransientStatus(Long jobID, int status, Long reseedTime,
String processID)
+ throws ManifoldCFException
+ {
+ writeStatus(jobID, status, reseedTime, processID);
+ }
+
+ /** Update a job's status, and its reseed time.
+ *@param jobID is the job id.
+ *@param status is the desired status.
+ *@param reseedTime is the reseed time.
+ */
+ public void writePermanentStatus(Long jobID, int status, Long reseedTime)
+ throws ManifoldCFException
+ {
+ writeStatus(jobID, status, reseedTime, null);
+ }
+
+ /** Update a job's status, and its reseed time.
+ *@param jobID is the job id.
+ *@param status is the desired status.
+ *@param reseedTime is the reseed time.
+ */
+ protected void writeStatus(Long jobID, int status, Long reseedTime, String
processID)
throws ManifoldCFException
{
ArrayList list = new ArrayList();
@@ -2230,7 +2254,7 @@ public class Jobs extends org.apache.man
new UnitaryClause(idField,jobID)});
HashMap map = new HashMap();
map.put(statusField,statusToString(status));
- map.put(processIDField,transientStates.contains(new
Integer(status))?ManifoldCF.getProcessID():null);
+ map.put(processIDField,processID);
map.put(reseedTimeField,reseedTime);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
@@ -2239,7 +2263,27 @@ public class Jobs extends org.apache.man
*@param jobID is the job id.
*@param status is the desired status.
*/
- public void writeStatus(Long jobID, int status)
+ public void writeTransientStatus(Long jobID, int status, String processID)
+ throws ManifoldCFException
+ {
+ writeStatus(jobID, status, processID);
+ }
+
+ /** Update a job's status.
+ *@param jobID is the job id.
+ *@param status is the desired status.
+ */
+ public void writePermanentStatus(Long jobID, int status)
+ throws ManifoldCFException
+ {
+ writeStatus(jobID, status, null);
+ }
+
+ /** Update a job's status.
+ *@param jobID is the job id.
+ *@param status is the desired status.
+ */
+ protected void writeStatus(Long jobID, int status, String processID)
throws ManifoldCFException
{
ArrayList list = new ArrayList();
@@ -2247,7 +2291,7 @@ public class Jobs extends org.apache.man
new UnitaryClause(idField,jobID)});
HashMap map = new HashMap();
map.put(statusField,statusToString(status));
- map.put(processIDField,transientStates.contains(new
Integer(status))?ManifoldCF.getProcessID():null);
+ map.put(processIDField,processID);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java?rev=1545117&r1=1545116&r2=1545117&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java
Mon Nov 25 00:29:40 2013
@@ -70,7 +70,7 @@ public class JobNotificationThread exten
// Before we begin, conditionally reset
resetManager.waitForReset(threadContext);
- JobNotifyRecord[] jobsNeedingNotification =
jobManager.getJobsReadyForInactivity();
+ JobNotifyRecord[] jobsNeedingNotification =
jobManager.getJobsReadyForInactivity(processID);
try
{
HashMap connectionNames = new HashMap();
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java?rev=1545117&r1=1545116&r2=1545117&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
Mon Nov 25 00:29:40 2013
@@ -95,7 +95,7 @@ public class SeedingThread extends Threa
Logging.threads.debug("Seeding thread woke up");
// Grab active, adaptive jobs (and set their state to xxxSEEDING as
a side effect)
- JobSeedingRecord[] seedJobs =
jobManager.getJobsReadyForSeeding(currentTime);
+ JobSeedingRecord[] seedJobs =
jobManager.getJobsReadyForSeeding(processID,currentTime);
// Process these jobs, and do the seeding. The seeding is based on
what came back
// in the job start record for sync time. If there's an
interruption, we just go on
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java?rev=1545117&r1=1545116&r2=1545117&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
Mon Nov 25 00:29:40 2013
@@ -84,7 +84,7 @@ public class StartDeleteThread extends T
// See if there are any starting jobs.
// Note: Since this following call changes the job state, we must be
careful to reset it on any kind of failure.
- JobDeleteRecord[] deleteJobs = jobManager.getJobsReadyForDelete();
+ JobDeleteRecord[] deleteJobs =
jobManager.getJobsReadyForDelete(processID);
try
{
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java?rev=1545117&r1=1545116&r2=1545117&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
Mon Nov 25 00:29:40 2013
@@ -88,7 +88,7 @@ public class StartupThread extends Threa
// See if there are any starting jobs.
// Note: Since this following call changes the job state, we must be
careful to reset it on any kind of failure.
- JobStartRecord[] startupJobs = jobManager.getJobsReadyForStartup();
+ JobStartRecord[] startupJobs =
jobManager.getJobsReadyForStartup(processID);
try
{
