Allow setup custom exception handlers for failing jobs (#742)

* Allow setup custom exception handlers for failing jobs

* Allow setup custom exception handlers for failing jobs

* Fix specs

---------

Co-authored-by: Pablo Cantero <pablohstc@gmail.com>

lib/shoryuken.rb

@@ -23,6 +23,7 @@
 require 'shoryuken/worker/inline_executor'
 require 'shoryuken/worker_registry'
 require 'shoryuken/default_worker_registry'
+require 'shoryuken/default_exception_handler'
 require 'shoryuken/middleware/chain'
 require 'shoryuken/middleware/server/auto_delete'
 Shoryuken::Middleware::Server.autoload :AutoExtendVisibility, 'shoryuken/middleware/server/auto_extend_visibility'
@@ -73,6 +74,8 @@ def self.healthy?
     :sqs_client=,
     :sqs_client_receive_message_opts,
     :sqs_client_receive_message_opts=,
+    :exception_handlers,
+    :exception_handlers=,
     :options,
     :logger,
     :register_worker,

lib/shoryuken/default_exception_handler.rb

@@ -0,0 +1,10 @@
+module Shoryuken
+  class DefaultExceptionHandler
+    extend Util
+
+    def self.call(exception, _queue, _sqs_msg)
+      logger.error { "Processor failed: #{exception.message}" }
+      logger.error { exception.backtrace.join("\n") } if exception.backtrace
+    end
+  end
+end

lib/shoryuken/options.rb

@@ -18,13 +18,14 @@ class Options
 
     attr_accessor :active_job_queue_name_prefixing, :cache_visibility_timeout, :groups,
                   :launcher_executor,
-                  :start_callback, :stop_callback, :worker_executor, :worker_registry
+                  :start_callback, :stop_callback, :worker_executor, :worker_registry, :exception_handlers
     attr_writer :default_worker_options, :sqs_client
     attr_reader :sqs_client_receive_message_opts
 
     def initialize
       self.groups = {}
       self.worker_registry = DefaultWorkerRegistry.new
+      self.exception_handlers = [DefaultExceptionHandler]
       self.active_job_queue_name_prefixing = false
       self.worker_executor = Worker::DefaultExecutor
       self.cache_visibility_timeout = false

lib/shoryuken/processor.rb

@@ -22,8 +22,7 @@ def process
         end
       end
     rescue Exception => ex
-      logger.error { "Processor failed: #{ex.message}" }
-      logger.error { ex.backtrace.join("\n") } unless ex.backtrace.nil?
+      Array(Shoryuken.exception_handlers).each { |handler| handler.call(ex, queue, sqs_msg) }
 
       raise
     end

spec/shoryuken/default_exception_handler_spec.rb

@@ -0,0 +1,71 @@
+require 'spec_helper'
+
+# rubocop:disable Metrics/BlockLength
+RSpec.describe Shoryuken::DefaultExceptionHandler do
+  class CustomErrorHandler
+    extend Shoryuken::Util
+
+    def self.call(_ex, queue, _msg)
+      logger.error("#{queue.to_s} failed to process the message")
+    end
+  end
+
+  before do
+    Shoryuken.worker_executor = Shoryuken::Worker::InlineExecutor
+    allow(manager).to receive(:async).and_return(manager)
+    allow(manager).to receive(:real_thread)
+    allow(Shoryuken::Client).to receive(:queues).with(queue).and_return(sqs_queue)
+  end
+
+  after do
+    Shoryuken.worker_executor = Shoryuken::Worker::DefaultExecutor
+  end
+
+  let(:manager) { double Shoryuken::Manager }
+  let(:sqs_queue) { double Shoryuken::Queue, visibility_timeout: 30 }
+  let(:queue) { 'default' }
+
+  let(:sqs_msg) do
+    double(
+      Shoryuken::Message,
+      queue_url: queue,
+      body: 'test',
+      message_attributes: {},
+      message_id: SecureRandom.uuid,
+      receipt_handle: SecureRandom.uuid
+    )
+  end
+
+  subject { Shoryuken::Processor.new(queue, sqs_msg) }
+
+  context "with default handler" do
+    before do
+      Shoryuken.exception_handlers = described_class
+    end
+
+    it "logs an error message" do
+      expect(Shoryuken::Logging.logger).to receive(:error).twice
+
+      allow_any_instance_of(TestWorker).to receive(:perform).and_raise(StandardError, "error")
+      allow(sqs_msg).to receive(:body)
+
+      expect { subject.process }.to raise_error(StandardError)
+    end
+  end
+
+  context "with custom handler" do
+    before do
+      Shoryuken.exception_handlers = [described_class, CustomErrorHandler]
+    end
+
+    it "logs default and custom error messages" do
+      expect(Shoryuken::Logging.logger).to receive(:error).twice
+      expect(Shoryuken::Logging.logger).to receive(:error).with("default failed to process the message").once
+
+      allow_any_instance_of(TestWorker).to receive(:perform).and_raise(StandardError, "error")
+      allow(sqs_msg).to receive(:body)
+
+      expect { subject.process }.to raise_error(StandardError)
+    end
+  end
+end

spec/spec_helper.rb

@@ -46,6 +46,8 @@ def perform(sqs_msg, body); end
     Shoryuken.options[:logfile]     = nil
     Shoryuken.options[:queues]      = nil
 
+    Shoryuken.options[:exception_handlers] = []
+
     TestWorker.get_shoryuken_options.clear
     TestWorker.get_shoryuken_options['queue'] = 'default'