diff --git a/lib/plug/conn.ex b/lib/plug/conn.ex
index 08f2d133..656484fa 100644
--- a/lib/plug/conn.ex
+++ b/lib/plug/conn.ex
@@ -1067,6 +1067,8 @@ defmodule Plug.Conn do
     * `:validate_utf8` - boolean that tells whether or not to validate the keys and
       values of the decoded query string are UTF-8 encoded. Defaults to `true`.
 
+   * `:validate_utf8_error` - status code if validation fails. Defaults to `400`.
+
   """
   @spec fetch_query_params(t, Keyword.t()) :: t
   def fetch_query_params(conn, opts \\ [])
@@ -1087,7 +1089,8 @@ defmodule Plug.Conn do
         query_string,
         %{},
         Plug.Conn.InvalidQueryError,
-        Keyword.get(opts, :validate_utf8, true)
+        Keyword.get(opts, :validate_utf8, true),
+        Keyword.get(opts, :validate_utf8_error, 400)
       )
 
     case params do
diff --git a/lib/plug/conn/query.ex b/lib/plug/conn/query.ex
index e7d9026a..7e8a089d 100644
--- a/lib/plug/conn/query.ex
+++ b/lib/plug/conn/query.ex
@@ -86,67 +86,84 @@ defmodule Plug.Conn.Query do
   Decodes the given `query`.
 
   The `query` is assumed to be encoded in the "x-www-form-urlencoded" format.
-  The format is decoded at first. Then, if `validate_utf8` is `true`, the decoded
-  result is validated for proper UTF-8 encoding.
 
   `initial` is the initial "accumulator" where decoded values will be added.
 
   `invalid_exception` is the exception module for the exception to raise on
   errors with decoding.
+
+  If `validate_utf8` is set to `true`, the function validates that the decoded
+  result is properly encoded in UTF-8. If validation fails, it raises an exception
+  based on the status code of `validate_utf8_error`.
+
+
   """
-  @spec decode(String.t(), keyword(), module(), boolean()) :: %{optional(String.t()) => term()}
+  @spec decode(String.t(), keyword(), module(), boolean(), integer()) :: %{
+          optional(String.t()) => term()
+        }
   def decode(
         query,
         initial \\ [],
         invalid_exception \\ Plug.Conn.InvalidQueryError,
-        validate_utf8 \\ true
+        validate_utf8 \\ true,
+        validate_utf8_error \\ 400
       )
 
-  def decode("", initial, _invalid_exception, _validate_utf8) do
+  def decode("", initial, _invalid_exception, _validate_utf8, _validate_utf8_error) do
     Map.new(initial)
   end
 
-  def decode(query, initial, invalid_exception, validate_utf8)
+  def decode(query, initial, invalid_exception, validate_utf8, validate_utf8_error)
       when is_binary(query) do
     parts = :binary.split(query, "&", [:global])
 
     parts
-    |> Enum.reduce(decode_init(), &decode_www_pair(&1, &2, invalid_exception, validate_utf8))
+    |> Enum.reduce(
+      decode_init(),
+      &decode_www_pair(&1, &2, invalid_exception, validate_utf8, validate_utf8_error)
+    )
     |> decode_done(initial)
   end
 
-  defp decode_www_pair("", acc, _invalid_exception, _validate_utf8) do
+  defp decode_www_pair("", acc, _invalid_exception, _validate_utf8, _validate_utf8_error) do
     acc
   end
 
-  defp decode_www_pair(binary, acc, invalid_exception, validate_utf8) do
+  defp decode_www_pair(binary, acc, invalid_exception, validate_utf8, validate_utf8_error) do
     current =
       case :binary.split(binary, "=") do
         [key, value] ->
-          {decode_www_form(key, invalid_exception, validate_utf8),
-           decode_www_form(value, invalid_exception, validate_utf8)}
+          {decode_www_form(key, invalid_exception, validate_utf8, validate_utf8_error),
+           decode_www_form(value, invalid_exception, validate_utf8, validate_utf8_error)}
 
         [key] ->
-          {decode_www_form(key, invalid_exception, validate_utf8), ""}
+          {decode_www_form(key, invalid_exception, validate_utf8, validate_utf8_error), ""}
       end
 
     decode_each(current, acc)
   end
 
