Author: kwright
Date: Sat Feb 8 13:50:16 2014
New Revision: 1566014
URL: http://svn.apache.org/r1566014
Log:
More work on CONNECTORS-885, trying to extend work to handle seeding as well.
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobSeedingRecord.java
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java?rev=1566014&r1=1566013&r2=1566014&view=diff
==============================================================================
---
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
(original)
+++
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
Sat Feb 8 13:50:16 2014
@@ -513,13 +513,21 @@ public interface IJobManager
throws ManifoldCFException;
/** Retry startup.
- *@param jobStartRecord is the current job notification record.
+ *@param jobStartRecord is the current job startup record.
*@param failTime is the new fail time (-1L if none).
*@param failRetryCount is the new fail retry count (-1 if none).
*/
public void retryStartup(JobStartRecord jobStartRecord, long failTime, int
failRetryCount)
throws ManifoldCFException;
+ /** Retry seeding.
+ *@param jobSeedingRecord is the current job seeding record.
+ *@param failTime is the new fail time (-1L if none).
+ *@param failRetryCount is the new fail retry count (-1 if none).
+ */
+ public void retrySeeding(JobSeedingRecord jobSeedingRecord, long failTime,
int failRetryCount)
+ throws ManifoldCFException;
+
/** Retry notification.
*@param jobNotifyRecord is the current job notification record.
*@param failTime is the new fail time (-1L if none).
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobSeedingRecord.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobSeedingRecord.java?rev=1566014&r1=1566013&r2=1566014&view=diff
==============================================================================
---
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobSeedingRecord.java
(original)
+++
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobSeedingRecord.java
Sat Feb 8 13:50:16 2014
@@ -27,15 +27,19 @@ public class JobSeedingRecord extends Jo
/** The last synch time */
protected final long synchTime;
- /** Whether this job was started or not */
- protected boolean wasStarted = false;
+ /** The fail time, or -1L if none */
+ protected final long failTime;
+ /** The fail count, or -1 if none */
+ protected final int failRetryCount;
/** Constructor.
*/
- public JobSeedingRecord(Long jobID, long synchTime)
+ public JobSeedingRecord(Long jobID, long synchTime, long failTime, int
failRetryCount)
{
super(jobID);
this.synchTime = synchTime;
+ this.failTime = failTime;
+ this.failRetryCount = failRetryCount;
}
/** Get the synch time.
@@ -46,4 +50,20 @@ public class JobSeedingRecord extends Jo
return synchTime;
}
+ /** Get the hard fail time.
+ *@return the fail time in ms since epoch, or -1L if none.
+ */
+ public long getFailTime()
+ {
+ return failTime;
+ }
+
+ /** Get the hard fail retry count.
+ *@return the fail retry count, or -1 if none.
+ */
+ public int getFailRetryCount()
+ {
+ return failRetryCount;
+ }
+
}
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=1566014&r1=1566013&r2=1566014&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
Sat Feb 8 13:50:16 2014
@@ -3901,6 +3901,65 @@ public class JobManager implements IJobM
}
}
+ /** Retry seeding.
+ *@param jsr is the current job seeding record.
+ *@param failTime is the new fail time (-1L if none).
+ *@param failCount is the new fail retry count (-1 if none).
+ */
+ @Override
+ public void retrySeeding(JobSeedingRecord jsr, long failTime, int failCount)
+ throws ManifoldCFException
+ {
+ Long jobID = jsr.getJobID();
+ long oldFailTime = jsr.getFailTime();
+ if (oldFailTime == -1L)
+ oldFailTime = failTime;
+ failTime = oldFailTime;
+ int oldFailCount = jsr.getFailRetryCount();
+ if (oldFailCount == -1)
+ oldFailCount = failCount;
+ else
+ {
+ oldFailCount--;
+ if (failCount != -1 && oldFailCount > failCount)
+ oldFailCount = failCount;
+ }
+ failCount = oldFailCount;
+
+ while (true)
+ {
+ long sleepAmt = 0L;
+ database.beginTransaction();
+ try
+ {
+ jobs.retrySeeding(jobID,failTime,failCount);
+ database.performCommit();
+ break;
+ }
+ catch (Error e)
+ {
+ database.signalRollback();
+ throw e;
+ }
+ catch (ManifoldCFException e)
+ {
+ database.signalRollback();
+ if (e.getErrorCode() == e.DATABASE_TRANSACTION_ABORT)
+ {
+ if (Logging.perf.isDebugEnabled())
+ Logging.perf.debug("Aborted transaction resetting job seeding:
"+e.getMessage());
+ sleepAmt = getRandomAmount();
+ continue;
+ }
+ throw e;
+ }
+ finally
+ {
+ database.endTransaction();
+ sleepFor(sleepAmt);
+ }
+ }
+ }
/** Retry notification.
*@param jnr is the current job notification record.
@@ -6470,6 +6529,8 @@ public class JobManager implements IJobM
sb.append(jobs.idField).append(",")
.append(jobs.lastCheckTimeField).append(",")
+ .append(jobs.failTimeField).append(",")
+ .append(jobs.failCountField).append(",")
.append(jobs.reseedIntervalField)
.append(" FROM ").append(jobs.getTableName()).append(" WHERE ")
.append(database.buildConjunctionClause(list,new ClauseDescription[]{
@@ -6501,6 +6562,19 @@ public class JobManager implements IJobM
else
reseedTime = null;
+ Long failTimeLong = (Long)row.getValue(jobs.failTimeField);
+ Long failRetryCountLong = (Long)row.getValue(jobs.failCountField);
+ long failTime;
+ if (failTimeLong == null)
+ failTime = -1L;
+ else
+ failTime = failTimeLong.longValue();
+ int failRetryCount;
+ if (failRetryCountLong == null)
+ failRetryCount = -1;
+ else
+ failRetryCount = (int)failRetryCountLong.longValue();
+
// Mark status of job as "active/seeding". Special status is needed
so that abort
// will not complete until seeding is completed.
jobs.writeTransientStatus(jobID,jobs.STATUS_ACTIVESEEDING,reseedTime,processID);
@@ -6509,7 +6583,7 @@ public class JobManager implements IJobM
Logging.jobs.debug("Marked job "+jobID+" for seeding");
}
- rval[i] = new JobSeedingRecord(jobID,synchTime);
+ rval[i] = new
JobSeedingRecord(jobID,synchTime,failTime,failRetryCount);
i++;
}
database.performCommit();
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java?rev=1566014&r1=1566013&r2=1566014&view=diff
==============================================================================
---
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
(original)
+++
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
Sat Feb 8 13:50:16 2014
@@ -105,6 +105,7 @@ public class Jobs extends org.apache.man
public static final int STATUS_READYFORSTARTUP = 23; // Job is
marked for minimal startup; startup thread has not taken it yet.
public static final int STATUS_READYFORSTARTUPMINIMAL = 24; // Job is
marked for startup; startup thread has not taken it yet.
public static final int STATUS_READYFORDELETE = 25; // Job is
marked for delete; delete thread has not taken it yet.
+ // This state may not be used anymore!!
public static final int STATUS_ABORTINGSEEDING = 26; // Same as
aborting, but seeding process is currently active also.
public static final int STATUS_ABORTINGFORRESTART = 27; // Same as
aborting, except after abort is complete startup will happen.
public static final int STATUS_ABORTINGFORRESTARTMINIMAL = 28; // Same as
aborting, except after abort is complete startup will happen.
@@ -270,34 +271,65 @@ public class Jobs extends org.apache.man
hopmodeMap.put("V",new Integer(HOPCOUNT_NEVERDELETE));
}
- /*
- protected static Set<Integer> transientStates;
- static
- {
- transientStates = new HashSet<Integer>();
- transientStates.add(new Integer(STATUS_DELETESTARTINGUP));
- transientStates.add(new Integer(STATUS_NOTIFYINGOFCOMPLETION));
- transientStates.add(new Integer(STATUS_STARTINGUP));
- transientStates.add(new Integer(STATUS_ABORTINGSTARTINGUP));
- transientStates.add(new Integer(STATUS_STARTINGUPMINIMAL));
- transientStates.add(new Integer(STATUS_ABORTINGSTARTINGUPMINIMAL));
- transientStates.add(new Integer(STATUS_ABORTINGSTARTINGUPFORRESTART));
- transientStates.add(new
Integer(STATUS_ABORTINGSTARTINGUPFORRESTARTMINIMAL));
- transientStates.add(new Integer(STATUS_ACTIVESEEDING));
- transientStates.add(new Integer(STATUS_PAUSINGSEEDING));
- transientStates.add(new Integer(STATUS_ACTIVEWAITINGSEEDING));
- transientStates.add(new Integer(STATUS_PAUSINGWAITINGSEEDING));
- transientStates.add(new Integer(STATUS_RESUMINGSEEDING));
- transientStates.add(new Integer(STATUS_ABORTINGSEEDING));
- transientStates.add(new Integer(STATUS_ABORTINGFORRESTARTSEEDING));
- transientStates.add(new Integer(STATUS_ABORTINGFORRESTARTSEEDINGMINIMAL));
- transientStates.add(new Integer(STATUS_PAUSEDSEEDING));
- transientStates.add(new Integer(STATUS_ACTIVEWAITSEEDING));
- transientStates.add(new Integer(STATUS_PAUSEDWAITSEEDING));
- transientStates.add(new Integer(STATUS_ACTIVESEEDING_UNINSTALLED));
- transientStates.add(new Integer(STATUS_ACTIVESEEDING_NOOUTPUT));
- transientStates.add(new Integer(STATUS_ACTIVESEEDING_NEITHER));
- }
+ /* Transient vs. non-transient states
+ *==================================
+ * There are two kinds of states. The first kind of state is that which is
considered permanent; if the
+ * whole world restarts, the state would not change. Such states reflect the
underlying status of the job.
+ * The second kind of state is that which is clearly transient but is
attached to a given process. So if the
+ * process is killed, the job should leave the transient state and return to
a permanent state. For example,
+ * when a given thread decided to undertake seeding, the job is annotated
with the process ID of the thread
+ * that is performing the seeding operation. These states are called
process-transient.
+ *
+ * The permanent states are:
+ * STATUS_INACTIVE
+ * STATUS_ACTIVE
+ * STATUS_ACTIVEWAIT
+ * STATUS_PAUSEDWAIT
+ * STATUS_PAUSED
+ * STATUS_READYFORNOTIFY
+ * STATUS_READYFORDELETE
+ * STATUS_DELETING
+ * STATUS_READYFORSTARTUP
+ * STATUS_READYFORSTARTUPMINIMAL
+ * STATUS_SHUTTINGDOWN
+ * STATUS_ABORTINGFORRESTART
+ * STATUS_ABORTINGFORRESTARTMINIMAL
+ * STATUS_ACTIVE_UNINSTALLED
+ * STATUS_ACTIVE_NOOUTPUT
+ * STATUS_ACTIVE_NEITHER
+ * STATUS_DELETING_NOOUTPUT
+ * STATUS_ABORTING
+ * STATUS_ACTIVEWAITING
+ * STATUS_PAUSING
+ * STATUS_PAUSINGWAITING
+ * STATUS_RESUMING
+ * STATUS_ABORTINGSHUTTINGDOWN
+ *
+ * These are the process-transient states:
+ * STATUS_DELETESTARTINGUP
+ * STATUS_NOTIFYINGOFCOMPLETION
+ * STATUS_STARTINGUP
+ * STATUS_STARTINGUPMINIMAL
+ * STATUS_ABORTINGSTARTINGUPFORRESTART
+ * STATUS_ABORTINGSTARTINGUPFORRESTARTMINIMAL
+ * STATUS_ACTIVEWAITINGSEEDING
+ * STATUS_ACTIVEWAITSEEDING
+ * STATUS_ACTIVESEEDING
+ * STATUS_PAUSINGSEEDING
+ * STATUS_PAUSINGWAITINGSEEDING
+ * STATUS_PAUSEDSEEDING
+ * STATUS_PAUSEDWAITSEEDING
+ * STATUS_RESUMINGSEEDING
+ * STATUS_ABORTINGSEEDING
+ * STATUS_ABORTINGFORRESTARTSEEDING
+ * STATUS_ABORTINGFORRESTARTSEEDINGMINIMAL
+ * STATUS_ACTIVESEEDING_UNINSTALLED
+ * STATUS_ACTIVESEEDING_NOOUTPUT
+ * STATUS_ACTIVESEEDING_NEITHER
+ * STATUS_ABORTINGSTARTINGUP
+ * STATUS_ABORTINGSTARTINGUPMINIMAL
+ *
+ *
*/
// Local variables
@@ -407,6 +439,7 @@ public class Jobs extends org.apache.man
IndexDescription statusProcessIndex = new IndexDescription(false,new
String[]{statusField,processIDField});
IndexDescription connectionIndex = new IndexDescription(false,new
String[]{connectionNameField});
IndexDescription outputIndex = new IndexDescription(false,new
String[]{outputNameField});
+ IndexDescription failTimeIndex = new IndexDescription(false,new
String[]{failTimeField});
// Get rid of indexes that shouldn't be there
Map indexes = getTableIndexes(null,null);
@@ -424,6 +457,8 @@ public class Jobs extends org.apache.man
connectionIndex = null;
else if (outputIndex != null && id.equals(outputIndex))
outputIndex = null;
+ else if (failTimeIndex != null && id.equals(failTimeIndex))
+ failTimeIndex = null;
else if (indexName.indexOf("_pkey") == -1)
// This index shouldn't be here; drop it
performRemoveIndex(indexName);
@@ -438,6 +473,8 @@ public class Jobs extends org.apache.man
performAddIndex(null,connectionIndex);
if (outputIndex != null)
performAddIndex(null,outputIndex);
+ if (failTimeIndex != null)
+ performAddIndex(null,failTimeIndex);
break;
@@ -1107,7 +1144,9 @@ public class Jobs extends org.apache.man
ArrayList list = new ArrayList();
HashMap map = new HashMap();
String query;
-
+
+ // All jobs with non-null failTime and failCount get set to null failTime
and failCount
+
// Starting up delete goes back to just being ready for delete
query = buildConjunctionClause(list,new ClauseDescription[]{
new UnitaryClause(statusField,statusToString(STATUS_DELETESTARTINGUP))});
@@ -1121,8 +1160,6 @@ public class Jobs extends org.apache.man
new
UnitaryClause(statusField,statusToString(STATUS_NOTIFYINGOFCOMPLETION))});
map.put(statusField,statusToString(STATUS_READYFORNOTIFY));
map.put(processIDField,null);
- map.put(failTimeField,null);
- map.put(failCountField,null);
performUpdate(map,"WHERE "+query,list,invKey);
// Starting up or aborting starting up goes back to just being ready
@@ -1252,7 +1289,12 @@ public class Jobs extends org.apache.man
public void restartCluster()
throws ManifoldCFException
{
- // Does nothing
+ HashMap map = new HashMap();
+ map.put(failTimeField,null);
+ ArrayList list = new ArrayList();
+ String query = buildConjunctionClause(list,new ClauseDescription[]{
+ new NullCheckClause(failTimeField,false)});
+ performUpdate(map,"WHERE "+query,list,null);
}
/** Signal to a job that its underlying output connector has gone away.
@@ -1715,9 +1757,111 @@ public class Jobs extends org.apache.man
map.put(failCountField,null);
else
map.put(failCountField,failCount);
+ map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
+ /** Retry seeding.
+ *@param jobID is the job identifier.
+ *@param failTime is the fail time, -1 == none
+ *@param failCount is the fail count to use, -1 == none.
+ */
+ public void retrySeeding(Long jobID, long failTime, int failCount)
+ throws ManifoldCFException
+ {
+ beginTransaction();
+ try
+ {
+ ArrayList list = new ArrayList();
+ String query = buildConjunctionClause(list,new ClauseDescription[]{
+ new UnitaryClause(idField,jobID)});
+ // Map the field values we see to those we want.
+ // This is exactly equivalent to finishing seeding, except for fail time
etc.
+ IResultSet set = performQuery("SELECT "+statusField+" FROM
"+getTableName()+" WHERE "+
+ query+" FOR UPDATE",list,null,null);
+ if (set.getRowCount() == 0)
+ throw new ManifoldCFException("Can't find job "+jobID.toString());
+ IResultRow row = set.getRow(0);
+ int status = stringToStatus((String)row.getValue(statusField));
+ int newStatus;
+ switch (status)
+ {
+ case STATUS_ACTIVESEEDING:
+ newStatus = STATUS_ACTIVE;
+ break;
+ case STATUS_PAUSINGSEEDING:
+ newStatus = STATUS_PAUSING;
+ break;
+ case STATUS_ACTIVEWAITINGSEEDING:
+ newStatus = STATUS_ACTIVEWAITING;
+ break;
+ case STATUS_PAUSINGWAITINGSEEDING:
+ newStatus = STATUS_PAUSINGWAITING;
+ break;
+ case STATUS_ACTIVESEEDING_UNINSTALLED:
+ newStatus = STATUS_ACTIVE_UNINSTALLED;
+ break;
+ case STATUS_ACTIVESEEDING_NOOUTPUT:
+ newStatus = STATUS_ACTIVE_NOOUTPUT;
+ break;
+ case STATUS_ACTIVESEEDING_NEITHER:
+ newStatus = STATUS_ACTIVE_NEITHER;
+ break;
+ case STATUS_ACTIVEWAITSEEDING:
+ newStatus = STATUS_ACTIVEWAIT;
+ break;
+ case STATUS_PAUSEDSEEDING:
+ newStatus = STATUS_PAUSED;
+ break;
+ case STATUS_PAUSEDWAITSEEDING:
+ newStatus = STATUS_PAUSEDWAIT;
+ break;
+ case STATUS_ABORTINGSEEDING:
+ newStatus = STATUS_ABORTING;
+ break;
+ case STATUS_ABORTINGFORRESTARTSEEDING:
+ newStatus = STATUS_ABORTINGFORRESTART;
+ break;
+ case STATUS_ABORTINGFORRESTARTSEEDINGMINIMAL:
+ newStatus = STATUS_ABORTINGFORRESTARTMINIMAL;
+ break;
+ default:
+ throw new ManifoldCFException("Unexpected job status encountered:
"+Integer.toString(status));
+ }
+ HashMap map = new HashMap();
+ map.put(statusField,statusToString(newStatus));
+ if (failTime == -1L)
+ map.put(failTimeField,null);
+ else
+ map.put(failTimeField,new Long(failTime));
+ if (failCount == -1)
+ map.put(failCountField,null);
+ else
+ map.put(failCountField,failCount);
+ map.put(processIDField,null);
+ performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
+ }
+ catch (Error e)
+ {
+ signalRollback();
+ throw e;
+ }
+ catch (RuntimeException e)
+ {
+ signalRollback();
+ throw e;
+ }
+ catch (ManifoldCFException e)
+ {
+ signalRollback();
+ throw e;
+ }
+ finally
+ {
+ endTransaction();
+ }
+ }
+
/** Retry notification.
*@param jobID is the job identifier.
*@param failTime is the fail time, -1 == none
@@ -1739,6 +1883,7 @@ public class Jobs extends org.apache.man
map.put(failCountField,null);
else
map.put(failCountField,failCount);
+ map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
@@ -1815,6 +1960,11 @@ public class Jobs extends org.apache.man
signalRollback();
throw e;
}
+ catch (RuntimeException e)
+ {
+ signalRollback();
+ throw e;
+ }
catch (Error e)
{
signalRollback();
@@ -1873,6 +2023,11 @@ public class Jobs extends org.apache.man
signalRollback();
throw e;
}
+ catch (RuntimeException e)
+ {
+ signalRollback();
+ throw e;
+ }
catch (Error e)
{
signalRollback();
@@ -1950,6 +2105,7 @@ public class Jobs extends org.apache.man
// Clear out the retry fields we might have set
map.put(failTimeField,null);
map.put(failCountField,null);
+ map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
catch (ManifoldCFException e)
@@ -1957,6 +2113,11 @@ public class Jobs extends org.apache.man
signalRollback();
throw e;
}
+ catch (RuntimeException e)
+ {
+ signalRollback();
+ throw e;
+ }
catch (Error e)
{
signalRollback();
@@ -2035,7 +2196,10 @@ public class Jobs extends org.apache.man
}
HashMap map = new HashMap();
map.put(statusField,statusToString(newStatus));
+ map.put(processIDField,null);
map.put(lastCheckTimeField,new Long(seedTime));
+ map.put(failTimeField,null);
+ map.put(failCountField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
catch (Error e)
@@ -2043,6 +2207,11 @@ public class Jobs extends org.apache.man
signalRollback();
throw e;
}
+ catch (RuntimeException e)
+ {
+ signalRollback();
+ throw e;
+ }
catch (ManifoldCFException e)
{
signalRollback();
@@ -2151,18 +2320,21 @@ public class Jobs extends org.apache.man
status == STATUS_ABORTINGSHUTTINGDOWN)
return false;
int newStatus;
+ boolean clearProcessID = false;
switch (status)
{
- case STATUS_SHUTTINGDOWN:
- newStatus = STATUS_ABORTINGSHUTTINGDOWN;
- break;
case STATUS_STARTINGUP:
case STATUS_ABORTINGSTARTINGUPFORRESTART:
newStatus = STATUS_ABORTING;
+ clearProcessID = true;
break;
case STATUS_STARTINGUPMINIMAL:
case STATUS_ABORTINGSTARTINGUPFORRESTARTMINIMAL:
newStatus = STATUS_ABORTING;
+ clearProcessID = true;
+ break;
+ case STATUS_SHUTTINGDOWN:
+ newStatus = STATUS_ABORTINGSHUTTINGDOWN;
break;
case STATUS_READYFORSTARTUP:
case STATUS_READYFORSTARTUPMINIMAL:
@@ -2192,7 +2364,8 @@ public class Jobs extends org.apache.man
case STATUS_PAUSEDWAITSEEDING:
case STATUS_ABORTINGFORRESTARTSEEDING:
case STATUS_ABORTINGFORRESTARTSEEDINGMINIMAL:
- newStatus = STATUS_ABORTINGSEEDING;
+ newStatus = STATUS_ABORTING;
+ clearProcessID = true;
break;
default:
throw new ManifoldCFException("Job "+jobID+" is not active");
@@ -2201,6 +2374,8 @@ public class Jobs extends org.apache.man
HashMap map = new HashMap();
map.put(statusField,statusToString(newStatus));
map.put(errorField,errorText);
+ if (clearProcessID)
+ map.put(processIDField,null);
map.put(failTimeField,null);
map.put(failCountField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
@@ -2402,6 +2577,8 @@ public class Jobs extends org.apache.man
map.put(statusField,statusToString(status));
map.put(processIDField,processID);
map.put(reseedTimeField,reseedTime);
+ map.put(failTimeField,null);
+ map.put(failCountField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java?rev=1566014&r1=1566013&r2=1566014&view=diff
==============================================================================
---
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
(original)
+++
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
Sat Feb 8 13:50:16 2014
@@ -155,11 +155,49 @@ public class SeedingThread extends Threa
}
catch (ServiceInterruption e)
{
- // Note the service interruption
- Logging.threads.warn("Service interruption for job
"+jobID,e);
- long retryInterval = e.getRetryTime() - currentTime;
- if (retryInterval >= 0L && retryInterval < waitTime)
- waitTime = retryInterval;
+ if (!e.jobInactiveAbort())
+ {
+ Logging.jobs.warn("Seeding service interruption reported
for job "+
+ jobID+" connection '"+connection.getName()+"': "+
+ e.getMessage());
+ }
+
+ ManifoldCFException abortOnFail;
+ if (!e.jobInactiveAbort() && e.isAbortOnFail())
+ abortOnFail = new ManifoldCFException("Failure
performing seeding"+((e.getCause()!=null)?":
"+e.getCause().getMessage():""),e.getCause());
+ else
+ abortOnFail = null;
+
+ // If either we are going to be requeuing beyond the fail
time, OR
+ // the number of retries available has hit 0, THEN we
treat this
+ // as either an "ignore" or a hard error.
+ if (!e.jobInactiveAbort() && (jsr.getFailTime() != -1L &&
jsr.getFailTime() < e.getRetryTime() ||
+ jsr.getFailRetryCount() == 0))
+ {
+ // Treat this as a hard failure.
+ if (e.isAbortOnFail())
+ {
+ // Note the error in the job, and transition to
inactive state
+ if
(jobManager.errorAbort(jobID,(abortOnFail==null)?"":"Repeated service
interruptions during seeding: "+abortOnFail.getMessage()+": ending job") &&
+ abortOnFail != null)
+ Logging.jobs.error("Repeated service interruptions
during seeding: "+abortOnFail.getMessage(),abortOnFail);
+ jsr.noteStarted();
+ }
+ else
+ {
+ // Not sure this can happen -- but just transition
silently to active state
+ jobManager.noteJobSeeded(jobID,currentTime);
+ jsr.noteStarted();
+ }
+ }
+ else
+ {
+ // Reset the job to the READYFORSTARTUP state, updating
the failtime and failcount fields
+ if (abortOnFail != null)
+
Logging.jobs.warn(abortOnFail.getMessage(),abortOnFail);
+
jobManager.retrySeeding(jsr,e.getFailTime(),e.getFailRetryCount());
+ jsr.noteStarted();
+ }
// Go on to the next job
continue;
}
@@ -186,8 +224,7 @@ public class SeedingThread extends Threa
throw e;
if (jobManager.errorAbort(jobID,e.getMessage()))
Logging.threads.error("Exception tossed: "+e.getMessage(),e);
- // We DO have to clean up, because there is otherwise no
- // way the job will be reset from the seeding state.
+ jsr.noteStarted();
}
}
}
