Author: kwright
Date: Fri Feb 7 11:39:40 2014
New Revision: 1565620
URL: http://svn.apache.org/r1565620
Log:
Fix for CONNECTORS-885. Use the new failType/failCount jobs fields to deal
with ServiceInterruptions correctly at startup time.
Modified:
manifoldcf/trunk/CHANGES.txt
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/JobStartRecord.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/StartupThread.java
Modified: manifoldcf/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/CHANGES.txt?rev=1565620&r1=1565619&r2=1565620&view=diff
==============================================================================
--- manifoldcf/trunk/CHANGES.txt (original)
+++ manifoldcf/trunk/CHANGES.txt Fri Feb 7 11:39:40 2014
@@ -3,6 +3,9 @@ $Id$
======================= 1.6-dev =====================
+CONNECTORS-885: Handle errors during startup better.
+(Karl Wright)
+
CONNECTORS-883: Handle errors from the output connector during
notification better.
(Karl Wright)
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=1565620&r1=1565619&r2=1565620&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
Fri Feb 7 11:39:40 2014
@@ -512,12 +512,20 @@ public interface IJobManager
public void resetCleaningDocumentMultiple(DocumentDescription[]
documentDescriptions, long checkTime)
throws ManifoldCFException;
+ /** Retry startup.
+ *@param jobStartRecord is the current job notification 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 notification.
*@param jobNotifyRecord is the current job notification record.
*@param failTime is the new fail time (-1L if none).
*@param failRetryCount is the new fail retry count (-1 if none).
*/
- public void retryNotification(JobNotifyRecord jnr, long failTime, int
failRetryCount)
+ public void retryNotification(JobNotifyRecord jobNotifyRecord, long
failTime, int failRetryCount)
throws ManifoldCFException;
/** Add an initial set of documents to the queue.
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobStartRecord.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobStartRecord.java?rev=1565620&r1=1565619&r2=1565620&view=diff
==============================================================================
---
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobStartRecord.java
(original)
+++
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/JobStartRecord.java
Fri Feb 7 11:39:40 2014
@@ -29,14 +29,20 @@ public class JobStartRecord extends JobR
protected final long synchTime;
/** The requestMinimum flag */
protected final boolean requestMinimum;
+ /** The fail time, or -1L if none */
+ protected final long failTime;
+ /** The fail count, or -1 if none */
+ protected final int failRetryCount;
/** Constructor.
*/
- public JobStartRecord(Long jobID, long synchTime, boolean requestMinimum)
+ public JobStartRecord(Long jobID, long synchTime, boolean requestMinimum,
long failTime, int failRetryCount)
{
super(jobID);
this.synchTime = synchTime;
this.requestMinimum = requestMinimum;
+ this.failTime = failTime;
+ this.failRetryCount = failRetryCount;
}
/** Get the synch time.
@@ -55,4 +61,20 @@ public class JobStartRecord extends JobR
return requestMinimum;
}
+ /** 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=1565620&r1=1565619&r2=1565620&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
Fri Feb 7 11:39:40 2014
@@ -3572,67 +3572,6 @@ public class JobManager implements IJobM
}
}
- /** Retry notification.
- *@param jobNotifyRecord is the current job notification 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 retryNotification(JobNotifyRecord jnr, long failTime, int
failCount)
- throws ManifoldCFException
- {
- Long jobID = jnr.getJobID();
- long oldFailTime = jnr.getFailTime();
- if (oldFailTime == -1L)
- oldFailTime = failTime;
- failTime = oldFailTime;
- int oldFailCount = jnr.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.retryNotification(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
notification: "+e.getMessage());
- sleepAmt = getRandomAmount();
- continue;
- }
- throw e;
- }
- finally
- {
- database.endTransaction();
- sleepFor(sleepAmt);
- }
- }
-
- }
-
/** Reset a set of cleaning documents for further processing in the future.
* This method is called after some unknown number of the documents were
cleaned, but then an ingestion service interruption occurred.
@@ -3902,6 +3841,128 @@ public class JobManager implements IJobM
return map;
}
+ /** Retry startup.
+ *@param jsr is the current job notification 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 retryStartup(JobStartRecord 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.retryStartup(jobID,jsr.getRequestMinimum(),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 startup:
"+e.getMessage());
+ sleepAmt = getRandomAmount();
+ continue;
+ }
+ throw e;
+ }
+ finally
+ {
+ database.endTransaction();
+ sleepFor(sleepAmt);
+ }
+ }
+ }
+
+
+ /** Retry notification.
+ *@param jnr is the current job notification 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 retryNotification(JobNotifyRecord jnr, long failTime, int
failCount)
+ throws ManifoldCFException
+ {
+ Long jobID = jnr.getJobID();
+ long oldFailTime = jnr.getFailTime();
+ if (oldFailTime == -1L)
+ oldFailTime = failTime;
+ failTime = oldFailTime;
+ int oldFailCount = jnr.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.retryNotification(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
notification: "+e.getMessage());
+ sleepAmt = getRandomAmount();
+ continue;
+ }
+ throw e;
+ }
+ finally
+ {
+ database.endTransaction();
+ sleepFor(sleepAmt);
+ }
+ }
+
+ }
+
/** Add an initial set of documents to the queue.
* This method is called during job startup, when the queue is being loaded.
* A set of document references is passed to this method, which updates the
status of the document
@@ -6568,6 +6629,8 @@ public class JobManager implements IJobM
ArrayList list = new ArrayList();
sb.append(jobs.idField).append(",")
+ .append(jobs.failTimeField).append(",")
+ .append(jobs.failCountField).append(",")
.append(jobs.lastCheckTimeField).append(",")
.append(jobs.statusField)
.append(" FROM ").append(jobs.getTableName()).append(" WHERE ")
@@ -6587,6 +6650,18 @@ public class JobManager implements IJobM
Long jobID = (Long)row.getValue(jobs.idField);
Long x = (Long)row.getValue(jobs.lastCheckTimeField);
int status =
jobs.stringToStatus((String)row.getValue(jobs.statusField));
+ 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();
boolean requestMinimum = (status ==
jobs.STATUS_READYFORSTARTUPMINIMAL);
@@ -6601,7 +6676,7 @@ public class JobManager implements IJobM
Logging.jobs.debug("Marked job "+jobID+" for startup");
}
- rval[i] = new JobStartRecord(jobID,synchTime,requestMinimum);
+ rval[i] = new
JobStartRecord(jobID,synchTime,requestMinimum,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=1565620&r1=1565619&r2=1565620&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
Fri Feb 7 11:39:40 2014
@@ -98,7 +98,9 @@ public class Jobs extends org.apache.man
public static final int STATUS_ABORTING = 18; //
Aborting (not yet aborted because documents still being processed)
public static final int STATUS_STARTINGUP = 19; //
Loading the queue (will go into ACTIVE if successful, or INACTIVE if not)
public static final int STATUS_STARTINGUPMINIMAL = 20; // Loading
the queue for minimal job run (will go into ACTIVE if successful, or INACTIVE
if not)
+ // This state may not be used anymore!!
public static final int STATUS_ABORTINGSTARTINGUP = 21; // Will abort
once the queue loading is complete
+ // This state may not be used anymore!!
public static final int STATUS_ABORTINGSTARTINGUPMINIMAL = 22; // Will
abort once the queue loading is complete
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.
@@ -1688,6 +1690,33 @@ public class Jobs extends org.apache.man
}
+ /** Retry startup.
+ *@param jobID is the job identifier.
+ *@param requestMinimum is true for a minimal crawl.
+ *@param failTime is the fail time, -1 == none
+ *@param failCount is the fail count to use, -1 == none.
+ */
+ public void retryStartup(Long jobID, boolean requestMinimum, long failTime,
int failCount)
+ throws ManifoldCFException
+ {
+ ArrayList list = new ArrayList();
+ String query = buildConjunctionClause(list,new ClauseDescription[]{
+ new UnitaryClause(idField,jobID)});
+ HashMap map = new HashMap();
+ if (requestMinimum)
+ map.put(statusField,statusToString(STATUS_READYFORSTARTUPMINIMAL));
+ else
+ map.put(statusField,statusToString(STATUS_READYFORSTARTUP));
+ 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);
+ performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
+ }
/** Retry notification.
*@param jobID is the job identifier.
@@ -1918,6 +1947,9 @@ public class Jobs extends org.apache.man
}
// The seeding was complete or we wouldn't have gotten called, so at
least note that.
map.put(lastCheckTimeField,new Long(startTime));
+ // Clear out the retry fields we might have set
+ map.put(failTimeField,null);
+ map.put(failCountField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
}
catch (ManifoldCFException e)
@@ -2091,6 +2123,7 @@ public class Jobs extends org.apache.man
map.put(errorField,errorText);
map.put(failTimeField,null);
map.put(failCountField,null);
+ map.put(processIDField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
return true;
}
@@ -2125,11 +2158,11 @@ public class Jobs extends org.apache.man
break;
case STATUS_STARTINGUP:
case STATUS_ABORTINGSTARTINGUPFORRESTART:
- newStatus = STATUS_ABORTINGSTARTINGUP;
+ newStatus = STATUS_ABORTING;
break;
case STATUS_STARTINGUPMINIMAL:
case STATUS_ABORTINGSTARTINGUPFORRESTARTMINIMAL:
- newStatus = STATUS_ABORTINGSTARTINGUPMINIMAL;
+ newStatus = STATUS_ABORTING;
break;
case STATUS_READYFORSTARTUP:
case STATUS_READYFORSTARTUPMINIMAL:
@@ -2168,6 +2201,8 @@ public class Jobs extends org.apache.man
HashMap map = new HashMap();
map.put(statusField,statusToString(newStatus));
map.put(errorField,errorText);
+ map.put(failTimeField,null);
+ map.put(failCountField,null);
performUpdate(map,"WHERE "+query,list,new StringSet(getJobStatusKey()));
return true;
}
Modified:
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java?rev=1565620&r1=1565619&r2=1565620&view=diff
==============================================================================
---
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
(original)
+++
manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
Fri Feb 7 11:39:40 2014
@@ -158,11 +158,49 @@ public class StartupThread extends Threa
}
catch (ServiceInterruption e)
{
- // Note the service interruption
- Logging.threads.warn("Service interruption for job
"+jobID+": "+e.getMessage(),e);
- long retryInterval = e.getRetryTime() - currentTime;
- if (retryInterval >= 0L && retryInterval < waitTime)
- waitTime = retryInterval;
+ if (!e.jobInactiveAbort())
+ {
+ Logging.jobs.warn("Startup service interruption reported
for job "+
+ jobID+" connection '"+connection.getName()+"': "+
+ e.getMessage());
+ }
+
+ ManifoldCFException abortOnFail;
+ if (!e.jobInactiveAbort() && e.isAbortOnFail())
+ abortOnFail = new ManifoldCFException("Failure
performing startup"+((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 startup: "+abortOnFail.getMessage()+": ending job") &&
+ abortOnFail != null)
+ Logging.jobs.error("Repeated service interruptions
during startup: "+abortOnFail.getMessage(),abortOnFail);
+ jsr.noteStarted();
+ }
+ else
+ {
+ // Not sure this can happen -- but just transition
silently to active state
+ jobManager.noteJobStarted(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.retryStartup(jsr,e.getFailTime(),e.getFailRetryCount());
+ jsr.noteStarted();
+ }
// Go on to the next job
continue;
}
@@ -182,10 +220,10 @@ public class StartupThread extends Threa
throw new InterruptedException();
if (e.getErrorCode() ==
ManifoldCFException.DATABASE_CONNECTION_ERROR)
throw e;
- // Note: The error abort below will put the job in the
"ABORTINGSTARTUP" state. We still need a reset at that point
- // to get all the way back to an "aborting" state.
+ // Note: The error abort below will put the job in the
"ABORTING" state.
if (jobManager.errorAbort(jobID,e.getMessage()))
Logging.threads.error("Exception tossed: "+e.getMessage(),e);
+ jsr.noteStarted();
}
}
}
