diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 42944556..355007c5 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -4,6 +4,30 @@ Changelog
 4.0.0 (unreleased)
 ******************
 
+Features:
+
+* *Backwards-incompatible*: Custom error handlers receive the
+  `marshmallow.Schema` instance as the third argument. Update any
+  functions decorated with `Parser.error_handler` to take a ``schema``
+  argument, like so:
+
+.. code-block:: python
+
+    # 3.x
+    @parser.error_handler
+    def handle_error(error, req):
+        raise CustomError(error.messages)
+
+
+    # 4.x
+    @parser.error_handler
+    def handle_error(error, req, schema):
+        raise CustomError(error.messages)
+
+
+See `marshmallow-code/marshmallow#840 (comment) <https://github.com/marshmallow-code/marshmallow/issues/840#issuecomment-403481686>`_
+for more information about this change.
+
 Bug fixes:
 
 * *Backwards-incompatible*: Rename ``webargs.async`` to
diff --git a/docs/framework_support.rst b/docs/framework_support.rst
index a81754dc..a81cd578 100644
--- a/docs/framework_support.rst
+++ b/docs/framework_support.rst
@@ -31,7 +31,8 @@ When using the :meth:`use_args <webargs.flaskparser.FlaskParser.use_args>` decor
 Error Handling
 ++++++++++++++
 
-Webargs uses Flask's ``abort`` function to raise an ``HTTPException`` when a validation error occurs. If you use the ``Flask.errorhandler`` method to handle errors, you can access validation messages from the ``data`` attribute of an error.
+Webargs uses Flask's ``abort`` function to raise an ``HTTPException`` when a validation error occurs.
+If you use the ``Flask.errorhandler`` method to handle errors, you can access validation messages from the ``data`` attribute of an error.
 
 Here is an example error handler that returns validation messages to the client as JSON.
 
diff --git a/docs/quickstart.rst b/docs/quickstart.rst
index bf7cda98..aeb5b330 100644
--- a/docs/quickstart.rst
+++ b/docs/quickstart.rst
@@ -185,14 +185,14 @@ Error Handling
 --------------
 
 Each parser has a default error handling method. To override the error handling callback, write a function that
-receives an error and the request and handles the error.
+receives an error, the request, and the `marshmallow.Schema` instance.
 Then decorate that function with :func:`Parser.error_handler <webargs.core.Parser.error_handler>`.
 
 .. code-block:: python
 
-    from webargs import core
+    from webargs import flaskparser
 
-    parser = core.Parser()
+    parser = flaskparser.FlaskParser()
 
 
     class CustomError(Exception):