-  defp decode_www_form(value, invalid_exception, validate_utf8) do
-    # TODO: Remove rescue as this can't fail from Elixir v1.13
+  defp decode_www_form(value, invalid_exception, validate_utf8, validate_utf8_error) do
     try do
       URI.decode_www_form(value)
     rescue
       ArgumentError ->
-        raise invalid_exception, "invalid urlencoded params, got #{value}"
+        raise invalid_exception,
+              "invalid urlencoded params, got #{value}"
+
+        # catch
+        #   :throw, :malformed_uri ->
+        #     raise invalid_exception,
+        #           "invalid urlencoded params, got #{value}"
     else
       binary ->
-        if validate_utf8 do
-          Plug.Conn.Utils.validate_utf8!(binary, invalid_exception, "urlencoded params")
-        end
-
-        binary
+        Plug.Conn.Utils.validate_utf8!(
+          binary,
+          "urlencoded params",
+          invalid_exception,
+          validate_utf8,
+          validate_utf8_error
+        )
     end
   end
 
diff --git a/lib/plug/conn/utils.ex b/lib/plug/conn/utils.ex
index 5bb7e20a..92e2df1d 100644
--- a/lib/plug/conn/utils.ex
+++ b/lib/plug/conn/utils.ex
@@ -2,6 +2,7 @@ defmodule Plug.Conn.Utils do
   @moduledoc """
   Utilities for working with connection data
   """
+  import Plug.Conn.Status, only: [reason_phrase: 1]
 
   @type params :: %{optional(binary) => binary}
 
@@ -12,6 +13,9 @@ defmodule Plug.Conn.Utils do
   @space [?\s, ?\t]
   @specials ~c|()<>@,;:\\"/[]?={}|
 
+  @client_error_status 400..499
+  @server_error_status 500..599
+
   @doc ~S"""
   Parses media types (with wildcards).
 
@@ -51,6 +55,7 @@ defmodule Plug.Conn.Utils do
       :error
 
   """
+
   @spec media_type(binary) :: {:ok, type :: binary, subtype :: binary, params} | :error
   def media_type(binary) when is_binary(binary) do
     case strip_spaces(binary) do
@@ -278,27 +283,126 @@ defmodule Plug.Conn.Utils do
   end
 
   @doc """
-  Validates the given binary is valid UTF-8.
+  Validates that the given binary is valid UTF-8.
+  Raises exception on failure based on `validate_utf8_error` status code.
   """
-  @spec validate_utf8!(binary, module, binary) :: :ok | no_return
-  def validate_utf8!(binary, exception, context)
+  @spec validate_utf8!(binary, binary, module, boolean, integer) :: binary | no_return
+  def validate_utf8!(
+        binary,
+        context,
+        invalid_exception,
+        validate_utf8 \\ true,
+        validate_utf8_error \\ 400
+      )
+
+  def validate_utf8!(
+        binary,
+        _context,
+        _invalid_exception,
+        false,
+        _validate_utf8_error
+      ),
+      do: binary
+
+  def validate_utf8!(
+        <<binary::binary>>,
+        context,
+        invalid_exception,
+        validate_utf8,
+        validate_utf8_error
+      ) do
+    do_validate_utf8!(
+      binary,
+      binary,
+      context,
+      invalid_exception,
+      validate_utf8,
+      validate_utf8_error
+    )
+  end
 
-  def validate_utf8!(<<binary::binary>>, exception, context) do
-    do_validate_utf8!(binary, exception, context)
+  defp do_validate_utf8!(
+         <<_::utf8, rest::bits>>,
+         return_binary,
+         context,
+         invalid_exception,
+         validate_utf8,
+         validate_utf8_error
+       ) do
+    do_validate_utf8!(
+      rest,
+      return_binary,
+      context,
+      invalid_exception,
+      validate_utf8,
+      validate_utf8_error
+    )
   end
 
-  defp do_validate_utf8!(<<_::utf8, rest::bits>>, exception, context) do
-    do_validate_utf8!(rest, exception, context)
+  defp do_validate_utf8!(
+         <<byte, _::bits>>,
+         _return_binary,
+         _context,
+         invalid_exception,
+         _validate_utf8,
+         400
+       ) do
+    raise invalid_exception,
+          "invalid UTF-8 on urlencoded params, got byte #{byte}"
   end
 
-  defp do_validate_utf8!(<<byte, _::bits>>, exception, context) do
-    raise exception, "invalid UTF-8 on #{context}, got byte #{byte}"
+  defp do_validate_utf8!(
+         <<_byte, _::bits>>,
+         _return_binary,
+         context,
+         invalid_exception,
+         _validate_utf8,
+         404
+       ) do
+    raise invalid_exception,
+          "resource could not be found in #{context}"
   end
 
-  defp do_validate_utf8!(<<>>, _exception, _context) do
-    :ok
+  defp do_validate_utf8!(
+         <<_byte, _::bits>>,
+         _return_binary,
+         _context,
+         _invalid_exception,
+         _validate_utf8,
+         validate_utf8_error
+       )
+       when validate_utf8_error in @client_error_status do
+    raise Plug.BadRequestError,
+      message: "could not process the request due to client error",
+      plug_status: validate_utf8_error,
+      plug_reason_phrase: reason_phrase(validate_utf8_error)
   end
 
+  defp do_validate_utf8!(
+         <<_byte, _::bits>>,
+         _return_binary,
+         _context,
+         _invalid_exception,
+         _validate_utf8,
+         validate_utf8_error
+       )
+       when validate_utf8_error in @server_error_status do
+    raise Plug.BadResponseError,
+      message: "could not process the request due to server error",
+      plug_status: validate_utf8_error,
+      plug_reason_phrase: reason_phrase(validate_utf8_error)
+  end
+
+  defp do_validate_utf8!(
+         <<>>,
+         return_binary,
+         _context,
+         _invalid_exception,
+         _validate_utf8,
+         _validate_utf8_error
+       ),
+       do: return_binary
+
   ## Helpers
 
   defp strip_spaces("\r\n" <> t), do: strip_spaces(t)
diff --git a/lib/plug/exceptions.ex b/lib/plug/exceptions.ex
index 843e5b04..33589efc 100644
--- a/lib/plug/exceptions.ex
+++ b/lib/plug/exceptions.ex
@@ -55,16 +55,59 @@ end
 
 defmodule Plug.BadRequestError do
   @moduledoc """
-  The request will not be processed due to a client error.
+  An exception raised when the request will not be processed due to a client error.
   """
+  defexception message: "could not process the request due to client error.",
+               plug_status: 400,
+               plug_message: "Bad Request"
+
+  def message(exception) do
+    "could not process the request due to client error. \n
+    (Status Code: #{exception.plug_status} #{exception.plug_message})"
+  end
+end
 
-  defexception message: "could not process the request due to client error", plug_status: 400
+defmodule Plug.BadResponseError do
+  @moduledoc """
+  An exception raised when the request will not be processed due to a server error.
+  """
+  defexception message: "could not process the request due to server error.",
+               plug_status: 500,
+               plug_message: "Internal Server Error"
+
+  def message(exception) do
+    "could not process the request due to server error: (Status Code: #{exception.plug_status} #{exception.plug_message})"
+  end
 end
 
 defmodule Plug.TimeoutError do
   @moduledoc """
-  Timeout while waiting for the request.
+  An exception raised when the request times out while waiting for the request.
   """
 
   defexception message: "timeout while waiting for request data", plug_status: 408
 end
+
+defmodule Plug.ResourceNotFoundError do
+  @moduledoc """
+  An exception raised when the requested resource cannot be located.
+  """
+
+  defexception message: "The requested resource could not be found", plug_status: 404
+end
+
+defmodule Plug.ClientError do
+  @moduledoc """
+  An exception raised when the requested resource returned a 4XX status code.
+  """
+
+  defexception message: "The requested resource could not be found", plug_status: 404
+end
+
+defmodule Plug.ServerError do
+  @moduledoc """
+  An exception raised when
+  """
+
+  defexception message: "The requested resource could not be found", plug_status: 404
+end
diff --git a/lib/plug/parsers.ex b/lib/plug/parsers.ex
index a5efb3cf..4e740c88 100644
--- a/lib/plug/parsers.ex
+++ b/lib/plug/parsers.ex
@@ -95,7 +95,10 @@ defmodule Plug.Parsers do
     * `:query_string_length` - the maximum allowed size for query strings
 
     * `:validate_utf8` - boolean that tells whether or not we want to
-        validate that parsed binaries are utf8 strings.
+        validate that parsed binaries are utf8 strings. Defaults to true.
+
+    * `:validate_utf8_error` - status code returned if validation fails.
+        Defaults to 400.
 
     * `:body_reader` - an optional replacement (or wrapper) for
       `Plug.Conn.read_body/2` to provide a function that gives access to the
@@ -254,12 +257,14 @@ defmodule Plug.Parsers do
     {pass, opts} = Keyword.pop(opts, :pass, [])
     {query_string_length, opts} = Keyword.pop(opts, :query_string_length, 1_000_000)
     validate_utf8 = Keyword.get(opts, :validate_utf8, true)
+    validate_utf8_error = Keyword.get(opts, :validate_utf8_error, 400)
 
     unless parsers do
       raise ArgumentError, "Plug.Parsers expects a set of parsers to be given in :parsers"
     end
 
-    {convert_parsers(parsers, opts), pass, query_string_length, validate_utf8}
+    {convert_parsers(parsers, opts), pass, query_string_length, validate_utf8,
+     validate_utf8_error}
   end
 
   defp convert_parsers(parsers, root_opts) do
@@ -286,13 +291,14 @@ defmodule Plug.Parsers do
   @impl true
   def call(%{method: method, body_params: %Plug.Conn.Unfetched{}} = conn, options)
       when method in @methods do
-    {parsers, pass, query_string_length, validate_utf8} = options
+    {parsers, pass, query_string_length, validate_utf8, validate_utf8_error} = options
     %{req_headers: req_headers} = conn
 
     conn =
       Conn.fetch_query_params(conn,
         length: query_string_length,
-        validate_utf8: validate_utf8
+        validate_utf8: validate_utf8,
+        validate_utf8_error: validate_utf8_error
       )
 
     case List.keyfind(req_headers, "content-type", 0) do
@@ -307,23 +313,41 @@ defmodule Plug.Parsers do
               params,
               pass,
               query_string_length,
-              validate_utf8
+              validate_utf8,
+              validate_utf8_error
             )
 
           :error ->
