diff --git a/django_celery_beat/schedulers.py b/django_celery_beat/schedulers.py
index 846b97a9..a73faff3 100644
--- a/django_celery_beat/schedulers.py
+++ b/django_celery_beat/schedulers.py
@@ -110,7 +110,6 @@ def is_due(self):
             now = self._default_now()
             if getattr(settings, 'DJANGO_CELERY_BEAT_TZ_AWARE', True):
                 now = maybe_make_aware(self._default_now())
-
             if now < self.model.start_time:
                 # The datetime is before the start date - don't run.
                 # send a delay to retry on start_time
@@ -119,6 +118,14 @@ def is_due(self):
                 )
                 return schedules.schedstate(False, delay)
 
+        # EXPIRED TASK: Disable task when expired
+        if self.model.expires is not None:
+            now = self._default_now()
+            if now >= self.model.expires:
+                self._disable(self.model)
+                # Don't recheck
+                return schedules.schedstate(False, NEVER_CHECK_TIMEOUT)
+
         # ONE OFF TASK: Disable one off tasks after they've ran once
         if self.model.one_off and self.model.enabled \
                 and self.model.total_run_count > 0:
diff --git a/t/unit/test_schedulers.py b/t/unit/test_schedulers.py
index d070bb45..f7889c77 100644
--- a/t/unit/test_schedulers.py
+++ b/t/unit/test_schedulers.py
@@ -291,6 +291,35 @@ def test_one_off_task(self):
         assert not isdue
         assert delay == NEVER_CHECK_TIMEOUT
 
+    def test_task_with_expires(self):
+        interval = 10
+        right_now = self.app.now()
+        one_second_later = right_now + timedelta(seconds=1)
+        m = self.create_model_interval(schedule(timedelta(seconds=interval)),
+                                       start_time=right_now,
+                                       expires=one_second_later)
+        e = self.Entry(m, app=self.app)
+        isdue, delay = e.is_due()
+        assert isdue
+        assert delay == interval
+
+        m2 = self.create_model_interval(schedule(timedelta(seconds=interval)),
+                                        start_time=right_now,
+                                        expires=right_now)
+        e2 = self.Entry(m2, app=self.app)
+        isdue, delay = e2.is_due()
+        assert not isdue
+        assert delay == NEVER_CHECK_TIMEOUT
+
+        one_second_ago = right_now - timedelta(seconds=1)
+        m2 = self.create_model_interval(schedule(timedelta(seconds=interval)),
+                                        start_time=right_now,
+                                        expires=one_second_ago)
+        e2 = self.Entry(m2, app=self.app)
+        isdue, delay = e2.is_due()
+        assert not isdue
+        assert delay == NEVER_CHECK_TIMEOUT
+
 
 @pytest.mark.django_db
 class test_DatabaseSchedulerFromAppConf(SchedulerCase):