<commit_before>/* $Id: SeedingThread.java 988245 2010-08-23 18:39:35Z kwright $ */

/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.manifoldcf.crawler.system;

import org.apache.manifoldcf.core.interfaces.*;
import org.apache.manifoldcf.agents.interfaces.*;
import org.apache.manifoldcf.crawler.interfaces.*;
import org.apache.manifoldcf.crawler.system.Logging;
import java.util.*;
import java.lang.reflect.*;

/** This class represents the background seeding thread.  Its job is to add
* seeded documents from the connector periodically, during adaptive crawls
* (which continue until stopped).  The actual use case is for creating a
* connector that handles RSS feeds, including keeping them current and
* handling deletions.
*/
public class SeedingThread extends Thread
{
  public static final String _rcsid = "@(#)$Id: SeedingThread.java 988245 2010-08-23 18:39:35Z kwright $";

  /** Worker thread pool reset manager */
  protected static SeedingResetManager resetManager = new SeedingResetManager();

  // Local data
  protected QueueTracker queueTracker;

  /** The number of documents that are added to the queue per transaction */
  protected final static int MAX_COUNT = 100;

  /** Constructor.
  */
  public SeedingThread(QueueTracker queueTracker)
    throws ManifoldCFException
  {
    super();
    setName("Seeding thread");
    setDaemon(true);
    this.queueTracker = queueTracker;
  }

