diff --git a/java/org/apache/coyote/http11/AbstractHttp11Protocol.java b/java/org/apache/coyote/http11/AbstractHttp11Protocol.java
index b7ec72c203d0..78747aa29620 100644
--- a/java/org/apache/coyote/http11/AbstractHttp11Protocol.java
+++ b/java/org/apache/coyote/http11/AbstractHttp11Protocol.java
@@ -212,6 +212,30 @@ public void setUseKeepAliveResponseHeader(boolean useKeepAliveResponseHeader) {
     }
 
 
+    private boolean dropConnectionOnErrorStatus = true;
+
+    /**
+     * Get the flag that controls whether the connection is closed after a response with a status code from the list of
+     * codes that Tomcat treats as requiring the connection to be closed. Errors that make it unsafe to continue to use
+     * the connection, such as requests that can not be parsed, always close the connection regardless of this setting.
+     *
+     * @return {@code true} if the connection will be closed after such a response
+     */
+    public boolean getDropConnectionOnErrorStatus() {
+        return dropConnectionOnErrorStatus;
+    }
+
+    /**
+     * Set the flag that controls whether the connection is closed after a response with a status code from the list of
+     * codes that Tomcat treats as requiring the connection to be closed.
+     *
+     * @param dropConnectionOnErrorStatus {@code true} if the connection should be closed after such a response
+     */
+    public void setDropConnectionOnErrorStatus(boolean dropConnectionOnErrorStatus) {
+        this.dropConnectionOnErrorStatus = dropConnectionOnErrorStatus;
+    }
+
+
     private String relaxedPathChars = null;
 
     /**
diff --git a/java/org/apache/coyote/http11/Http11Processor.java b/java/org/apache/coyote/http11/Http11Processor.java
index a8bee04f9408..897614f8b37e 100644
--- a/java/org/apache/coyote/http11/Http11Processor.java
+++ b/java/org/apache/coyote/http11/Http11Processor.java
@@ -408,6 +408,7 @@ public SocketState service(SocketWrapperBase<?> socketWrapper) throws IOExceptio
                     // If we fail here, then the response is likely already
                     // committed, so we can't try and set headers.
                     if (keepAlive && !getErrorState().isError() && !isAsync() &&
+                            protocol.getDropConnectionOnErrorStatus() &&
                             statusDropsConnection(response.getStatus())) {
                         setErrorState(ErrorState.CLOSE_CLEAN, null);
                     }
@@ -1028,7 +1029,8 @@ protected final void prepareResponse() throws IOException {
 
         // If we know that the request is bad this early, add the
         // Connection: close header.
-        if (keepAlive && statusDropsConnection(statusCode)) {
+        if (keepAlive && statusDropsConnection(statusCode) &&
+                (protocol.getDropConnectionOnErrorStatus() || getErrorState().isError())) {
             keepAlive = false;
         }
         if (!keepAlive) {
diff --git a/test/org/apache/coyote/http11/TestHttp11Processor.java b/test/org/apache/coyote/http11/TestHttp11Processor.java
index 6ed4bbb8385d..643622775170 100644
--- a/test/org/apache/coyote/http11/TestHttp11Processor.java
+++ b/test/org/apache/coyote/http11/TestHttp11Processor.java
@@ -29,6 +29,7 @@
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketAddress;
+import java.net.SocketException;
 import java.nio.CharBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -1787,6 +1788,178 @@ private void doTestKeepAliveHeader(boolean sendKeepAlive, int keepAliveTimeout,
     }
 
 
+    @Test
+    public void testDropConnectionOnErrorStatusDefault() throws Exception {
+        Client client = startDropConnectionTest(null, null);
+        client.setRequest(new String[] { dropConnectionRequest("GET", "/error", "") });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.isResponse50x());
+        assertConnectionClosedAfterResponse(client);
+    }
+
+
+    @Test
+    public void testDropConnectionOnErrorStatusDisabled() throws Exception {
+        Client client = startDropConnectionTest("false", null);
+        client.setRequest(new String[] { dropConnectionRequest("GET", "/error", "") });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.isResponse50x());
+        Assert.assertNull(getConnectionHeader(client));
+        assertConnectionReusable(client);
+    }
+
+
+    @Test
+    public void testDropConnectionOnErrorStatusDisabledInvalidRequestLine() throws Exception {
+        Client client = startDropConnectionTest("false", null);
+        client.setRequest(new String[] { "GET /foo\u0001 HTTP/1.1" + CRLF + "Host: localhost" + CRLF + CRLF });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.isResponse400());
+        assertConnectionClosedAfterResponse(client);
+    }
+
+
+    @Test
+    public void testDropConnectionOnErrorStatusDisabledMissingHost() throws Exception {
+        Client client = startDropConnectionTest("false", null);
+        client.setRequest(new String[] { "GET /foo HTTP/1.1" + CRLF + CRLF });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.isResponse400());
+        assertConnectionClosedAfterResponse(client);
+    }
+
+
+    @Test
+    public void testDropConnectionOnErrorStatusDisabledUnsupportedEncoding() throws Exception {
+        Client client = startDropConnectionTest("false", null);
+        // @formatter:off
+        String request =
+                "POST /foo HTTP/1.1" + CRLF +
+                "Host: localhost" + CRLF +
+                "Transfer-Encoding: unsupported" + CRLF +
+                CRLF;
+        // @formatter:on
+        client.setRequest(new String[] { request });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.isResponse50x());
+        Assert.assertTrue(client.getResponseLine(), client.getResponseLine().contains(" 501"));
+        assertConnectionClosedAfterResponse(client);
+    }
+
+
+    @Test
+    public void testDropConnectionOnErrorStatusDisabledSwallowBody() throws Exception {
+        Client client = startDropConnectionTest("false", null);
+        client.setRequest(new String[] { dropConnectionRequest("POST", "/error", "0123456789") });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.isResponse50x());
+        Assert.assertNull(getConnectionHeader(client));
+        // The next request will fail to parse if the unread body was not swallowed
+        assertConnectionReusable(client);
+    }
+
+
+    @Test
+    public void testDropConnectionOnErrorStatusDisabledBodyExceedsMaxSwallowSize() throws Exception {
+        Client client = startDropConnectionTest("false", "10");
+        client.setRequest(new String[] { dropConnectionRequest("POST", "/error", "0123456789ABCDEF") });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.isResponse50x());
+        assertConnectionClosedAfterResponse(client);
+    }
+
+
+    private Client startDropConnectionTest(String dropConnectionOnErrorStatus, String maxSwallowSize)
+            throws Exception {
+        Tomcat tomcat = getTomcatInstance();
+
+        if (dropConnectionOnErrorStatus != null) {
+            Assert.assertTrue(
+                    tomcat.getConnector().setProperty("dropConnectionOnErrorStatus", dropConnectionOnErrorStatus));
+        }
+        if (maxSwallowSize != null) {
+            Assert.assertTrue(tomcat.getConnector().setProperty("maxSwallowSize", maxSwallowSize));
+        }
+
+        // No file system docBase required
+        Context ctx = getProgrammaticRootContext();
+
+        Tomcat.addServlet(ctx, "ErrorStatusServlet", new ErrorStatusServlet());
+        ctx.addServletMapping("/*", "ErrorStatusServlet");
+
+        tomcat.start();
+
+        Client client = new Client(tomcat.getConnector().getLocalPort());
+        client.setUseContentLength(true);
+        client.connect();
+        return client;
+    }
+
+
+    private static String dropConnectionRequest(String method, String path, String body) {
+        // @formatter:off
+        return method + " " + path + " HTTP/1.1" + CRLF +
+                "Host: localhost" + CRLF +
+                (body.isEmpty() ? "" : "Content-Length: " + body.length() + CRLF) +
+                CRLF +
+                body;
+        // @formatter:on
+    }
+
+
+    private static String getConnectionHeader(Client client) {
+        for (String header : client.getResponseHeaders()) {
+            if (header.regionMatches(true, 0, "Connection:", 0, 11)) {
+                return header.substring(11).trim();
+            }
+        }
+        return null;
+    }
+
+
+    private static void assertConnectionClosedAfterResponse(Client client) throws IOException {
+        Assert.assertEquals("close", getConnectionHeader(client));
+        try {
+            Assert.assertNull(client.readLine());
+        } catch (SocketException e) {
+            // A TCP reset also shows that the server closed the connection
+        }
+    }
+
+
+    private static void assertConnectionReusable(Client client) throws Exception {
+        client.setRequest(new String[] { dropConnectionRequest("GET", "/ok", "") });
+        client.processRequest(true);
+
+        Assert.assertTrue(client.getResponseLine(), client.isResponse200());
+        Assert.assertEquals("OK", client.getResponseBody());
+    }
+
+
+    private static final class ErrorStatusServlet extends HttpServlet {
+
+        private static final long serialVersionUID = 1L;
+
+        @Override
+        protected void service(HttpServletRequest req, HttpServletResponse resp) throws IOException {
+            if (req.getRequestURI().equals("/error")) {
+                resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+            } else {
+                resp.setContentType("text/plain");
+                resp.setContentLength(2);
+                resp.getWriter().print("OK");
+            }
+        }
+    }
+
+
     /**
      * Test servlet that prints out the values of HttpServletRequest.getServerName() and
      * HttpServletRequest.getServerPort() in the response body, e.g.: "request.getServerName() is [foo] and
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index fc1d308edd95..f5c89f4abef9 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -664,6 +664,13 @@
         Detect and disconnect a mis-behaving HTTP/2 client that sends large
         numbers of unexpected settings acknowledgement frames. (markt)
       </fix>
+      <add>
+        Add the <code>dropConnectionOnErrorStatus</code> attribute to the HTTP/1.1
+        connector to allow the closing of the connection after a response with
+        selected 4xx and 5xx status codes to be disabled, reducing the resources
+        used to re-establish those connections. The default value of
+        <code>true</code> retains the current behaviour. (ooeunz)
+      </add>
     </changelog>
   </subsection>
   <subsection name="Jasper">
diff --git a/webapps/docs/config/http.xml b/webapps/docs/config/http.xml
index c87099b58772..40f178baa4fd 100644
--- a/webapps/docs/config/http.xml
+++ b/webapps/docs/config/http.xml
@@ -557,6 +557,17 @@
       </p>
     </attribute>
 
+    <attribute name="dropConnectionOnErrorStatus" required="false">
+      <p>(bool) Use this attribute to control whether the connection is closed
+      after a response with a status code of 400, 408, 411, 413, 414, 500, 501
+      or 503. If set to <code>false</code>, the connection may be re-used for
+      further requests after such a response. The connection will still be
+      closed if it is not safe to re-use it, for example if the request could
+      not be parsed, if the request used an unsupported transfer encoding or if
+      the unread part of the request body exceeds <strong>maxSwallowSize</strong>.
+      If not specified, this attribute is set to <code>true</code>.</p>
+    </attribute>
+
     <attribute name="executor" required="false">
       <p>A reference to the name in an <a href="executor.html">Executor</a>
       element. If this attribute is set, and the named executor exists, the