Issue #2021933 by Crell, lz1irq: Catch exceptions from queue workers.

core/includes/common.inc

@@ -3247,8 +3247,15 @@ function drupal_cron_run() {
       $end = time() + (isset($info['cron']['time']) ? $info['cron']['time'] : 15);
       $queue = \Drupal::queue($queue_name);
       while (time() < $end && ($item = $queue->claimItem())) {
-        call_user_func_array($callback, array($item->data));
-        $queue->deleteItem($item);
+        try {
+          call_user_func_array($callback, array($item->data));
+          $queue->deleteItem($item);
+        }
+        catch (Exception $e) {
+          // In case of exception log it and leave the item in the queue
+          // to be processed again later.
+          watchdog_exception('cron', $e);
+        }
       }
     }
   }

core/modules/system/lib/Drupal/system/Tests/System/CronQueueTest.php

@@ -0,0 +1,53 @@
+<?php
+
+/**
+ * @file
+ * Contains Drupal\system\Tests\System\CronQueueTest.
+ */
+
+namespace Drupal\system\Tests\System;
+
+use Drupal\Core\Database\Database;
+use Drupal\Core\Queue\DatabaseQueue;
+use Drupal\Core\Queue\Memory;
+use Drupal\simpletest\DrupalUnitTestBase;
+use Drupal\simpletest\WebTestBase;
+
+/**
+ * Tests the handling of exceptions thrown by queue workers.
+ */
+class CronQueueTest extends WebTestBase {
+
+  /**
+   * The modules to enable.
+   *
+   * @var array
+   */
+  public static $modules = array('cron_queue_test');
+
+  public static function getInfo() {
+    return array(
+      'name' => 'Cron Queue functionality',
+      'description' => 'Tests the Cron Queue runner.',
+      'group' => 'Queue',
+    );
+  }
+
+  /**
+   * Tests that exceptions thrown by workers are handled properly.
+   */
+  public function testExceptions() {
+
+    $queue = $this->container->get('queue')->get('cron_queue_test_exception');
+
+    // Enqueue an item for processing.
+    $queue->createItem(array($this->randomName() => $this->randomName()));
+
+    // Run cron; the worker for this queue should throw an exception and handle
+    // it.
+    $this->cronRun();
+
+    // The item should be left in the queue.
+    $this->assertEqual($queue->numberOfItems(), 1, 'Failing item still in the queue after throwing an exception.');
+  }
+}

core/modules/system/tests/modules/cron_queue_test/cron_queue_test.info.yml

@@ -0,0 +1,7 @@
+name: 'Cron Queue test'
+type: module
+description: 'Support module for the cron queue runner.'
+package: Testing
+version: VERSION
+core: 8.x
+hidden: true

core/modules/system/tests/modules/cron_queue_test/cron_queue_test.module

@@ -0,0 +1,17 @@
+<?php
+
+function cron_queue_test_queue_info() {
+  $queues['cron_queue_test_exception'] = array(
+    'title' => t('Exception test'),
+    'worker callback' => 'cron_queue_test_exception',
+    // Only needed if this queue should be processed by cron.
+    'cron' => array(
+      'time' => 60,
+    ),
+  );
+  return $queues;
+}
+
+function cron_queue_test_exception($item) {
+  throw new Exception('That is not supposed to happen.');
+}