  public void run()
  {
    resetManager.registerMe();

    try
    {
      // Create a thread context object.
      IThreadContext threadContext = ThreadContextFactory.make();
      IJobManager jobManager = JobManagerFactory.make(threadContext);
      IRepositoryConnectionManager connectionMgr = RepositoryConnectionManagerFactory.make(threadContext);

      IDBInterface database = DBInterfaceFactory.make(threadContext,
        ManifoldCF.getMasterDatabaseName(),
        ManifoldCF.getMasterDatabaseUsername(),
        ManifoldCF.getMasterDatabasePassword());

      String[] identifiers = new String[MAX_COUNT];
      // Loop
      while (true)
      {
        // Do another try/catch around everything in the loop
        try
        {
          // Before we begin, conditionally reset
          resetManager.waitForReset(threadContext);

          long currentTime = System.currentTimeMillis();

          // Accumulate the wait before doing the next check.
          // We start with 10 seconds, which is the maximum.  If there's a service request
          // that's faster than that, we'll adjust the time downward.
          long waitTime = 60000L;

          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);

          // 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
          // to the next job, since the whole thing will retry anyhow.
          try
          {

            if (seedJobs.length == 0)
            {
              Logging.threads.debug("Seeding thread found nothing to do");
              ManifoldCF.sleep(waitTime);
              continue;
            }

            if (Logging.threads.isDebugEnabled())
              Logging.threads.debug("Seeding thread: Found "+Integer.toString(seedJobs.length)+" jobs to seed");

            // Loop through jobs
            int i = 0;
            while (i < seedJobs.length)
            {
              JobSeedingRecord jsr = seedJobs[i++];
              Long jobID = jsr.getJobID();
              try
              {
                long lastJobTime = jsr.getSynchTime();
                IJobDescription jobDescription = jobManager.load(jobID,true);
                int jobType = jobDescription.getType();

                int hopcountMethod = jobDescription.getHopcountMode();

                IRepositoryConnection connection = connectionMgr.load(jobDescription.getConnectionName());
                IRepositoryConnector connector = RepositoryConnectorFactory.grab(threadContext,
                  connection.getClassName(),
                  connection.getConfigParams(),
                  connection.getMaxConnections());
                // Null will come back if the connector instance could not be obtained, so just skip in that case.
                if (connector == null)
                  continue;
                try
                {


                  // Get the number of link types.
                  String[] legalLinkTypes = connector.getRelationshipTypes();

                  int model = connector.getConnectorModel();

                  try
                  {

                    SeedingActivity activity = new SeedingActivity(connection.getName(),connectionMgr,jobManager,queueTracker,
                      connection,connector,jobID,legalLinkTypes,false,hopcountMethod);

                    if (Logging.threads.isDebugEnabled())
                      Logging.threads.debug("Seeding thread: Getting seeds for job "+jobID.toString());

                    connector.addSeedDocuments(activity,jobDescription.getSpecification(),lastJobTime,currentTime,jobType);

                    activity.doneSeeding(model==connector.MODEL_PARTIAL);

                    if (Logging.threads.isDebugEnabled())
                      Logging.threads.debug("Seeding thread: Done processing seeds from job "+jobID.toString());


                  }
                  catch (ServiceInterruption e)
                  {
                    // Note the service interruption
                    Logging.threads.error("Service interruption for job "+jobID,e);
                    long retryInterval = e.getRetryTime() - currentTime;
                    if (retryInterval >= 0L && retryInterval < waitTime)
                      waitTime = retryInterval;
                    // Go on to the next job
                    continue;
                  }
                }
                finally
                {
                  RepositoryConnectorFactory.release(connector);
                }


                if (Logging.threads.isDebugEnabled())
                  Logging.threads.debug("Seeding thread: Successfully reseeded job "+jobID.toString());

                // Note that this job has been seeded!
                jobManager.noteJobSeeded(jobID,currentTime);
                jsr.noteStarted();

              }
              catch (ManifoldCFException e)
              {
                if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)<commit_msg>Service interruption while seeding should not be logged as an ERROR.<commit_after>/* $Id: SeedingThread.java 988245 2010-08-23 18:39:35Z kwright $ */

/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.manifoldcf.crawler.system;

import org.apache.manifoldcf.core.interfaces.*;
import org.apache.manifoldcf.agents.interfaces.*;
import org.apache.manifoldcf.crawler.interfaces.*;
import org.apache.manifoldcf.crawler.system.Logging;
import java.util.*;
import java.lang.reflect.*;

/** This class represents the background seeding thread.  Its job is to add
* seeded documents from the connector periodically, during adaptive crawls
* (which continue until stopped).  The actual use case is for creating a
* connector that handles RSS feeds, including keeping them current and
* handling deletions.
*/
public class SeedingThread extends Thread
{
  public static final String _rcsid = "@(#)$Id: SeedingThread.java 988245 2010-08-23 18:39:35Z kwright $";

  /** Worker thread pool reset manager */
  protected static SeedingResetManager resetManager = new SeedingResetManager();

  // Local data
  protected QueueTracker queueTracker;

  /** The number of documents that are added to the queue per transaction */
  protected final static int MAX_COUNT = 100;

  /** Constructor.
  */
  public SeedingThread(QueueTracker queueTracker)
    throws ManifoldCFException
  {
    super();
    setName("Seeding thread");
    setDaemon(true);
    this.queueTracker = queueTracker;
  }

  public void run()
  {
    resetManager.registerMe();

    try
    {
      // Create a thread context object.
      IThreadContext threadContext = ThreadContextFactory.make();
      IJobManager jobManager = JobManagerFactory.make(threadContext);
      IRepositoryConnectionManager connectionMgr = RepositoryConnectionManagerFactory.make(threadContext);

      IDBInterface database = DBInterfaceFactory.make(threadContext,
        ManifoldCF.getMasterDatabaseName(),
        ManifoldCF.getMasterDatabaseUsername(),
        ManifoldCF.getMasterDatabasePassword());

      String[] identifiers = new String[MAX_COUNT];
      // Loop
      while (true)
      {
        // Do another try/catch around everything in the loop
        try
        {
          // Before we begin, conditionally reset
          resetManager.waitForReset(threadContext);

          long currentTime = System.currentTimeMillis();

          // Accumulate the wait before doing the next check.
          // We start with 10 seconds, which is the maximum.  If there's a service request
          // that's faster than that, we'll adjust the time downward.
          long waitTime = 60000L;

          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);

          // 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
          // to the next job, since the whole thing will retry anyhow.
          try
          {

            if (seedJobs.length == 0)
            {
              Logging.threads.debug("Seeding thread found nothing to do");
              ManifoldCF.sleep(waitTime);
              continue;
            }

            if (Logging.threads.isDebugEnabled())
              Logging.threads.debug("Seeding thread: Found "+Integer.toString(seedJobs.length)+" jobs to seed");

            // Loop through jobs
            int i = 0;
            while (i < seedJobs.length)
            {
              JobSeedingRecord jsr = seedJobs[i++];
              Long jobID = jsr.getJobID();
              try
              {
                long lastJobTime = jsr.getSynchTime();
                IJobDescription jobDescription = jobManager.load(jobID,true);
                int jobType = jobDescription.getType();

                int hopcountMethod = jobDescription.getHopcountMode();

                IRepositoryConnection connection = connectionMgr.load(jobDescription.getConnectionName());
                IRepositoryConnector connector = RepositoryConnectorFactory.grab(threadContext,
                  connection.getClassName(),
                  connection.getConfigParams(),
                  connection.getMaxConnections());
                // Null will come back if the connector instance could not be obtained, so just skip in that case.
                if (connector == null)
                  continue;
                try
                {


                  // Get the number of link types.
                  String[] legalLinkTypes = connector.getRelationshipTypes();

                  int model = connector.getConnectorModel();

                  try
                  {

                    SeedingActivity activity = new SeedingActivity(connection.getName(),connectionMgr,jobManager,queueTracker,
                      connection,connector,jobID,legalLinkTypes,false,hopcountMethod);

                    if (Logging.threads.isDebugEnabled())
                      Logging.threads.debug("Seeding thread: Getting seeds for job "+jobID.toString());

                    connector.addSeedDocuments(activity,jobDescription.getSpecification(),lastJobTime,currentTime,jobType);

                    activity.doneSeeding(model==connector.MODEL_PARTIAL);

                    if (Logging.threads.isDebugEnabled())
                      Logging.threads.debug("Seeding thread: Done processing seeds from job "+jobID.toString());


                  }
                  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;
                    // Go on to the next job
                    continue;
                  }
                }
                finally
                {
                  RepositoryConnectorFactory.release(connector);
                }


                if (Logging.threads.isDebugEnabled())
                  Logging.threads.debug("Seeding thread: Successfully reseeded job "+jobID.toString());

                // Note that this job has been seeded!
                jobManager.noteJobSeeded(jobID,currentTime);
                jsr.noteStarted();

              }
              catch (ManifoldCFException e)
              {
                if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)
{'fraction_non_alphanumeric': 0.07878094725511302, 'fraction_numerical': 0.007669537136706136, 'mean_word_length': 2.1830835117773018, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4828299', 'n_tokens_mistral': 3789, 'n_tokens_neox': 3460, 'n_words': 1133}