sudarshan2906 opened a new issue #11225: URL: https://github.com/apache/airflow/issues/11225
Hi I am using airflow 1.10.12 with celery executor with SQS. When configuring `predefined_queues` in celery as mentioned [here](https://docs.celeryproject.org/en/latest/getting-started/brokers/sqs.html#predefined-queues) I am getting the bellow error: ``` [2020-10-01 17:37:54,498: CRITICAL/MainProcess] Unrecoverable error: AttributeError("'str' object has no attribute 'items'") Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/kombu/transport/virtual/base.py", line 921, in create_channel return self._avail_channels.pop() IndexError: pop from empty list During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/celery/worker/worker.py", line 208, in start self.blueprint.start(self) File "/usr/local/lib/python3.7/site-packages/celery/bootsteps.py", line 119, in start step.start(parent) File "/usr/local/lib/python3.7/site-packages/celery/bootsteps.py", line 369, in start return self.obj.start() File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 318, in start blueprint.start(self) File "/usr/local/lib/python3.7/site-packages/celery/bootsteps.py", line 119, in start step.start(parent) File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/connection.py", line 23, in start c.connection = c.connect() File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 405, in connect conn = self.connection_for_read(heartbeat=self.amqheartbeat) File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 412, in connection_for_read self.app.connection_for_read(heartbeat=heartbeat)) File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 439, in ensure_connected callback=maybe_shutdown, File "/usr/local/lib/python3.7/site-packages/kombu/connection.py", line 389, in ensure_connection self._ensure_connection(*args, **kwargs) File "/usr/local/lib/python3.7/site-packages/kombu/connection.py", line 445, in _ensure_connection callback, timeout=timeout File "/usr/local/lib/python3.7/site-packages/kombu/utils/functional.py", line 344, in retry_over_time return fun(*args, **kwargs) File "/usr/local/lib/python3.7/site-packages/kombu/connection.py", line 874, in _connection_factory self._connection = self._establish_connection() File "/usr/local/lib/python3.7/site-packages/kombu/connection.py", line 809, in _establish_connection conn = self.transport.establish_connection() File "/usr/local/lib/python3.7/site-packages/kombu/transport/virtual/base.py", line 941, in establish_connection self._avail_channels.append(self.create_channel(self)) File "/usr/local/lib/python3.7/site-packages/kombu/transport/virtual/base.py", line 923, in create_channel channel = self.Channel(connection) File "/usr/local/lib/python3.7/site-packages/kombu/transport/SQS.py", line 134, in __init__ self._update_queue_cache(self.queue_name_prefix) File "/usr/local/lib/python3.7/site-packages/kombu/transport/SQS.py", line 140, in _update_queue_cache for queue_name, q in self.predefined_queues.items(): AttributeError: 'str' object has no attribute 'items' ``` airflow.cfg: ``` [celery_broker_transport_options] 'predefined_queues': = { 'my-q': { 'url': 'https://ap-southeast-2.queue.amazonaws.com/123456/my-q', 'access_key_id': 'xxx', 'secret_access_key': 'xxx', } } ``` ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
