diff --git a/docs/topics/signals.rst b/docs/topics/signals.rst
index cf1588df8..d40c0e1df 100644
--- a/docs/topics/signals.rst
+++ b/docs/topics/signals.rst
@@ -135,6 +135,29 @@ item_dropped
         to be dropped
     :type exception: :exc:`~scrapy.exceptions.DropItem` exception
 
+item_error
+------------
+
+.. signal:: item_error
+.. function:: item_error(item, response, spider, failure)
+
+    Sent when a :ref:`topics-item-pipeline` generates an error (ie. raises
+    an exception), except :exc:`~scrapy.exceptions.DropItem` exception.
+
+    This signal supports returning deferreds from their handlers.
+
+    :param item: the item dropped from the :ref:`topics-item-pipeline`
+    :type item: dict or :class:`~scrapy.item.Item` object
+
+    :param response: the response being processed when the exception was raised
+    :type response: :class:`~scrapy.http.Response` object
+
+    :param spider: the spider which raised the exception
+    :type spider: :class:`~scrapy.spiders.Spider` object
+
+    :param failure: the exception raised as a Twisted `Failure`_ object
+    :type failure: `Failure`_ object
+
 spider_closed
 -------------
 
diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py
index c08e37367..ee1e95a0c 100644
--- a/scrapy/core/scraper.py
+++ b/scrapy/core/scraper.py
@@ -232,6 +232,9 @@ class Scraper(object):
                 logger.error('Error processing %(item)s', {'item': item},
                              exc_info=failure_to_exc_info(output),
                              extra={'spider': spider})
+                return self.signals.send_catch_log_deferred(
+                    signal=signals.item_error, item=item, response=response,
+                    spider=spider, failure=output)
         else:
             logkws = self.logformatter.scraped(output, response, spider)
             logger.log(*logformatter_adapter(logkws), extra={'spider': spider})
diff --git a/scrapy/signals.py b/scrapy/signals.py
index de0886fb6..e36c27203 100644
--- a/scrapy/signals.py
+++ b/scrapy/signals.py
@@ -17,6 +17,7 @@ response_received = object()
 response_downloaded = object()
 item_scraped = object()
 item_dropped = object()
+item_error = object()
 
 # for backwards compatibility
 stats_spider_opened = spider_opened
diff --git a/tests/pipelines.py b/tests/pipelines.py
index ddfbc7a99..7e2895a5c 100644
--- a/tests/pipelines.py
+++ b/tests/pipelines.py
@@ -9,3 +9,9 @@ class ZeroDivisionErrorPipeline(object):
 
     def process_item(self, item, spider):
         return item
+
+
+class ProcessWithZeroDivisionErrorPipiline(object):
+
+    def process_item(self, item, spider):
+        1/0
diff --git a/tests/test_engine.py b/tests/test_engine.py
index 04113ddcf..719c0c60c 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -74,6 +74,14 @@ class DictItemsSpider(TestSpider):
     item_cls = dict
 
 
+class ItemZeroDivisionErrorSpider(TestSpider):
+    custom_settings = {
+        "ITEM_PIPELINES": {
+            "tests.pipelines.ProcessWithZeroDivisionErrorPipiline": 300,
+        }
+    }
+
+
 def start_test_site(debug=False):
     root_dir = os.path.join(tests_datadir, "test_site")
     r = static.File(root_dir)
@@ -95,6 +103,7 @@ class CrawlerRun(object):
         self.respplug = []
         self.reqplug = []
         self.reqdropped = []
+        self.itemerror = []
         self.itemresp = []
         self.signals_catched = {}
         self.spider_class = spider_class
@@ -112,6 +121,7 @@ class CrawlerRun(object):
 
         self.crawler = get_crawler(self.spider_class)
         self.crawler.signals.connect(self.item_scraped, signals.item_scraped)
+        self.crawler.signals.connect(self.item_error, signals.item_error)
         self.crawler.signals.connect(self.request_scheduled, signals.request_scheduled)
         self.crawler.signals.connect(self.request_dropped, signals.request_dropped)
         self.crawler.signals.connect(self.response_downloaded, signals.response_downloaded)
@@ -136,6 +146,9 @@ class CrawlerRun(object):
         u = urlparse(url)
         return u.path
 
+    def item_error(self, item, response, spider, failure):
+        self.itemerror.append((item, response, spider, failure))
+
     def item_scraped(self, item, spider, response):
         self.itemresp.append((item, response))
 
@@ -175,6 +188,10 @@ class EngineTest(unittest.TestCase):
         self._assert_scheduled_requests(urls_to_visit=7)
         self._assert_dropped_requests()
 
+        self.run = CrawlerRun(ItemZeroDivisionErrorSpider)
+        yield self.run.run()
+        self._assert_items_error()
+
     def _assert_visited_urls(self):
         must_be_visited = ["/", "/redirect", "/redirected",
                            "/item1.html", "/item2.html", "/item999.html"]
@@ -209,6 +226,20 @@ class EngineTest(unittest.TestCase):
             if self.run.getpath(response.url) == '/redirect':
                 self.assertEqual(302, response.status)
 
+    def _assert_items_error(self):
+        self.assertEqual(2, len(self.run.itemerror))
+        for item, response, spider, failure in self.run.itemerror:
+            self.assertEqual(failure.value.__class__, ZeroDivisionError)
+            self.assertEqual(spider, self.run.spider)
+
+            self.assertEqual(item['url'], response.url)
+            if 'item1.html' in item['url']:
+                self.assertEqual('Item 1 name', item['name'])
+                self.assertEqual('100', item['price'])
+            if 'item2.html' in item['url']:
+                self.assertEqual('Item 2 name', item['name'])
+                self.assertEqual('200', item['price'])
+
     def _assert_scraped_items(self):
         self.assertEqual(2, len(self.run.itemresp))
         for item, response in self.run.itemresp: