return an empty list on SQS internal error. (#1059)

* return empty list on SQS internal error.

* add test for on_list_ready error codes

2 files changed, 28 insertions, 3 deletions

diff --git a/kombu/asynchronous/aws/connection.py b/kombu/asynchronous/aws/connection.py
index e88612bb..8d6d4353 100644
--- a/kombu/asynchronous/aws/connection.py
+++ b/kombu/asynchronous/aws/connection.py
@@ -173,6 +173,7 @@ class AsyncAWSQueryConnection(AsyncConnection):
     STATUS_CODE_OK = 200
     STATUS_CODE_REQUEST_TIMEOUT = 408
     STATUS_CODE_NETWORK_CONNECT_TIMEOUT_ERROR = 599
+    STATUS_CODE_INTERNAL_ERROR = 500
     STATUS_CODES_TIMEOUT = (
         STATUS_CODE_REQUEST_TIMEOUT,
         STATUS_CODE_NETWORK_CONNECT_TIMEOUT_ERROR
@@ -237,9 +238,13 @@ class AsyncAWSQueryConnection(AsyncConnection):
                 service_model.operation_model(operation), response.response
             )
             return parsed
-        elif response.status in self.STATUS_CODES_TIMEOUT:
-            # When the server returns a timeout, the response is interpreted
-            # as an empty list. This prevents hanging the Celery worker.
+        elif (
+            response.status in self.STATUS_CODES_TIMEOUT or
+            response.status == self.STATUS_CODE_INTERNAL_ERROR
+        ):
+            # When the server returns a timeout or internal error,
+            # the response is interpreted as an empty list.
+            # This prevents hanging the Celery worker.
             return []
         else:
             raise self._for_status(response, response.read())
diff --git a/t/unit/asynchronous/aws/test_connection.py b/t/unit/asynchronous/aws/test_connection.py
index b378c020..50fe3c9c 100644
--- a/t/unit/asynchronous/aws/test_connection.py
+++ b/t/unit/asynchronous/aws/test_connection.py
@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, unicode_literals
 
+import pytest
+
 from contextlib import contextmanager
 
 from case import Mock
@@ -246,6 +248,24 @@ class test_AsyncAWSQueryConnection(AWSCase):
         params = parse_qs(parsed.query)
         assert 'Action' not in params
 
+    @pytest.mark.parametrize('error_status_code', [
+        AsyncAWSQueryConnection.STATUS_CODE_REQUEST_TIMEOUT,
+        AsyncAWSQueryConnection.STATUS_CODE_NETWORK_CONNECT_TIMEOUT_ERROR,
+        AsyncAWSQueryConnection.STATUS_CODE_INTERNAL_ERROR
+    ])
+    def test_on_list_ready_error_response(self, error_status_code):
+        mocked_response_error = self.Response(
+            error_status_code,
+            "error_status_code"
+        )
+        result = self.x._on_list_ready(
+            "parent",
+            "markers",
+            "operation",
+            mocked_response_error
+        )
+        assert result == []
+
     def Response(self, status, body):
         r = Mock(name='response')
         r.status = status