diff --git a/lib/delayed/backend/base.rb b/lib/delayed/backend/base.rb
index 56fe47fa5..ea27bbc06 100644
--- a/lib/delayed/backend/base.rb
+++ b/lib/delayed/backend/base.rb
@@ -124,6 +124,10 @@ def max_attempts
         payload_object.max_attempts if payload_object.respond_to?(:max_attempts)
       end
 
+      def max_run_time
+        payload_object.max_run_time if payload_object.respond_to?(:max_run_time)
+      end
+
       def fail!
         update_attributes(:failed_at => self.class.db_time_now)
       end
diff --git a/lib/delayed/backend/shared_spec.rb b/lib/delayed/backend/shared_spec.rb
index 3fb2691bf..9a15384b0 100644
--- a/lib/delayed/backend/shared_spec.rb
+++ b/lib/delayed/backend/shared_spec.rb
@@ -398,12 +398,34 @@ def create_job(opts = {})
       expect(@job.max_attempts).to be_nil
     end
 
-    it 'uses the max_retries value on the payload when defined' do
+    it 'uses the max_attempts value on the payload when defined' do
       expect(@job.payload_object).to receive(:max_attempts).and_return(99)
       expect(@job.max_attempts).to eq(99)
     end
   end
 
+  describe '#max_run_time' do
+    before(:each) { @job = described_class.enqueue SimpleJob.new }
+
+    it 'is not defined' do
+      expect(@job.max_run_time).to be_nil
+    end
+
+    it 'results in a default run time when not defined' do
+      expect(worker.max_run_time(@job)).to eq(Delayed::Worker::DEFAULT_MAX_RUN_TIME)
+    end
+
+    it 'uses the max_run_time value on the payload when defined' do
+      @job.payload_object.stub(:max_run_time).and_return(30.minutes)
+      expect(@job.max_run_time).to eq(30.minutes)
+    end
+
+    it 'results in an overridden run time when defined' do
+      @job.payload_object.stub(:max_run_time).and_return(45.minutes)
+      expect(worker.max_run_time(@job)).to eq(45.minutes)
+    end
+  end
+
   describe 'yaml serialization' do
     context 'when serializing jobs' do
       it 'raises error ArgumentError for new records' do
diff --git a/lib/delayed/worker.rb b/lib/delayed/worker.rb
index fbdd92ac4..23db81ae6 100644
--- a/lib/delayed/worker.rb
+++ b/lib/delayed/worker.rb
@@ -202,7 +202,7 @@ def work_off(num = 100)
     def run(job)
       job_say job, 'RUNNING'
       runtime =  Benchmark.realtime do
-        Timeout.timeout(self.class.max_run_time.to_i, WorkerTimeout) { job.invoke_job }
+        Timeout.timeout(max_run_time(job).to_i, WorkerTimeout) { job.invoke_job }
         job.destroy
       end
       job_say job, format('COMPLETED after %.4f', runtime)
@@ -262,6 +262,10 @@ def max_attempts(job)
       job.max_attempts || self.class.max_attempts
     end
 
+    def max_run_time(job)
+      job.max_run_time || self.class.max_run_time
+    end
+
   protected
 
     def handle_failed_job(job, error)