-            reduce(conn, parsers, ct, "", %{}, pass, query_string_length, validate_utf8)
+            reduce(
+              conn,
+              parsers,
+              ct,
+              "",
+              %{},
+              pass,
+              query_string_length,
+              validate_utf8,
+              validate_utf8_error
+            )
         end
 
       _ ->
-        {conn, params} = merge_params(conn, %{}, query_string_length, validate_utf8)
+        {conn, params} =
+          merge_params(conn, %{}, query_string_length, validate_utf8, validate_utf8_error)
 
         %{conn | params: params, body_params: %{}}
     end
   end
 
-  def call(%{body_params: body_params} = conn, {_, _, query_string_length, validate_utf8}) do
+  def call(
+        %{body_params: body_params} = conn,
+        {_, _, query_string_length, validate_utf8, validate_utf8_error}
+      ) do
     body_params = make_empty_if_unfetched(body_params)
-    {conn, params} = merge_params(conn, body_params, query_string_length, validate_utf8)
+
+    {conn, params} =
+      merge_params(conn, body_params, query_string_length, validate_utf8, validate_utf8_error)
+
     %{conn | params: params, body_params: body_params}
   end
 
@@ -335,24 +359,49 @@ defmodule Plug.Parsers do
          params,
          pass,
          query_string_length,
-         validate_utf8
+         validate_utf8,
+         validate_utf8_error
        ) do
     case parser.parse(conn, type, subtype, params, options) do
       {:ok, body, conn} ->
-        {conn, params} = merge_params(conn, body, query_string_length, validate_utf8)
+        {conn, params} =
+          merge_params(conn, body, query_string_length, validate_utf8, validate_utf8_error)
+
         %{conn | params: params, body_params: body}
 
       {:next, conn} ->
-        reduce(conn, rest, type, subtype, params, pass, query_string_length, validate_utf8)
+        reduce(
+          conn,
+          rest,
+          type,
+          subtype,
+          params,
+          pass,
+          query_string_length,
+          validate_utf8,
+          validate_utf8_error
+        )
 
       {:error, :too_large, _conn} ->
         raise RequestTooLargeError
     end
   end
 
-  defp reduce(conn, [], type, subtype, _params, pass, query_string_length, validate_utf8) do
+  defp reduce(
+         conn,
+         [],
+         type,
+         subtype,
+         _params,
+         pass,
+         query_string_length,
+         validate_utf8,
+         validate_utf8_error
+       ) do
     if accepted_mime?(type, subtype, pass) do
