Try to fix async requests getting stuck (#466)

Author: squeakymouse

model-engine/model_engine_server/core/celery/app.py

@@ -476,6 +476,11 @@ def _get_broker_endpoint_and_transport_options(
         # Going to try this with defaults first.
         out_broker_transport_options["region"] = os.environ.get("AWS_REGION", "us-west-2")
 
+        # changing wait_time_seconds from the default of 10 based on https://github.com/celery/celery/discussions/7283
+        # goal is to prevent async requests from being stuck in pending when workers die; the hypothesis is that this is caused by SQS long polling
+        out_broker_transport_options["wait_time_seconds"] = 0
+        out_broker_transport_options["polling_interval"] = 5
+
         # NOTE: The endpoints should ideally use predefined queues. However, the sender probably needs the flexibility
         # of not requiring predefined queues.
         # assert (

model-engine/model_engine_server/inference/forwarding/echo_server.py

@@ -3,6 +3,7 @@
 """
 import argparse
 import subprocess
+import time
 
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
@@ -19,7 +20,10 @@ def healthcheck():
 
 @app.post("/predict")
 async def predict(request: Request):
-    return await request.json()
+    dictionary = await request.json()
+    if "delay" in dictionary:
+        time.sleep(dictionary["delay"])
+    return dictionary
 
 
 @app.post("/predict500")