manifoldcf-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kwri...@apache.org
Subject svn commit: r1638083 - in /manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler: interfaces/IJobManager.java jobs/JobManager.java jobs/JobQueue.java system/SetPriorityThread.java
Date Tue, 11 Nov 2014 13:40:55 GMT
Author: kwright
Date: Tue Nov 11 13:40:55 2014
New Revision: 1638083

URL: http://svn.apache.org/r1638083
Log:
Use database fields to permit multiple reprioritization threads to be active at the same time,
cross cluster.

Modified:
    manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
    manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
    manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java
    manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java

Modified: manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java?rev=1638083&r1=1638082&r2=1638083&view=diff
==============================================================================
--- manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
(original)
+++ manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
Tue Nov 11 13:40:55 2014
@@ -225,10 +225,11 @@ public interface IJobManager
 
   /** Get a list of not-yet-processed documents to reprioritize.  Documents in all jobs will
be
   * returned by this method.  Up to n document descriptions will be returned.
+  *@param processID is the process that requests the reprioritization documents.
   *@param n is the maximum number of document descriptions desired.
   *@return the document descriptions.
   */
-  public DocumentDescription[] getNextNotYetProcessedReprioritizationDocuments(int n)
+  public DocumentDescription[] getNextNotYetProcessedReprioritizationDocuments(String processID,
int n)
     throws ManifoldCFException;
 
   /** Save a set of document priorities.  In the case where a document was eligible to have
its

Modified: manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java?rev=1638083&r1=1638082&r2=1638083&view=diff
==============================================================================
--- manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
(original)
+++ manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
Tue Nov 11 13:40:55 2014
@@ -33,6 +33,7 @@ public class JobManager implements IJobM
   public static final String _rcsid = "@(#)$Id: JobManager.java 998576 2010-09-19 01:11:02Z
kwright $";
 
   protected static final String stufferLock = "_STUFFER_";
+  protected static final String reprioritizationLock = "_REPRIORITIZER_";
   protected static final String deleteStufferLock = "_DELETESTUFFER_";
   protected static final String expireStufferLock = "_EXPIRESTUFFER_";
   protected static final String cleanStufferLock = "_CLEANSTUFFER_";
@@ -2064,47 +2065,106 @@ public class JobManager implements IJobM
 
   /** Get a list of not-yet-processed documents to reprioritize.  Documents in all jobs will
be
   * returned by this method.  Up to n document descriptions will be returned.
+  *@param processID is the process that requests the reprioritization documents.
   *@param n is the maximum number of document descriptions desired.
   *@return the document descriptions.
   */
   @Override
