diff --git a/docs/user_guide/decoupled_models.md b/docs/user_guide/decoupled_models.md
index fbe6f4c298..9f8ad5b063 100644
--- a/docs/user_guide/decoupled_models.md
+++ b/docs/user_guide/decoupled_models.md
@@ -1,5 +1,5 @@
 <!--
-# Copyright 2022-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright 2022-2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions
@@ -95,7 +95,13 @@ your application should be cognizant that the callback function you registered w
 `TRITONSERVER_InferenceRequestSetResponseCallback` can be invoked any number of times,
 each time with a new response. You can take a look at [grpc_server.cc](https://github.com/triton-inference-server/server/blob/main/src/grpc/grpc_server.cc)
 
-### Knowing When a Decoupled Inference Request is Complete
+### Using Decoupled Models in Ensembles
+
+When using decoupled models within an [ensemble](ensemble_models.md), you may encounter unbounded memory growth if a decoupled model produces responses faster than downstream models can consume them. To address this, Triton provides the `max_inflight_responses` configuration field, which limits the number of concurrent inflight responses at each ensemble step.
+
+For more details and examples, see [Managing Memory Usage in Ensembles with Decoupled Models](ensemble_models.md#managing-memory-usage-in-ensembles-with-decoupled-models).
+
+## Knowing When a Decoupled Inference Request is Complete
 
 An inference request is considered complete when a response containing the
 `TRITONSERVER_RESPONSE_COMPLETE_FINAL` flag is received from a model/backend.
diff --git a/docs/user_guide/ensemble_models.md b/docs/user_guide/ensemble_models.md
index 4012ec60c7..9978e0f4b4 100644
--- a/docs/user_guide/ensemble_models.md
+++ b/docs/user_guide/ensemble_models.md
@@ -1,5 +1,5 @@
 <!--
-# Copyright 2018-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright 2018-2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions
@@ -183,6 +183,66 @@ performance, you can use
 [Model Analyzer](https://github.com/triton-inference-server/model_analyzer)
 to find the optimal model configurations.
 
+## Managing Memory Usage in Ensembles with Decoupled Models
+
+An *inflight response* is an intermediate output generated by an upstream model and held in memory until it is consumed by a downstream model within an ensemble pipeline. When an ensemble pipeline contains [decoupled models](decoupled_models.md) that produce responses faster than downstream models can process them, inflight responses accumulate internally and may cause unbounded memory growth. This commonly occurs in data preprocessing pipelines where a fast decoupled model (such as DALI, which efficiently streams and preprocesses data) feeds into a slower inference model (such as ONNX Runtime or TensorRT, which are compute-intensive and operate at a lower throughput).
+
+Consider an example ensemble model with two steps:
+1. **DALI preprocessor** (decoupled): Produces 100 preprocessed images/sec
+2. **ONNX inference model**: Consumes 10 images/sec
+
+Here, the DALI model produces responses 10× faster than the ONNX model can process them. Without backpressure, these intermediate tensors accumulate in memory, eventually leading to out-of-memory errors.
+
+The `max_inflight_responses` field in the ensemble configuration limits the number of concurrent inflight responses between ensemble steps per request.
+When this limit is reached, faster upstream models are paused (blocked) until downstream models finish processing, effectively preventing unbounded memory growth.
+
+```
+ensemble_scheduling {
+  max_inflight_responses: 16
+
+  step [
+    {
+      model_name: "dali_preprocess"
+      model_version: -1
+      input_map { key: "RAW_IMAGE", value: "IMAGE" }
+      output_map { key: "PREPROCESSED_IMAGE", value: "preprocessed" }
+    },
+    {
+      model_name: "onnx_inference"
+      model_version: -1
+      input_map { key: "INPUT", value: "preprocessed" }
+      output_map { key: "OUTPUT", value: "RESULT" }
+    }
+  ]
+}
+```
+
+**Configuration:**
+* **`max_inflight_responses: 16`**: For each ensemble request (not globally), at most 16 responses from `dali_preprocess`
+  can wait for `onnx_inference` to process. Once this per-step limit is reached, `dali_preprocess` is blocked until the downstream step completes a response.
+* **Default (`0`)**: No limit - allows unlimited inflight responses (original behavior).
+
+### When to Use This Feature
+
+Use `max_inflight_responses` when your ensemble includes:
+* **Decoupled models** that produce multiple responses per request
+* **Speed mismatch**: Upstream models significantly faster than downstream models
+* **Memory constraints**: Limited GPU/CPU memory available
+
+### Choosing the Right Value
+
+The optimal value depends on your deployment configuration, including batch size, request rate, available memory, and throughput characteristics.:
+
+* **Too low** (e.g., 1-2): The producer step is frequently blocked, underutilizing faster models
+* **Too high** (e.g., 1000+): Memory usage increases, reducing the effectiveness of backpressure
+* **Recommended**: Start with a small value and tune based on memory usage and throughput monitoring
+
+### Performance Considerations
+
+* **Zero overhead when disabled**: If `max_inflight_responses: 0` (default),
+  no synchronization overhead is incurred.
+* **Minimal overhead when enabled**: Uses a blocking/wakeup mechanism per ensemble step, where upstream models are paused ("blocked") when the inflight response limit is reached and resumed ("woken up") as downstream models consume responses. This synchronization ensures memory usage stays within bounds, though it may increase latency.
+
 ## Additional Resources
 
 You can find additional end-to-end ensemble examples in the links below:
diff --git a/qa/L0_simple_ensemble/backpressure_test_models/decoupled_producer/1/model.py b/qa/L0_simple_ensemble/backpressure_test_models/decoupled_producer/1/model.py
new file mode 100644
index 0000000000..fc8ff2f691
--- /dev/null
+++ b/qa/L0_simple_ensemble/backpressure_test_models/decoupled_producer/1/model.py
@@ -0,0 +1,54 @@
+# Copyright 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#  * Neither the name of NVIDIA CORPORATION nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
+import numpy as np
+import triton_python_backend_utils as pb_utils
+
+
+class TritonPythonModel:
+    """
+    Decoupled model that produces N responses based on input value.
+    """
+
+    def execute(self, requests):
+        for request in requests:
+            # Get input - number of responses to produce
+            in_tensor = pb_utils.get_input_tensor_by_name(request, "IN")
+            count = in_tensor.as_numpy()[0]
+
+            response_sender = request.get_response_sender()
+
+            # Produce 'count' responses, each with 0.5 as the output value
+            for i in range(count):
+                out_tensor = pb_utils.Tensor("OUT", np.array([0.5], dtype=np.float32))
+                response = pb_utils.InferenceResponse(output_tensors=[out_tensor])
+                response_sender.send(response)
+
+            # Send final flag
+            response_sender.send(flags=pb_utils.TRITONSERVER_RESPONSE_COMPLETE_FINAL)
+
+        return None
diff --git a/qa/L0_simple_ensemble/backpressure_test_models/decoupled_producer/config.pbtxt b/qa/L0_simple_ensemble/backpressure_test_models/decoupled_producer/config.pbtxt
new file mode 100644
index 0000000000..4ef3a444ce
--- /dev/null
+++ b/qa/L0_simple_ensemble/backpressure_test_models/decoupled_producer/config.pbtxt
@@ -0,0 +1,58 @@
+# Copyright 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#  * Neither the name of NVIDIA CORPORATION nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
+name: "decoupled_producer"
+backend: "python"
+max_batch_size: 0
+
+input [
+  {
+    name: "IN"
+    data_type: TYPE_INT32
+    dims: [ 1 ]
+  }
+]
+
+output [
+  {
+    name: "OUT"
+    data_type: TYPE_FP32
+    dims: [ 1 ]
+  }
+]
+
+instance_group [
+  {
+    count: 1
+    kind: KIND_CPU
+  }
+]
+
+model_transaction_policy {
+  decoupled: true
+}
+
diff --git a/qa/L0_simple_ensemble/backpressure_test_models/ensemble_disabled_max_inflight_responses/config.pbtxt b/qa/L0_simple_ensemble/backpressure_test_models/ensemble_disabled_max_inflight_responses/config.pbtxt
new file mode 100644
index 0000000000..804299cb21
--- /dev/null
+++ b/qa/L0_simple_ensemble/backpressure_test_models/ensemble_disabled_max_inflight_responses/config.pbtxt
@@ -0,0 +1,75 @@
+# Copyright 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#  * Neither the name of NVIDIA CORPORATION nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
+platform: "ensemble"
+max_batch_size: 0
+
+input [
+  {
+    name: "IN"
+    data_type: TYPE_INT32
+    dims: [ 1 ]
+  }
+]
+
+output [
+  {
+    name: "OUT"
+    data_type: TYPE_FP32
+    dims: [ 1 ]
+  }
+]
+
+ensemble_scheduling {
+  step [
+    {
+      model_name: "decoupled_producer"
+      model_version: -1
+      input_map {
+        key: "IN"
+        value: "IN"
+      }
+      output_map {
+        key: "OUT"
+        value: "intermediate"
+      }
+    },
+    {
+      model_name: "slow_consumer"
+      model_version: -1
+      input_map {
+        key: "INPUT0"
+        value: "intermediate"
+      }
+      output_map {
+        key: "OUTPUT0"
+        value: "OUT"
+      }
+    }
+  ]
+}
+
diff --git a/qa/L0_simple_ensemble/ensemble_backpressure_test.py b/qa/L0_simple_ensemble/ensemble_backpressure_test.py
new file mode 100755
index 0000000000..fc312db554
--- /dev/null
+++ b/qa/L0_simple_ensemble/ensemble_backpressure_test.py
@@ -0,0 +1,278 @@
+#!/usr/bin/env python3
+
+# Copyright 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#  * Neither the name of NVIDIA CORPORATION nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import sys
+
+sys.path.append("../common")
+
+import queue
+import time
+import unittest
+from contextlib import ExitStack
+from functools import partial
+
+import numpy as np
+import test_util as tu
+import tritonclient.grpc as grpcclient
+from tritonclient.utils import InferenceServerException
+
+SERVER_URL = "localhost:8001"
+DEFAULT_RESPONSE_TIMEOUT = 60
+EXPECTED_INFER_OUTPUT = 0.5
+MODEL_ENSEMBLE_ENABLED = "ensemble_enabled_max_inflight_responses"
+MODEL_ENSEMBLE_DISABLED = "ensemble_disabled_max_inflight_responses"
+
+
+class UserData:
+    def __init__(self):
+        self._response_queue = queue.Queue()
+
+
+def callback(user_data, result, error):
+    if error:
+        user_data._response_queue.put(error)
+    else:
+        user_data._response_queue.put(result)
+
+
+class EnsembleBackpressureTest(tu.TestResultCollector):
+    """
+    Tests for ensemble backpressure feature (max_inflight_responses).
+    """
+
+    def _prepare_infer_args(self, input_value):
+        """
+        Create InferInput/InferRequestedOutput lists
+        """
+        input_data = np.array([input_value], dtype=np.int32)
+        infer_input = [grpcclient.InferInput("IN", input_data.shape, "INT32")]
+        infer_input[0].set_data_from_numpy(input_data)
+        outputs = [grpcclient.InferRequestedOutput("OUT")]
+        return infer_input, outputs
+
+    def _collect_responses(self, user_data):
+        """
+        Collect responses from user_data until the final response flag is seen.
+        """
+        responses = []
+        while True:
+            try:
+                result = user_data._response_queue.get(timeout=DEFAULT_RESPONSE_TIMEOUT)
+            except queue.Empty:
+                self.fail(
+                    f"No response received within {DEFAULT_RESPONSE_TIMEOUT} seconds."
+                )
+
+            self.assertNotIsInstance(
+                result, Exception, f"Callback returned an exception: {result}"
+            )
+
+            response = result.get_response()
+            # Add response to list if it has data (not empty final-only response)
+            if len(response.outputs) > 0:
+                responses.append(result)
+
+            # Check if this is the final response
+            final = response.parameters.get("triton_final_response")
+            if final and final.bool_param:
+                break
+
+        return responses
+
+    def _run_inference(self, model_name, expected_count):
+        """
+        Helper function to run inference and verify responses.
+        """
+        user_data = UserData()
+        with grpcclient.InferenceServerClient(SERVER_URL) as triton_client:
+            try:
+                inputs, outputs = self._prepare_infer_args(expected_count)
+                triton_client.start_stream(callback=partial(callback, user_data))
+                triton_client.async_stream_infer(
+                    model_name=model_name, inputs=inputs, outputs=outputs
+                )
+
+                # Collect and verify responses
+                responses = self._collect_responses(user_data)
+                self.assertEqual(
+                    len(responses),
+                    expected_count,
+                    f"Expected {expected_count} responses, got {len(responses)}",
+                )
+
+                # Verify correctness of responses
+                for idx, resp in enumerate(responses):
+                    output = resp.as_numpy("OUT")
+                    self.assertAlmostEqual(
+                        output[0],
+                        EXPECTED_INFER_OUTPUT,
+                        places=5,
+                        msg=f"Response {idx} has incorrect value - {output[0]}",
+                    )
+            finally:
+                triton_client.stop_stream()
+
+    def test_backpressure_limits_inflight(self):
+        """
+        Test that max_inflight_responses correctly limits concurrent
+        responses.
+        """
+        self._run_inference(model_name=MODEL_ENSEMBLE_ENABLED, expected_count=32)
+
+    def test_backpressure_disabled(self):
+        """
+        Test that an ensemble model without max_inflight_responses parameter works correctly.
+        """
+        self._run_inference(model_name=MODEL_ENSEMBLE_DISABLED, expected_count=32)
+
+    def test_backpressure_concurrent_requests(self):
+        """
+        Test that backpressure works correctly with multiple concurrent requests.
+        Each request should have independent backpressure state.
+        """
+        num_concurrent = 8
+        expected_per_request = 8
+        user_datas = [UserData() for _ in range(num_concurrent)]
+
+        with ExitStack() as stack:
+            clients = [
+                stack.enter_context(grpcclient.InferenceServerClient(SERVER_URL))
+                for _ in range(num_concurrent)
+            ]
+
+            inputs, outputs = self._prepare_infer_args(expected_per_request)
+
+            # Start all concurrent requests
+            for i in range(num_concurrent):
+                clients[i].start_stream(callback=partial(callback, user_datas[i]))
+                clients[i].async_stream_infer(
+                    model_name=MODEL_ENSEMBLE_ENABLED, inputs=inputs, outputs=outputs
+                )
+
+            # Collect and verify responses for all requests
+            for i, ud in enumerate(user_datas):
+                responses = self._collect_responses(ud)
+                self.assertEqual(
+                    len(responses),
+                    expected_per_request,
+                    f"Request {i}: expected {expected_per_request} responses, got {len(responses)}",
+                )
+                # Verify correctness of responses
+                for idx, resp in enumerate(responses):
+                    output = resp.as_numpy("OUT")
+                    self.assertAlmostEqual(
+                        output[0],
+                        EXPECTED_INFER_OUTPUT,
+                        places=5,
+                        msg=f"Response {idx} for request {i} has incorrect value - {output[0]}",
+                    )
+
+            # Stop all streams
+            for client in clients:
+                client.stop_stream()
+
+    def test_backpressure_request_cancellation(self):
+        """
+        Test that cancellation unblocks producers waiting on backpressure and that
+        the client receives a cancellation error.
+        """
+        # Use a large count to ensure the producer gets blocked by backpressure.
+        # The model is configured with max_inflight_responses = 4.
+        input_value = 32
+        user_data = UserData()
+
+        with grpcclient.InferenceServerClient(SERVER_URL) as triton_client:
+            inputs, outputs = self._prepare_infer_args(input_value)
+            triton_client.start_stream(callback=partial(callback, user_data))
+
+            # Start the request
+            triton_client.async_stream_infer(
+                model_name=MODEL_ENSEMBLE_ENABLED, inputs=inputs, outputs=outputs
+            )
+
+            responses = []
+            try:
+                result = user_data._response_queue.get(timeout=5)
+                if isinstance(result, InferenceServerException):
+                    self.fail(f"Got error before cancellation: {result}")
+                resp = result.get_response()
+                if len(resp.outputs) > 0:
+                    responses.append(result)
+            except queue.Empty:
+                self.fail("Stream did not produce any response before cancellation.")
+
+            # Cancel the stream. This should unblock any waiting producers and result in a CANCELLED error.
+            triton_client.stop_stream(cancel_requests=True)
+
+            # Allow some time for cancellation
+            time.sleep(1)
+
+            cancellation_found = False
+            while True:
+                try:
+                    result = user_data._response_queue.get(timeout=1)
+                    if isinstance(result, InferenceServerException):
+                        self.assertEqual(
+                            result.status(),
+                            "StatusCode.CANCELLED",
+                            f"Expected CANCELLED status, got: {result.status()}",
+                        )
+                        cancellation_found = True
+                        break
+                    else:
+                        response = result.get_response()
+                        if len(response.outputs) > 0:
+                            responses.append(result)
+                        # Check for final response
+                        final = response.parameters.get("triton_final_response")
+                        if final and final.bool_param:
+                            break
+                except queue.Empty:
+                    break
+
+            # Verify the cancellation error was received
+            self.assertTrue(
+                cancellation_found,
+                "Did not receive the expected cancellation error from the server.",
+            )
+
+            # Verify we received only a partial set of responses
+            self.assertLess(
+                len(responses),
+                input_value,
+                "Expected partial responses due to cancellation, but received all of them.",
+            )
+            self.assertGreater(
+                len(responses),
+                0,
+                "Expected to receive at least one response before cancellation.",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/qa/L0_simple_ensemble/test.sh b/qa/L0_simple_ensemble/test.sh
index 927f36ea32..de74e81ee5 100755
--- a/qa/L0_simple_ensemble/test.sh
+++ b/qa/L0_simple_ensemble/test.sh
@@ -1,5 +1,5 @@
 #!/bin/bash
-# Copyright 2019-2024, NVIDIA CORPORATION. All rights reserved.
+# Copyright 2019-2025, NVIDIA CORPORATION. All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions
@@ -31,15 +31,16 @@ SIMPLE_TEST_PY=./ensemble_test.py
 
 CLIENT_LOG="./client.log"
 
+TEST_MODEL_DIR="`pwd`/models"
 TEST_RESULT_FILE='test_results.txt'
 SERVER=/opt/tritonserver/bin/tritonserver
-SERVER_ARGS="--model-repository=`pwd`/models"
+SERVER_ARGS="--model-repository=${TEST_MODEL_DIR}"
 SERVER_LOG="./inference_server.log"
 source ../common/util.sh
 
 # ensure ensemble models have version sub-directory
-mkdir -p `pwd`/models/ensemble_add_sub_int32_int32_int32/1
-mkdir -p `pwd`/models/ensemble_partial_add_sub/1
+mkdir -p ${TEST_MODEL_DIR}/ensemble_add_sub_int32_int32_int32/1
+mkdir -p ${TEST_MODEL_DIR}/ensemble_partial_add_sub/1
 
 rm -f $CLIENT_LOG $SERVER_LOG
 
@@ -146,6 +147,119 @@ set -e
 kill $SERVER_PID
 wait $SERVER_PID
 
+######## Test ensemble backpressure feature (max_inflight_responses parameter)
+MODEL_DIR="`pwd`/backpressure_test_models"
+mkdir -p ${MODEL_DIR}/ensemble_disabled_max_inflight_responses/1
+
+rm -rf ${MODEL_DIR}/slow_consumer
+mkdir -p ${MODEL_DIR}/slow_consumer/1
+cp ../python_models/ground_truth/model.py ${MODEL_DIR}/slow_consumer/1
+cp ../python_models/ground_truth/config.pbtxt ${MODEL_DIR}/slow_consumer/
+sed -i 's/name: "ground_truth"/name: "slow_consumer"/g' ${MODEL_DIR}/slow_consumer/config.pbtxt
+
+# Create ensemble with "max_inflight_responses = 4"
+rm -rf ${MODEL_DIR}/ensemble_enabled_max_inflight_responses
+mkdir -p ${MODEL_DIR}/ensemble_enabled_max_inflight_responses/1
+cp ${MODEL_DIR}/ensemble_disabled_max_inflight_responses/config.pbtxt ${MODEL_DIR}/ensemble_enabled_max_inflight_responses/
+sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n  max_inflight_responses: 4/g' \
+  ${MODEL_DIR}/ensemble_enabled_max_inflight_responses/config.pbtxt
+
+BACKPRESSURE_TEST_PY=./ensemble_backpressure_test.py
+SERVER_LOG="./ensemble_backpressure_test_server.log"
+CLIENT_LOG="./ensemble_backpressure_test_client.log"
+rm -f $SERVER_LOG $CLIENT_LOG
+
+SERVER_ARGS="--model-repository=${MODEL_DIR}"
+run_server
+if [ "$SERVER_PID" == "0" ]; then
+    echo -e "\n***\n*** Failed to start $SERVER\n***"
+    cat $SERVER_LOG
+    exit 1
+fi
+
+set +e
+python $BACKPRESSURE_TEST_PY -v >> $CLIENT_LOG 2>&1
+if [ $? -ne 0 ]; then
+    RET=1
+else
+    check_test_results $TEST_RESULT_FILE 4
+    if [ $? -ne 0 ]; then
+        cat $CLIENT_LOG
+        echo -e "\n***\n*** Test Result Verification Failed\n***"
+        RET=1
+    fi
+fi
+set -e
+
+kill $SERVER_PID
+wait $SERVER_PID
+
+set +e
+# Verify valid config was loaded successfully
+if ! grep -q "Ensemble model 'ensemble_enabled_max_inflight_responses' configured with max_inflight_responses: 4" $SERVER_LOG; then
+    echo -e "\n***\n*** FAILED: Valid model did not load successfully\n***"
+    RET=1
+fi
+set -e
+
+
+######## Test invalid value for "max_inflight_responses"
+INVALID_PARAM_MODEL_DIR="`pwd`/invalid_param_test_models"
+SERVER_ARGS="--model-repository=${INVALID_PARAM_MODEL_DIR}"
+SERVER_LOG="./invalid_max_inflight_responses_server.log"
+rm -rf $SERVER_LOG ${INVALID_PARAM_MODEL_DIR}
+
+mkdir -p ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_negative_limit/1
+mkdir -p ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_string_limit/1
+mkdir -p ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_large_value_limit/1
+cp -r ${MODEL_DIR}/decoupled_producer ${MODEL_DIR}/slow_consumer ${INVALID_PARAM_MODEL_DIR}/
+
+# max_inflight_responses = -5
+cp ${MODEL_DIR}/ensemble_disabled_max_inflight_responses/config.pbtxt ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_negative_limit/
+sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n  max_inflight_responses: -5/g' \
+  ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_negative_limit/config.pbtxt
+
+# max_inflight_responses = "invalid_value"
+cp ${MODEL_DIR}/ensemble_disabled_max_inflight_responses/config.pbtxt ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_string_limit/
+sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n  max_inflight_responses: "invalid_value"/g' \
+  ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_string_limit/config.pbtxt
+
+# max_inflight_responses = 12345678901
+cp ${MODEL_DIR}/ensemble_disabled_max_inflight_responses/config.pbtxt ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_large_value_limit/
+sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n  max_inflight_responses: 12345678901/g' \
+  ${INVALID_PARAM_MODEL_DIR}/ensemble_invalid_large_value_limit/config.pbtxt
+
+
+run_server
+if [ "$SERVER_PID" != "0" ]; then
+    echo -e "\n***\n*** FAILED: unexpected success starting $SERVER\n***"
+    kill $SERVER_PID
+    wait $SERVER_PID
+    cat $SERVER_LOG
+    RET=1
+fi
+
+set +e
+# Verify negative value caused model load failure
+if ! grep -q "Expected integer, got: -" $SERVER_LOG; then
+    echo -e "\n***\n*** FAILED: Negative value should fail model load\n***"
+    RET=1
+fi
+
+# Verify invalid string caused model load failure
+if ! grep -q 'Expected integer, got: "invalid_value"' $SERVER_LOG; then
+    echo -e "\n***\n*** FAILED: Invalid string should fail model load\n***"
+    RET=1
+fi
+
+# Verify very large value caused model load failure
+if ! grep -q "Integer out of range (12345678901)" $SERVER_LOG; then
+    echo -e "\n***\n*** FAILED: Large value should fail model load\n***"
+    RET=1
+fi
+set -e
+
+
 if [ $RET -eq 0 ]; then
   echo -e "\n***\n*** Test Passed\n***"
 else