Add implementation to treat message size limit exceed as an error

modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/PassThroughConstants.java

@@ -277,4 +277,6 @@ public class PassThroughConstants {
     public static final String CORRELATION_DEFAULT_HEADER = "activityid";
     public static final String TRANSPORT_LATENCY_LOGGER = "transport-latency";
     public static final String TRANSPORT_IN_URL = "TransportInURL";
+    public static final String MESSAGE_SIZE_LIMIT_EXCEEDED = "MESSAGE_SIZE_LIMIT_EXCEEDED";
+    public static final String DROP_MESSAGE_DUE_TO_SIZE_LIMIT_EXCEEDED = "DROP_MESSAGE_DUE_TO_SIZE_LIMIT_EXCEEDED";
 }

modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/PassThroughHttpSender.java

@@ -537,6 +537,12 @@ public void submitResponse(MessageContext msgContext)
         conn.getContext().setAttribute(PassThroughConstants.RESPONSE_MESSAGE_CONTEXT, msgContext);
 
         SourceContext.setResponse(conn, sourceResponse);
+        if (conn.getContext().getAttribute(PassThroughConstants.DROP_MESSAGE_DUE_TO_SIZE_LIMIT_EXCEEDED) != null) {
+            msgContext.setProperty(PassThroughConstants.NO_ENTITY_BODY, true);
+            // Clean up the attribute after using, otherwise, a keep-alive connection will drop the message of
+            // subsequent connections, even if they are within the size limit.
+            conn.getContext().removeAttribute(PassThroughConstants.DROP_MESSAGE_DUE_TO_SIZE_LIMIT_EXCEEDED);
+        }
         Boolean noEntityBody = (Boolean) msgContext.getProperty(PassThroughConstants.NO_ENTITY_BODY);
         Pipe pipe = (Pipe) msgContext.getProperty(PassThroughConstants.PASS_THROUGH_PIPE);
         if ((noEntityBody == null || !noEntityBody) || pipe != null) {

modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceHandler.java

@@ -471,6 +471,16 @@ public void responseReady(NHttpServerConnection conn) {
                     }
                 }
 
+                if (isMessageSizeValidationEnabled) {
+                    if (conn.getContext().getAttribute(PassThroughConstants.MESSAGE_SIZE_LIMIT_EXCEEDED) != null) {
+                        response.setStatus(HttpStatus.SC_BAD_GATEWAY);
+                        response.setStatusLine("Response Entity Too Large");
+                        // Clean up the attribute after using, otherwise, a keep-alive connection will treat subsequent
+                        // responses as message size limit exceeded payloads, even if they are within the size limit.
+                        conn.getContext().removeAttribute(PassThroughConstants.MESSAGE_SIZE_LIMIT_EXCEEDED);
+                    }
+                }
+
                 response.start(conn);
                 HttpContext context = conn.getContext();
                 if (transportLatencyLog.isDebugEnabled()) {

modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceResponse.java

@@ -347,6 +347,11 @@ public void addHeader(String name, String value) {
     public void setStatus(int status) {
         this.status = status;
     }
+
+    public void setStatusLine(String statusLine) {
+        this.statusLine = statusLine;
+    }
+
     public void removeHeader(String name) {
         if (headers.get(name) != null) {
             headers.remove(name);

modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/TargetHandler.java

@@ -25,6 +25,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.http.ConnectionClosedException;
+import org.apache.http.Header;
 import org.apache.http.HttpException;
 import org.apache.http.HttpRequest;
 import org.apache.http.HttpResponse;
@@ -89,10 +90,12 @@ public class TargetHandler implements NHttpClientEventHandler {
     private static boolean isMessageSizeValidationEnabled = false;
 
     private static int validMaxMessageSize = Integer.MAX_VALUE;
+    private static boolean dropMessageWhenSizeLimitExceeded = false;
 
     public static final String PROPERTY_FILE = "passthru-http.properties";
     public static final String MESSAGE_SIZE_VALIDATION = "message.size.validation.enabled";
     public static final String VALID_MAX_MESSAGE_SIZE = "valid.max.message.size.in.bytes";
+    public static final String DROP_MESSAGE_WHEN_SIZE_LIMIT_EXCEEDED = "drop.message.when.size.limit.exceeded";
     public static final String CONNECTION_POOL = "CONNECTION_POOL";
 
     private List<StreamInterceptor> streamInterceptors;
@@ -125,6 +128,9 @@ public TargetHandler(DeliveryAgent deliveryAgent,
         String validMaxMessageSizeStr = MiscellaneousUtil
                 .getProperty(props, VALID_MAX_MESSAGE_SIZE, String.valueOf(Integer.MAX_VALUE));
         isMessageSizeValidationEnabled = Boolean.valueOf(validationProperty);
+        String dropMessageWhenSizeLimitExceededProperty = MiscellaneousUtil
+                .getProperty(props, DROP_MESSAGE_WHEN_SIZE_LIMIT_EXCEEDED, "false");
+        dropMessageWhenSizeLimitExceeded = Boolean.valueOf(dropMessageWhenSizeLimitExceededProperty);
         try {
             validMaxMessageSize = Integer.valueOf(validMaxMessageSizeStr);
         } catch (NumberFormatException e) {
@@ -382,6 +388,34 @@ public void responseReceived(NHttpClientConnection conn) {
         HttpResponse response = conn.getHttpResponse();
         ProtocolState connState;
         try {
+            if (isMessageSizeValidationEnabled) {
+                Header contentLengthHeader = response.getFirstHeader(PassThroughConstants.HTTP_CONTENT_LENGTH);
+                if (contentLengthHeader != null) {
+                    String value = contentLengthHeader.getValue();
+                    try {
+                        long length = Long.parseLong(value);
+                        if (length > validMaxMessageSize) {
+                            MessageContext requestMsgContext = TargetContext.get(conn).getRequestMsgCtx();
+                            if (requestMsgContext != null) {
+                                NHttpServerConnection sourceConn = (NHttpServerConnection) requestMsgContext
+                                        .getProperty(PassThroughConstants.PASS_THROUGH_SOURCE_CONNECTION);
+                                if (sourceConn != null) {
+                                    sourceConn.getContext()
+                                            .setAttribute(PassThroughConstants.MESSAGE_SIZE_LIMIT_EXCEEDED, true);
+                                    if (dropMessageWhenSizeLimitExceeded) {
+                                        sourceConn.getContext().setAttribute(
+                                                PassThroughConstants.DROP_MESSAGE_DUE_TO_SIZE_LIMIT_EXCEEDED, true);
+                                    }
+                                }
+                            }
+                        }
+                    } catch (NumberFormatException e) {
+                        log.warn("Invalid Content-Length '" + value +
+                                "' from backend, response status code might be inaccurate if the payload exceeds " +
+                                "valid payload size range", e);
+                    }
+                }
+            }
             boolean isError = false;
             String method = null;
             ProxyTunnelHandler tunnelHandler = (ProxyTunnelHandler) context.getAttribute(PassThroughConstants.TUNNEL_HANDLER);