-      {conn, params} = merge_params(conn, %{}, query_string_length, validate_utf8)
+      {conn, params} =
+        merge_params(conn, %{}, query_string_length, validate_utf8, validate_utf8_error)
+
       %{conn | params: params}
     else
       raise UnsupportedMediaTypeError, media_type: "#{type}/#{subtype}"
@@ -365,14 +414,15 @@ defmodule Plug.Parsers do
   defp accepted_mime?(type, subtype, pass),
     do: "#{type}/#{subtype}" in pass || "#{type}/*" in pass
 
-  defp merge_params(conn, body_params, query_string_length, validate_utf8) do
+  defp merge_params(conn, body_params, query_string_length, validate_utf8, validate_utf8_error) do
     %{params: params, path_params: path_params} = conn
     params = make_empty_if_unfetched(params)
 
     conn =
       Plug.Conn.fetch_query_params(conn,
         length: query_string_length,
-        validate_utf8: validate_utf8
+        validate_utf8: validate_utf8,
+        validate_utf8_error: validate_utf8_error
       )
 
     {conn,
diff --git a/lib/plug/parsers/multipart.ex b/lib/plug/parsers/multipart.ex
index 5a4ec1a0..01c31f8d 100644
--- a/lib/plug/parsers/multipart.ex
+++ b/lib/plug/parsers/multipart.ex
@@ -29,6 +29,9 @@ defmodule Plug.Parsers.MULTIPART do
     * `:validate_utf8` - specifies whether multipart body parts should be validated
       as utf8 binaries. Defaults to true
 
+    * `:validate_utf8_error` - specifies the status code if validation fails,
+      defaults to 400.
+
     * `:multipart_to_params` - a MFA that receives the multipart headers and the
       connection and it must return a tuple of `{:ok, params, conn}`
 
@@ -185,7 +188,13 @@ defmodule Plug.Parsers.MULTIPART do
           parse_multipart_body(Plug.Conn.read_part_body(conn, opts), limit, opts, "")
 
         if Keyword.get(opts, :validate_utf8, true) do
-          Plug.Conn.Utils.validate_utf8!(body, Plug.Parsers.BadEncodingError, "multipart body")
+          Plug.Conn.Utils.validate_utf8!(
+            body,
+            "multipart body",
+            Plug.Parsers.BadEncodingError,
+            Keyword.get(opts, :validate_utf8),
+            Keyword.get(opts, :validate_utf8_error, 400)
+          )
         end
 
         {conn, limit, [{name, headers, body} | acc]}
diff --git a/lib/plug/parsers/urlencoded.ex b/lib/plug/parsers/urlencoded.ex
index 04260e30..219a9257 100644
--- a/lib/plug/parsers/urlencoded.ex
+++ b/lib/plug/parsers/urlencoded.ex
@@ -31,13 +31,15 @@ defmodule Plug.Parsers.URLENCODED do
     case apply(mod, fun, [conn, opts | args]) do
       {:ok, body, conn} ->
         validate_utf8 = Keyword.get(opts, :validate_utf8, true)
+        validate_utf8_error = Keyword.get(opts, :validate_utf8_error, 400)
 
         {:ok,
          Plug.Conn.Query.decode(
            body,
            %{},
            Plug.Parsers.BadEncodingError,
-           validate_utf8
+           validate_utf8,
+           validate_utf8_error
          ), conn}
 
       {:more, _data, conn} ->
diff --git a/test/plug/conn/query_test.exs b/test/plug/conn/query_test.exs
index 35d903ee..5f6035a5 100644
--- a/test/plug/conn/query_test.exs
+++ b/test/plug/conn/query_test.exs
@@ -1,7 +1,16 @@
 defmodule Plug.Conn.QueryTest do
   use ExUnit.Case, async: true
 
-  import Plug.Conn.Query, only: [decode: 1, encode: 1, encode: 2]
+  import Plug.Conn.Query,
+    only: [
+      decode: 1,
+      encode: 1,
+      encode: 2,
+      decode_init: 0,
+      decode_each: 2,
+      decode_done: 2
+    ]
+
   doctest Plug.Conn.Query
 
   describe "decode" do
@@ -181,7 +190,9 @@ defmodule Plug.Conn.QueryTest do
     end
 
     defp decode_pair(pairs) do
-      Enum.reduce(Enum.reverse(pairs), %{}, &Plug.Conn.Query.decode_pair(&1, &2))
+      pairs
+      |> Enum.reduce(decode_init(), &decode_each(&1, &2))
+      |> decode_done([])
     end
   end
 end