@@ -200,7 +200,7 @@ Then decorate that function with :func:`Parser.error_handler <webargs.core.Parse
 
 
     @parser.error_handler
-    def handle_error(error, req):
+    def handle_error(error, req, schema):
         raise CustomError(error.messages)
 
 Nesting Fields
diff --git a/tests/apps/flask_app.py b/tests/apps/flask_app.py
index d782287c..0a1b7d27 100644
--- a/tests/apps/flask_app.py
+++ b/tests/apps/flask_app.py
@@ -183,4 +183,5 @@ def echo_use_kwargs_missing(username, password):
 # Return validation errors as JSON
 @app.errorhandler(422)
 def handle_validation_error(err):
+    assert isinstance(err.data["schema"], ma.Schema)
     return J({"errors": err.exc.messages}), 422
diff --git a/tests/test_core.py b/tests/test_core.py
index 5ce0f517..a435484a 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -265,7 +265,7 @@ def test_handle_error_called_when_parsing_raises_error(
 def test_handle_error_reraises_errors(web_request):
     p = Parser()
     with pytest.raises(ValidationError):
-        p.handle_error(ValidationError("error raised"), web_request)
+        p.handle_error(ValidationError("error raised"), web_request, Schema())
 
 
 @mock.patch("webargs.core.Parser.parse_headers")
@@ -287,7 +287,8 @@ def test_custom_error_handler(parse_json, web_request):
     class CustomError(Exception):
         pass
 
-    def error_handler(error, req):
+    def error_handler(error, req, schema):
+        assert isinstance(schema, Schema)
         raise CustomError(error)
 
     parse_json.side_effect = ValidationError("parse_json failed")
@@ -306,7 +307,8 @@ class CustomError(Exception):
     parser = Parser()
 
     @parser.error_handler
-    def handle_error(error, req):
+    def handle_error(error, req, schema):
+        assert isinstance(schema, Schema)
         raise CustomError(error)
 
     with pytest.raises(CustomError):
diff --git a/webargs/aiohttpparser.py b/webargs/aiohttpparser.py
index e19ce8e2..8dac815b 100644
--- a/webargs/aiohttpparser.py
+++ b/webargs/aiohttpparser.py
@@ -132,7 +132,7 @@ def get_request_from_view_args(self, view, args, kwargs):
         assert isinstance(req, web.Request), "Request argument not found for handler"
         return req
 
-    def handle_error(self, error, req):
+    def handle_error(self, error, req, schema):
         """Handle ValidationErrors and return a JSON response of error messages to the client."""
         error_class = exception_map.get(error.status_code)
         if not error_class:
diff --git a/webargs/asyncparser.py b/webargs/asyncparser.py
index bcd4d66b..02f03696 100644
--- a/webargs/asyncparser.py
+++ b/webargs/asyncparser.py
@@ -76,7 +76,7 @@ async def parse(
             data = result.data if core.MARSHMALLOW_VERSION_INFO[0] < 3 else result
             self._validate_arguments(data, validators)
         except ma.exceptions.ValidationError as error:
-            self._on_validation_error(error, req)
+            self._on_validation_error(error, req, schema)
         finally:
             self.clear_cache()
         if force_all:
diff --git a/webargs/bottleparser.py b/webargs/bottleparser.py
index 304c66a8..ada807b2 100644
--- a/webargs/bottleparser.py
+++ b/webargs/bottleparser.py
@@ -57,7 +57,7 @@ def parse_files(self, req, name, field):
         """Pull a file from the request."""
         return core.get_value(req.files, name, field)
 
-    def handle_error(self, error, req):
+    def handle_error(self, error, req, schema):
         """Handles errors during parsing. Aborts the current request with a
         400 error.
         """
diff --git a/webargs/core.py b/webargs/core.py
index 4771adde..06b389b7 100755
--- a/webargs/core.py
+++ b/webargs/core.py
@@ -317,7 +317,7 @@ def _parse_request(self, schema, req, locations):
                     parsed[argname] = parsed_value
         return parsed
 
-    def _on_validation_error(self, error, req):
+    def _on_validation_error(self, error, req, schema):
         if isinstance(error, ma.exceptions.ValidationError) and not isinstance(
             error, ValidationError
         ):
@@ -331,9 +331,9 @@ def _on_validation_error(self, error, req):
                 kwargs["status_code"] = self.DEFAULT_VALIDATION_STATUS
             error = ValidationError(error.messages, **kwargs)
         if self.error_callback:
-            self.error_callback(error, req)
+            self.error_callback(error, req, schema)
         else:
-            self.handle_error(error, req)
+            self.handle_error(error, req, schema)
 
     def _validate_arguments(self, data, validators):
         for validator in validators:
@@ -393,7 +393,7 @@ def parse(self, argmap, req=None, locations=None, validate=None, force_all=False
             data = result.data if MARSHMALLOW_VERSION_INFO[0] < 3 else result
             self._validate_arguments(data, validators)
         except ma.exceptions.ValidationError as error:
-            self._on_validation_error(error, req)
+            self._on_validation_error(error, req, schema)
         finally:
             self.clear_cache()
         if force_all:
@@ -514,7 +514,7 @@ def location_handler(self, name):
             from webargs import core
             parser = core.Parser()
 
-            @parser.location_handler('name')
+            @parser.location_handler("name")
             def parse_data(request, name, field):
                 return request.data.get(name)
 
@@ -529,20 +529,24 @@ def decorator(func):
 
     def error_handler(self, func):
         """Decorator that registers a custom error handling function. The
-        function should received the raised error and the request object. Overrides
+        function should received the raised error, request object, and the
+        `marshmallow.Schema` instance used to parse the request. Overrides
         the parser's ``handle_error`` method.
 
         Example: ::
 
-            from webargs import core
-            parser = core.Parser()
+            from webargs import flaskparser
+
+            parser = flaskparser.FlaskParser()
+
 
             class CustomError(Exception):
                 pass
 
+
             @parser.error_handler
-            def handle_error(error, request):
-                raise CustomError(error)
+            def handle_error(error, req, schema):
+                raise CustomError(error.messages)
 
         :param callable func: The error callback to register.
         """
@@ -587,7 +591,7 @@ def parse_files(self, req, name, arg):
         """
         return missing
 
-    def handle_error(self, error, req):
+    def handle_error(self, error, req, schema):
         """Called if an error occurs while parsing args. By default, just logs and
         raises ``error``.
         """
diff --git a/webargs/falconparser.py b/webargs/falconparser.py
index e09389c8..3a88a67c 100644
--- a/webargs/falconparser.py
+++ b/webargs/falconparser.py
@@ -140,7 +140,7 @@ def parse_files(self, req, name, field):
             "Parsing files not yet supported by {0}".format(self.__class__.__name__)
         )
 
-    def handle_error(self, error, req):
+    def handle_error(self, error, req, schema):
         """Handles errors during parsing."""
         status = status_map.get(error.status_code)
         if status is None:
diff --git a/webargs/flaskparser.py b/webargs/flaskparser.py
index 1e29a391..e1136e60 100644
--- a/webargs/flaskparser.py
+++ b/webargs/flaskparser.py
@@ -93,12 +93,12 @@ def parse_files(self, req, name, field):
         """Pull a file from the request."""
         return core.get_value(req.files, name, field)
 
-    def handle_error(self, error, req):
+    def handle_error(self, error, req, schema):
         """Handles errors during parsing. Aborts the current HTTP request and
         responds with a 422 error.
         """
         status_code = getattr(error, "status_code", self.DEFAULT_VALIDATION_STATUS)
-        abort(status_code, messages=error.messages, exc=error)
+        abort(status_code, exc=error, messages=error.messages, schema=schema)
 
     def get_default_request(self):
         """Override to use Flask's thread-local request objec by default"""
diff --git a/webargs/pyramidparser.py b/webargs/pyramidparser.py
index b91f64c8..98dbb6f8 100644
--- a/webargs/pyramidparser.py
+++ b/webargs/pyramidparser.py
@@ -73,7 +73,7 @@ def parse_matchdict(self, req, name, field):
         """Pull a value from the request's `matchdict`."""
         return core.get_value(req.matchdict, name, field)
 
-    def handle_error(self, error, req):
+    def handle_error(self, error, req, schema):
         """Handles errors during parsing. Aborts the current HTTP request and
         responds with a 400 error.
         """
diff --git a/webargs/tornadoparser.py b/webargs/tornadoparser.py
index 67349f29..d74579a6 100644
--- a/webargs/tornadoparser.py
+++ b/webargs/tornadoparser.py
@@ -113,7 +113,7 @@ def parse_files(self, req, name, field):
         """Pull a file from the request."""
         return get_value(req.files, name, field)
 
-    def handle_error(self, error, req):
+    def handle_error(self, error, req, schema):
         """Handles errors during parsing. Raises a `tornado.web.HTTPError`
         with a 400 error.
         """