-  public DocumentDescription[] getNextNotYetProcessedReprioritizationDocuments(int n)
+  public DocumentDescription[] getNextNotYetProcessedReprioritizationDocuments(String processID,
int n)
     throws ManifoldCFException
   {
-    StringBuilder sb = new StringBuilder("SELECT ");
-    ArrayList list = new ArrayList();
+    // Retry loop - in case we get a deadlock despite our best efforts
+    while (true)
+    {
+      long sleepAmt = 0L;
 
-    // This query MUST return only documents that are in a pending state which belong to
an active job!!!
+      // Write lock insures that only one thread cluster-wide can be doing this at a given
time, so FOR UPDATE is unneeded.
+      lockManager.enterWriteLock(reprioritizationLock);
+      try
+      {
+        // Start the transaction now
+        database.beginTransaction();
+        try
+        {
 
-    sb.append(jobQueue.idField).append(",")
-      .append(jobQueue.docHashField).append(",")
-      .append(jobQueue.docIDField).append(",")
-      .append(jobQueue.jobIDField)
-      .append(" FROM ").append(jobQueue.getTableName()).append(" WHERE ")
-      .append(database.buildConjunctionClause(list,new ClauseDescription[]{
-        new UnitaryClause(jobQueue.needPriorityField,jobQueue.needPriorityToString(true))})).append("
")
-      .append(database.constructOffsetLimitClause(0,n));
+          StringBuilder sb = new StringBuilder("SELECT ");
+          ArrayList list = new ArrayList();
 
-    // Analyze jobqueue tables unconditionally, since it's become much more sensitive in
8.3 than it used to be.
-    //jobQueue.unconditionallyAnalyzeTables();
+          // This query MUST return only documents that are in a pending state which belong
to an active job!!!
 
-    //IResultSet set = database.performQuery(sb.toString(),list,null,null,n,null);
-    IResultSet set = database.performQuery(sb.toString(),list,null,null);
+          sb.append(jobQueue.idField).append(",")
+            .append(jobQueue.docHashField).append(",")
+            .append(jobQueue.docIDField).append(",")
+            .append(jobQueue.jobIDField)
+            .append(" FROM ").append(jobQueue.getTableName()).append(" WHERE ")
+            .append(database.buildConjunctionClause(list,new ClauseDescription[]{
+              new UnitaryClause(jobQueue.needPriorityField,jobQueue.needPriorityToString(JobQueue.NEEDPRIORITY_TRUE))})).append("
")
+            .append(database.constructOffsetLimitClause(0,n));
 
-    DocumentDescription[] rval = new DocumentDescription[set.getRowCount()];
+          // Analyze jobqueue tables unconditionally, since it's become much more sensitive
in 8.3 than it used to be.
+          //jobQueue.unconditionallyAnalyzeTables();
 
-    int i = 0;
-    while (i < set.getRowCount())
-    {
-      IResultRow row = set.getRow(i);
-      rval[i] =new DocumentDescription((Long)row.getValue(jobQueue.idField),
-        (Long)row.getValue(jobQueue.jobIDField),
-        (String)row.getValue(jobQueue.docHashField),
-        (String)row.getValue(jobQueue.docIDField));
-      i++;
-    }
+          //IResultSet set = database.performQuery(sb.toString(),list,null,null,n,null);
+          IResultSet set = database.performQuery(sb.toString(),list,null,null);
 
-    return rval;
+          DocumentDescription[] rval = new DocumentDescription[set.getRowCount()];
+          // To avoid deadlock, we want to update the document id hashes in order.  This
means reading into a structure I can sort by docid hash,
+          // before updating any rows in jobqueue.
+          String[] docIDHashes = new String[set.getRowCount()];
+          Map<String,DocumentDescription> storageMap = new HashMap<String,DocumentDescription>();
+
+          for (int i = 0 ; i < set.getRowCount() ; i++)
+          {
+            IResultRow row = set.getRow(i);
+            String docHash = (String)row.getValue(jobQueue.docHashField);
+            Long jobID = (Long)row.getValue(jobQueue.jobIDField);
+            rval[i] = new DocumentDescription((Long)row.getValue(jobQueue.idField),
+              jobID,
+              docHash,
+              (String)row.getValue(jobQueue.docIDField));
+            // Compute the doc ID hash
+            docIDHashes[i] = docHash + ":" + jobID;
+            storageMap.put(docIDHashes[i],rval[i]);
+          }
+          
+          java.util.Arrays.sort(docIDHashes);
+          for (String docIDHash : docIDHashes)
+          {
+            DocumentDescription dd = storageMap.get(docIDHash);
+            jobQueue.markNeedPriorityInProgress(dd.getID(),processID);
+          }
+
+          database.performCommit();
+          return rval;
+        }
+        catch (ManifoldCFException e)
+        {
+          database.signalRollback();
+          if (e.getErrorCode() == e.DATABASE_TRANSACTION_ABORT)
+          {
+            if (Logging.perf.isDebugEnabled())
+              Logging.perf.debug("Aborted transaction finding documents for reprioritization:
"+e.getMessage());
+            sleepAmt = getRandomAmount();
+            continue;
+          }
+          throw e;
+        }
+        catch (Error e)
+        {
+          database.signalRollback();
+          throw e;
+        }
+        finally
+        {
+          database.endTransaction();
+        }
+      }
+      finally
+      {
+        lockManager.leaveWriteLock(reprioritizationLock);
+        sleepFor(sleepAmt);
+      }
+    }
   }
 
   /** Save a set of document priorities.  In the case where a document was eligible to have
its
@@ -2159,8 +2219,8 @@ public class JobManager implements IJobM
           {
             IResultRow row = set.getRow(0);
             // Grab the needPriority value
-            boolean needPriority = jobQueue.stringToNeedPriority((String)row.getValue(jobQueue.needPriorityField));
-            if (needPriority)
+            int needPriority = jobQueue.stringToNeedPriority((String)row.getValue(jobQueue.needPriorityField));
+            if (needPriority == JobQueue.NEEDPRIORITY_INPROGRESS)
             {
               IPriorityCalculator priority = priorities[index];
               jobQueue.writeDocPriority(dd.getID(),priority);
@@ -2882,8 +2942,8 @@ public class JobManager implements IJobM
             // To avoid deadlock, we want to update the document id hashes in order.  This
means reading into a structure I can sort by docid hash,
             // before updating any rows in jobqueue.
             String[] docIDHashes = new String[set.getRowCount()];
-            Map storageMap = new HashMap();
-            Map statusMap = new HashMap();
+            Map<String,DocumentDescription> storageMap = new HashMap<String,DocumentDescription>();
+            Map<String,Integer> statusMap = new HashMap<String,Integer>();
 
             int i = 0;
             while (i < set.getRowCount())

Modified: manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java
URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java?rev=1638083&r1=1638082&r2=1638083&view=diff
==============================================================================
--- manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java
(original)
+++ manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java
Tue Nov 11 13:40:55 2014
@@ -45,6 +45,7 @@ import java.util.*;
  * <tr><td>processid</td><td>VARCHAR(16)</td><td></td></tr>
  * <tr><td>seedingprocessid</td><td>VARCHAR(16)</td><td></td></tr>
  * <tr><td>needpriority</td><td>CHAR(1)</td><td></td></tr>
+ * <tr><td>needpriorityprocessid</td><td>VARCHAR(16)</td><td></td></tr>
  * </table>
  * <br><br>
  * 
@@ -77,6 +78,11 @@ public class JobQueue extends org.apache
   public final static int ACTION_RESCAN = 0;
   public final static int ACTION_REMOVE = 1;
 
+  // Need priority status
+  public final static int NEEDPRIORITY_FALSE = 0;
+  public final static int NEEDPRIORITY_INPROGRESS = 1;
+  public final static int NEEDPRIORITY_TRUE = 2;
+  
   // State descriptions are as follows:
   // PENDING means a newly-added reference that has not been scanned before.
   // ACTIVE means a newly-added reference that is being scanned for the first time.
@@ -118,15 +124,15 @@ public class JobQueue extends org.apache
   public static final String processIDField = "processid";
   public static final String seedingProcessIDField = "seedingprocessid";
   public static final String needPriorityField = "needpriority";
+  public static final String needPriorityProcessIDField = "needpriorityprocessid";
   
   public static final double noDocPriorityValue = 1e9;
   public static final Double nullDocPriority = new Double(noDocPriorityValue + 1.0);
   
-  protected static Map statusMap;
+  protected static final Map<String,Integer> statusMap = new HashMap<String,Integer>();
 
   static
   {
-    statusMap = new HashMap();
     statusMap.put("P",new Integer(STATUS_PENDING));
     statusMap.put("A",new Integer(STATUS_ACTIVE));
     statusMap.put("C",new Integer(STATUS_COMPLETE));
@@ -142,25 +148,32 @@ public class JobQueue extends org.apache
     statusMap.put("H",new Integer(STATUS_HOPCOUNTREMOVED));
   }
 
-  protected static Map seedstatusMap;
+  protected static final Map<String,Integer> seedstatusMap = new HashMap<String,Integer>();
 
   static
   {
-    seedstatusMap = new HashMap();
     seedstatusMap.put("F",new Integer(SEEDSTATUS_NOTSEED));
     seedstatusMap.put("S",new Integer(SEEDSTATUS_SEED));
     seedstatusMap.put("N",new Integer(SEEDSTATUS_NEWSEED));
   }
 
-  protected static Map actionMap;
+  protected static final Map<String,Integer> actionMap = new HashMap<String,Integer>();
 
   static
   {
-    actionMap = new HashMap();
     actionMap.put("R",new Integer(ACTION_RESCAN));
     actionMap.put("D",new Integer(ACTION_REMOVE));
   }
 
+  protected static final Map<String,Integer> needPriorityMap = new HashMap<String,Integer>();
+  
+  static
+  {
+    needPriorityMap.put("T",new Integer(NEEDPRIORITY_TRUE));
+    needPriorityMap.put("I",new Integer(NEEDPRIORITY_INPROGRESS));
+    needPriorityMap.put("F",new Integer(NEEDPRIORITY_FALSE));
+  }
+  
   /** Prerequisite event manager */
   protected PrereqEventManager prereqEventManager;
 
@@ -209,6 +222,7 @@ public class JobQueue extends org.apache
         map.put(processIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false));
         map.put(seedingProcessIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false));
         map.put(needPriorityField,new ColumnDescription("CHAR(1)",false,true,null,null,false));
+        map.put(needPriorityProcessIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false));
         performCreate(map,null);
       }
       else
