<commit_before><?php
/**
 * Base class for all job handler workers.
 * 
 * @package Scheduler
 */
abstract class KJobHandlerWorker extends KBatchBase
{
	/**
	 * @param KalturaBatchJob $job
	 * @return KalturaBatchJob
	 */
	abstract protected function exec(KalturaBatchJob $job);
	
	protected function init()
	{
		$this->saveQueueFilter($this->getJobType());
	}
	
	protected function getMaxJobsEachRun()
	{
		if(!$this->taskConfig->maxJobsEachRun)
			return 1;
		
		return $this->taskConfig->maxJobsEachRun;
	}
	
	protected function getJobs()
	{
		return $this->kClient->batch->getExclusiveJobs($this->getExclusiveLockKey(), $this->taskConfig->maximumExecutionTime, $this->getMaxJobsEachRun(), $this->getFilter(), $this->getJobType());
	}
	
	public function run($jobs = null)
	{
		if($this->taskConfig->isInitOnly())
			return $this->init();
		
		if(is_null($jobs))
			$jobs = $this->getJobs();
		
		KalturaLog::info(count($jobs) . " jobs to handle");
		
		if(! count($jobs) > 0)
		{
			KalturaLog::info("Queue size: 0 sent to scheduler");
			$this->saveSchedulerQueue($this->getJobType());
			return null;
		}
		
		foreach($jobs as &$job)
		{
			try
			{
				$job = $this->exec($job);
			}
			catch(KalturaException $kex)
			{
				$this->unimpersonate();
				$job = $this->closeJob($job, KalturaBatchJobErrorTypes::KALTURA_API, $kex->getCode(), "Error: " . $kex->getMessage(), KalturaBatchJobStatus::FAILED);
			}
			catch(kTemporaryException $ktex)
			{
				$this->unimpersonate();
				$job = $this->closeJob($job, KalturaBatchJobErrorTypes::RUNTIME, $ktex->getCode(), "Error: " . $ktex->getMessage(), KalturaBatchJobStatus::RETRY);
			}
			catch(KalturaClientException $kcex)
			{
				$this->unimpersonate();
				$job = $this->closeJob($job, KalturaBatchJobErrorTypes::KALTURA_CLIENT, $kcex->getCode(), "Error: " . $kcex->getMessage(), KalturaBatchJobStatus::RETRY);<commit_msg>qnd: don't crash if the getExclusive request crashed.<commit_after><?php
/**
 * Base class for all job handler workers.
 * 
 * @package Scheduler
 */
abstract class KJobHandlerWorker extends KBatchBase
{
	/**
	 * @param KalturaBatchJob $job
	 * @return KalturaBatchJob
	 */
	abstract protected function exec(KalturaBatchJob $job);
	
	protected function init()
	{
		$this->saveQueueFilter($this->getJobType());
	}
	
	protected function getMaxJobsEachRun()
	{
		if(!$this->taskConfig->maxJobsEachRun)
			return 1;
		
		return $this->taskConfig->maxJobsEachRun;
	}
	
	protected function getJobs()
	{
		return $this->kClient->batch->getExclusiveJobs($this->getExclusiveLockKey(), $this->taskConfig->maximumExecutionTime, $this->getMaxJobsEachRun(), $this->getFilter(), $this->getJobType());
	}
	
	public function run($jobs = null)
	{
		if($this->taskConfig->isInitOnly())
			return $this->init();
		
		if(is_null($jobs))
		{
			try
			{
				$jobs = $this->getJobs();
			}
			catch (Exception $e)
			{
				KalturaLog::err($e->getMessage());
				return null;
			}
		}
		
		KalturaLog::info(count($jobs) . " jobs to handle");
		
		if(! count($jobs) > 0)
		{
			KalturaLog::info("Queue size: 0 sent to scheduler");
			$this->saveSchedulerQueue($this->getJobType());
			return null;
		}
		
		foreach($jobs as &$job)
		{
			try
			{
				$job = $this->exec($job);
			}
			catch(KalturaException $kex)
			{
				$this->unimpersonate();
				$job = $this->closeJob($job, KalturaBatchJobErrorTypes::KALTURA_API, $kex->getCode(), "Error: " . $kex->getMessage(), KalturaBatchJobStatus::FAILED);
			}
			catch(kTemporaryException $ktex)
			{
				$this->unimpersonate();
				$job = $this->closeJob($job, KalturaBatchJobErrorTypes::RUNTIME, $ktex->getCode(), "Error: " . $ktex->getMessage(), KalturaBatchJobStatus::RETRY);
			}
			catch(KalturaClientException $kcex)
			{
				$this->unimpersonate();
				$job = $this->closeJob($job, KalturaBatchJobErrorTypes::KALTURA_CLIENT, $kcex->getCode(), "Error: " . $kcex->getMessage(), KalturaBatchJobStatus::RETRY);
{'fraction_non_alphanumeric': 0.17032686414708886, 'fraction_numerical': 0.0015321756894790602, 'mean_word_length': 4.3075880758807585, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 74, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4216644', 'n_tokens_mistral': 1592, 'n_tokens_neox': 1362, 'n_words': 229}