@@ -343,12 +357,22 @@ public class JobQueue extends org.apache
   public void restart(String processID)
     throws ManifoldCFException
   {
-    // Map ACTIVE back to PENDING.
+    // Map NEEDPRIORITY_INPROCESS back to NEEDPRIORITY_TRUE
     HashMap map = new HashMap();
-    map.put(statusField,statusToString(STATUS_PENDING));
-    map.put(processIDField,null);
     ArrayList list = new ArrayList();
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_TRUE));
+    map.put(needPriorityProcessIDField,null);
     String query = buildConjunctionClause(list,new ClauseDescription[]{
+      new UnitaryClause(needPriorityField,needPriorityToString(NEEDPRIORITY_INPROGRESS)),
+      new UnitaryClause(processIDField,processID)});
+    performUpdate(map,"WHERE "+query,list,null);
+
+    // Map ACTIVE back to PENDING.
+    map.clear();
+    list.clear();
+    map.put(statusField,statusToString(STATUS_PENDING));
+    map.put(processIDField,null);
+    query = buildConjunctionClause(list,new ClauseDescription[]{
       new MultiClause(statusField,new Object[]{
         statusToString(STATUS_ACTIVE),
         statusToString(STATUS_ACTIVENEEDRESCAN)}),
@@ -408,12 +432,21 @@ public class JobQueue extends org.apache
   public void restart()
     throws ManifoldCFException
   {
-    // Map ACTIVE back to PENDING.
+    // Map NEEDPRIORITY_INPROGRESS to NEEDPRIORITY_TRUE
     HashMap map = new HashMap();
-    map.put(statusField,statusToString(STATUS_PENDING));
-    map.put(processIDField,null);
     ArrayList list = new ArrayList();
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_TRUE));
+    map.put(needPriorityProcessIDField,null);
     String query = buildConjunctionClause(list,new ClauseDescription[]{
+      new UnitaryClause(needPriorityField,needPriorityToString(NEEDPRIORITY_INPROGRESS))});
+    performUpdate(map,"WHERE "+query,list,null);
+
+    // Map ACTIVE back to PENDING.
+    map.clear();
+    list.clear();
+    map.put(statusField,statusToString(STATUS_PENDING));
+    map.put(processIDField,null);
+    query = buildConjunctionClause(list,new ClauseDescription[]{
       new MultiClause(statusField,new Object[]{
         statusToString(STATUS_ACTIVE),
         statusToString(STATUS_ACTIVENEEDRESCAN)})});
@@ -733,7 +766,7 @@ public class JobQueue extends org.apache
     // Map COMPLETE to PENDINGPURGATORY
     HashMap map = new HashMap();
     map.put(statusField,statusToString(STATUS_PENDINGPURGATORY));
-    map.put(needPriorityField,needPriorityToString(true));
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_TRUE));
     // Do not reset priorities here!  They should all be blank at this point.
     map.put(checkTimeField,new Long(0L));
     map.put(checkActionField,actionToString(ACTION_RESCAN));
@@ -791,7 +824,7 @@ public class JobQueue extends org.apache
     // Map COMPLETE to PENDINGPURGATORY.
     HashMap map = new HashMap();
     map.put(statusField,statusToString(STATUS_PENDINGPURGATORY));
-    map.put(needPriorityField,needPriorityToString(true));
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_TRUE));
     // Do not reset priorities here!  They should all be blank at this point.
     map.put(checkTimeField,new Long(0L));
     map.put(checkActionField,actionToString(ACTION_RESCAN));
@@ -865,12 +898,27 @@ public class JobQueue extends org.apache
     noteModifications(0,0,1);
   }
 
+  /** Prepare to calculate a document priority for a given row. */
+  public void markNeedPriorityInProgress(Long rowID, String processID)
+    throws ManifoldCFException
+  {
+    HashMap map = new HashMap();
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_INPROGRESS));
+    map.put(needPriorityProcessIDField,processID);
+    ArrayList list = new ArrayList();
+    String query = buildConjunctionClause(list,new ClauseDescription[]{
+      new UnitaryClause(idField,rowID)});
+    performUpdate(map,"WHERE "+query,list,null);
+    noteModifications(0,1,0);
+  }
+  
   /** Write out a document priority */
   public void writeDocPriority(Long rowID, IPriorityCalculator priority)
     throws ManifoldCFException
   {
     HashMap map = new HashMap();
-    map.put(needPriorityField,needPriorityToString(false));
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
+    map.put(needPriorityProcessIDField,null);
     map.put(docPriorityField,new Double(priority.getDocumentPriority()));
     ArrayList list = new ArrayList();
     String query = buildConjunctionClause(list,new ClauseDescription[]{
@@ -884,7 +932,8 @@ public class JobQueue extends org.apache
     throws ManifoldCFException
   {
     HashMap map = new HashMap();
-    map.put(needPriorityField,needPriorityToString(false));
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
+    map.put(needPriorityProcessIDField,null);
     map.put(docPriorityField,nullDocPriority);
     ArrayList list = new ArrayList();
     String query = buildConjunctionClause(list,new ClauseDescription[]{
@@ -900,7 +949,7 @@ public class JobQueue extends org.apache
     throws ManifoldCFException
   {
     HashMap map = new HashMap();
-    map.put(needPriorityField,needPriorityToString(true));
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_TRUE));
     map.put(docPriorityField,nullDocPriority);
     ArrayList list = new ArrayList();
     String query = buildConjunctionClause(list,new ClauseDescription[]{
@@ -929,7 +978,7 @@ public class JobQueue extends org.apache
       checkTimeValue = null;
       // Remove document priority; we don't want to pollute the queue.  See CONNECTORS-290.
       map.put(docPriorityField,nullDocPriority);
-      map.put(needPriorityField,needPriorityToString(false));
+      map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
       break;
     case STATUS_ACTIVENEEDRESCAN:
     case STATUS_ACTIVENEEDRESCANPURGATORY:
@@ -1241,7 +1290,7 @@ public class JobQueue extends org.apache
       map.put(failCountField,null);
       // Update the doc priority.
       map.put(docPriorityField,new Double(desiredPriority.getDocumentPriority()));
-      map.put(needPriorityField,needPriorityToString(false));
+      map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
       break;
 
     case STATUS_PENDING:
@@ -1318,7 +1367,7 @@ public class JobQueue extends org.apache
     map.put(seedingProcessIDField,processID);
     // Set the document priority
     map.put(docPriorityField,new Double(desiredDocPriority.getDocumentPriority()));
-    map.put(needPriorityField,needPriorityToString(false));
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
     performInsert(map,null);
     prereqEventManager.addRows(recordID,prereqEvents);
     noteModifications(1,0,0);
@@ -1517,7 +1566,7 @@ public class JobQueue extends org.apache
       map.put(failCountField,null);
       // Going into pending: set the docpriority.
       map.put(docPriorityField,new Double(desiredPriority.getDocumentPriority()));
-      map.put(needPriorityField,needPriorityToString(false));
+      map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
       break;
     case STATUS_COMPLETE:
     case STATUS_BEINGCLEANED:
@@ -1534,7 +1583,7 @@ public class JobQueue extends org.apache
         map.put(failCountField,null);
         // Going into pending: set the docpriority.
         map.put(docPriorityField,new Double(desiredPriority.getDocumentPriority()));
-        map.put(needPriorityField,needPriorityToString(false));
+        map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
         break;
       }
       return;
@@ -1563,7 +1612,7 @@ public class JobQueue extends org.apache
         map.put(failCountField,null);
         // Going into pending: set the docpriority.
         map.put(docPriorityField,new Double(desiredPriority.getDocumentPriority()));
-        map.put(needPriorityField,needPriorityToString(false));
+        map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
         break;
       }
       return;
@@ -1587,7 +1636,7 @@ public class JobQueue extends org.apache
         map.put(failCountField,null);
         // Going into pending: set the docpriority.
         map.put(docPriorityField,new Double(desiredPriority.getDocumentPriority()));
-        map.put(needPriorityField,needPriorityToString(false));
+        map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
         break;
       }
       return;
@@ -1643,7 +1692,7 @@ public class JobQueue extends org.apache
     map.put(statusField,statusToString(STATUS_PENDING));
     // Be sure to set the priority also
     map.put(docPriorityField,new Double(desiredDocPriority.getDocumentPriority()));
-    map.put(needPriorityField,needPriorityToString(false));
+    map.put(needPriorityField,needPriorityToString(NEEDPRIORITY_FALSE));
     performInsert(map,null);
     prereqEventManager.addRows(recordID,prereqEvents);
     noteModifications(1,0,0);
@@ -1678,7 +1727,7 @@ public class JobQueue extends org.apache
   {
     if (x == null || x.length() == 0)
       return SEEDSTATUS_NOTSEED;
-    Integer y = (Integer)seedstatusMap.get(x);
+    Integer y = seedstatusMap.get(x);
     if (y == null)
       throw new ManifoldCFException("Unknown seed status code: "+x);
     return y.intValue();
@@ -1689,7 +1738,7 @@ public class JobQueue extends org.apache
   public static int stringToAction(String value)
     throws ManifoldCFException
   {
-    Integer x = (Integer)actionMap.get(value);
+    Integer x = actionMap.get(value);
     if (x == null)
       throw new ManifoldCFException("Unknown action string: '"+value+"'");
     return x.intValue();
@@ -1712,22 +1761,33 @@ public class JobQueue extends org.apache
 
   /** Convert need priority value to boolean.
   */
-  public static boolean stringToNeedPriority(String value)
+  public static int stringToNeedPriority(String value)
     throws ManifoldCFException
   {
-    if (value != null && value.equals("T"))
-      return true;
-    return false;
+    if (value == null || value.length() == 0)
+      return NEEDPRIORITY_FALSE;
+    Integer x = needPriorityMap.get(value);
+    if (x == null)
+      throw new ManifoldCFException("Unknown needpriority string: '"+value+"'");
+    return x.intValue();
   }
   
   /** Convert boolean to need priority value.
   */
-  public static String needPriorityToString(boolean value)
+  public static String needPriorityToString(int value)
     throws ManifoldCFException
   {
-    if (value)
+    switch (value)
+    {
+    case NEEDPRIORITY_TRUE:
       return "T";
-    return "F";
+    case NEEDPRIORITY_FALSE:
+      return "F";
+    case NEEDPRIORITY_INPROGRESS:
+      return "I";
+    default:
+      throw new ManifoldCFException("Bad needpriority value: "+Integer.toString(value));
+    }
   }
   
   /** Convert status field value to integer.
@@ -1737,7 +1797,7 @@ public class JobQueue extends org.apache
   public static int stringToStatus(String value)
     throws ManifoldCFException
   {
-    Integer x = (Integer)statusMap.get(value);
+    Integer x = statusMap.get(value);
     if (x == null)
       throw new ManifoldCFException("Unknown status string: '"+value+"'");
     return x.intValue();

Modified: manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java?rev=1638083&r1=1638082&r2=1638083&view=diff
==============================================================================
--- manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java
(original)
+++ manifoldcf/branches/CONNECTORS-1100/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java
Tue Nov 11 13:40:55 2014
@@ -135,7 +135,7 @@ public class SetPriorityThread extends T
             // We may well wind up calculating priority for documents that wind up having
their
             // state changed before we can write back, but this is okay because update is
only
             // going to be permitted for rows that still have the right state.
-            DocumentDescription[] descs = jobManager.getNextNotYetProcessedReprioritizationDocuments(1000);
+            DocumentDescription[] descs = jobManager.getNextNotYetProcessedReprioritizationDocuments(processID,1000);
             if (descs.length > 0)
             {
               ManifoldCF.writeDocumentPriorities(threadContext,



Mime
View raw message