Task sync fails when using celery executor with AWS SQS broker.

[r4um]

Apache Airflow version

Other Airflow 2 version

What happened

The change in #3830 introduced multiprocessing while syncing task status, when using celery executor with AWS SQS broker, the dag runs fail with the following backtrace in the scheduler logs.

[2022-02-01 12:52:52,595] {celery_executor.py:299} ERROR - Error sending Celery task: Unable to parse response (no element found: line 1, column 0), invalid XML received. Further retries may succeed:
b''
Celery Task ID: TaskInstanceKey(dag_id='shared-dev_jobflow-airflow_k8s_job_multiple_tasks', task_id='t4', run_id='manual__2022-02-01T12:52:40.329850+00:00', try_number=1)
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/botocore/parsers.py", line 480, in _parse_xml_string_to_dom
    root = parser.close()
  File "<string>", line None
xml.etree.ElementTree.ParseError: no element found: line 1, column 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/executors/celery_executor.py", line 172, in send_task_to_executor
    result = task_to_run.apply_async(args=[command], queue=queue)
  File "/usr/local/lib/python3.7/site-packages/celery/app/task.py", line 579, in apply_async
    **options
  File "/usr/local/lib/python3.7/site-packages/celery/app/base.py", line 788, in send_task
    amqp.send_task_message(P, name, message, **options)
  File "/usr/local/lib/python3.7/site-packages/celery/app/amqp.py", line 519, in send_task_message
    **properties
  File "/usr/local/lib/python3.7/site-packages/kombu/messaging.py", line 180, in publish
    exchange_name, declare, timeout
  File "/usr/local/lib/python3.7/site-packages/kombu/connection.py", line 524, in _ensured
    return fun(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/kombu/messaging.py", line 193, in _publish
    [maybe_declare(entity) for entity in declare]
  File "/usr/local/lib/python3.7/site-packages/kombu/messaging.py", line 193, in <listcomp>
    [maybe_declare(entity) for entity in declare]
  File "/usr/local/lib/python3.7/site-packages/kombu/messaging.py", line 99, in maybe_declare
    return maybe_declare(entity, self.channel, retry, **retry_policy)
  File "/usr/local/lib/python3.7/site-packages/kombu/common.py", line 110, in maybe_declare
    return _maybe_declare(entity, channel)
  File "/usr/local/lib/python3.7/site-packages/kombu/common.py", line 150, in _maybe_declare
    entity.declare(channel=channel)
  File "/usr/local/lib/python3.7/site-packages/kombu/entity.py", line 606, in declare
    self._create_queue(nowait=nowait, channel=channel)
  File "/usr/local/lib/python3.7/site-packages/kombu/entity.py", line 615, in _create_queue
    self.queue_declare(nowait=nowait, passive=False, channel=channel)
  File "/usr/local/lib/python3.7/site-packages/kombu/entity.py", line 650, in queue_declare
    nowait=nowait,
  File "/usr/local/lib/python3.7/site-packages/kombu/transport/virtual/base.py", line 528, in queue_declare
    return queue_declare_ok_t(queue, self._size(queue), 0)
  File "/usr/local/lib/python3.7/site-packages/kombu/transport/SQS.py", line 607, in _size
    AttributeNames=['ApproximateNumberOfMessages'])
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 388, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 695, in _make_api_call
    operation_model, request_dict, request_context)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 714, in _make_request
    return self._endpoint.make_request(operation_model, request_dict)
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 102, in make_request
    return self._send_request(request_dict, operation_model)
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 135, in _send_request
    request, operation_model, context)
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 166, in _get_response
    request, operation_model)
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 217, in _do_get_response
    response_dict, operation_model.output_shape)
  File "/usr/local/lib/python3.7/site-packages/botocore/parsers.py", line 245, in parse
    parsed = self._do_parse(response, shape)
  File "/usr/local/lib/python3.7/site-packages/botocore/parsers.py", line 551, in _do_parse
    return self._parse_body_as_xml(response, shape, inject_metadata=True)
  File "/usr/local/lib/python3.7/site-packages/botocore/parsers.py", line 555, in _parse_body_as_xml
    root = self._parse_xml_string_to_dom(xml_contents)
  File "/usr/local/lib/python3.7/site-packages/botocore/parsers.py", line 485, in _parse_xml_string_to_dom
    (e, xml_string))
botocore.parsers.ResponseParserError: Unable to parse response (no element found: line 1, column 0), invalid XML received. Further retries may succeed: 
b''

[2022-02-01 12:52:52,821] {scheduler_job.py:572} ERROR - Executor reports task instance <TaskInstance: shared-dev_jobflow-airflow_k8s_job_multiple_tasks.t4 manual__2022-02-01T12:52:40.329850+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2022-02-01 12:52:52,825] {taskinstance.py:1705} ERROR - Executor reports task instance <TaskInstance: shared-dev_jobflow-airflow_k8s_job_multiple_tasks.t4 manual__2022-02-01T12:52:40.329850+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2022-02-01 12:52:52,834] {taskinstance.py:1280} INFO - Marking task as FAILED. dag_id=shared-dev_jobflow-airflow_k8s_job_multiple_tasks, task_id=t4, execution_date=20220201T125240, start_date=, end_date=20220201T125252

This is likely due to boto3 session and resource objects not being thread safe[1]. Setting sync_parallelism to 1, the problem does not occur.

[1] https://boto3.amazonaws.com/v1/documentation/api/1.14.31/guide/session.html#multithreading-or-multiprocessing-with-sessions

What you think should happen instead

No response

How to reproduce

Use celery executor and AWS SQS as the celery broker. Launch a dag with multiple tasks and dependencies among them. e.g 5 tasks with dependencies as

t3 << [t1, t2]
t4 << [t3]
t5 << [t3]

Task type doesn't matter. In our case, it is the KubernetesPodOperator.

Operating System

CentOS Linux 7 (Core)

Versions of Apache Airflow Providers

airflow version 2.2.2
Python version 3.7.9
Using airflow constraints for 2.2.2 while installing packages.

Deployment

Other

Deployment details

Deployed under AWS EKS via containers.

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[potiuk]

AWS SQS broker is not officially supported by Airflow. Only RabbitMQ and Redis are. We had an extensive discussion about making it supported and documented in #24019. There were voices in the discussion that SQS has many quirks and that there are many edge-cases and that SQS does not have the same level of support and functionality as RabbitMQ and Redis in Celery. And we are not going to actvely work on it, so if you want to continue using SQS you are basically on your own.

If you want to continue using it I think the only option is that you provide a PR wihich fixes this problem, add automated tests and documentation on how to configure it. Without it we won't be able to respond to issues and any issue with SQS is not considered a bug but at most a discussion. You can add PRs that you will tests locally that fix your SQS installation, but until tests are there covering integration with SQS we will not consider any issue with it as a "bug" and we will not even attempt to make it continue to work (becasue we do not have any tests for SQS, we cannot automatically test that SQS continues to work after any change like that).

This is just to set expectations right - if you want to invest your time and effort into making SQS work - it's ok, you are free to do so. If other people will help with it - that's also fine. If you want to discuss what needs to be done etc. - fine as well,. but this is not a bug and there is no active work currently to make SQS work.

[luigi0229]

I'm having the same issue, and i'm not using SQS. I have remote logging setup to an S3 bucket, and the scheduler is able to write the task logs there just fine, but the server isn't able to bubble them up to the UI, and i get a very similar message. I'm running the barebones docker deployment on kube, no special settings other than remote logging. Only have one dag that prints a string, and i get this error. Any guidance or feedback you can provide will be appreciated.

[2023-11-17T07:05:53.352+0000] {app.py:1744} ERROR - Exception on /api/v1/dags/prepare_test_dag/dagRuns/manual__2023-11-17T06:47:08.743392+00:00/taskInstances/prepare_tests_task/logs/1 [GET]
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/parsers.py", line 505, in _parse_xml_string_to_dom
    parser.feed(xml_string)
xml.etree.ElementTree.ParseError: not well-formed (invalid token): line 1, column 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/flask/app.py", line 2529, in wsgi_app
    response = self.full_dispatch_request()
  File "/home/airflow/.local/lib/python3.8/site-packages/flask/app.py", line 1825, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home/airflow/.local/lib/python3.8/site-packages/flask/app.py", line 1823, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home/airflow/.local/lib/python3.8/site-packages/flask/app.py", line 1799, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
  File "/home/airflow/.local/lib/python3.8/site-packages/connexion/decorators/decorator.py", line 68, in wrapper
    response = function(request)
  File "/home/airflow/.local/lib/python3.8/site-packages/connexion/decorators/uri_parsing.py", line 149, in wrapper
    response = function(request)
  File "/home/airflow/.local/lib/python3.8/site-packages/connexion/decorators/validation.py", line 399, in wrapper
    return function(request)
  File "/home/airflow/.local/lib/python3.8/site-packages/connexion/decorators/response.py", line 113, in wrapper
    return _wrapper(request, response)
  File "/home/airflow/.local/lib/python3.8/site-packages/connexion/decorators/response.py", line 90, in _wrapper
    self.operation.api.get_connexion_response(response, self.mimetype)
  File "/home/airflow/.local/lib/python3.8/site-packages/connexion/apis/abstract.py", line 366, in get_connexion_response
    return cls._framework_to_connexion_response(response=response, mimetype=mimetype)
  File "/home/airflow/.local/lib/python3.8/site-packages/connexion/apis/flask_api.py", line 165, in _framework_to_connexion_response
    body=response.get_data() if not response.direct_passthrough else None,
  File "/home/airflow/.local/lib/python3.8/site-packages/werkzeug/wrappers/response.py", line 314, in get_data
    self._ensure_sequence()
  File "/home/airflow/.local/lib/python3.8/site-packages/werkzeug/wrappers/response.py", line 376, in _ensure_sequence
    self.make_sequence()
  File "/home/airflow/.local/lib/python3.8/site-packages/werkzeug/wrappers/response.py", line 391, in make_sequence
    self.response = list(self.iter_encoded())
  File "/home/airflow/.local/lib/python3.8/site-packages/werkzeug/wrappers/response.py", line 50, in _iter_encoded
    for item in iterable:
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/log/log_reader.py", line 87, in read_log_stream
    logs, metadata = self.read_log_chunks(ti, current_try_number, metadata)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/log/log_reader.py", line 64, in read_log_chunks
    logs, metadatas = self.log_handler.read(ti, try_number, metadata=metadata)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/log/file_task_handler.py", line 413, in read
    log, out_metadata = self._read(task_instance, try_number_element, metadata)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/amazon/aws/log/s3_task_handler.py", line 149, in _read
    return super()._read(ti, try_number, metadata)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/log/file_task_handler.py", line 313, in _read
    remote_messages, remote_logs = self._read_remote_logs(ti, try_number, metadata)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/amazon/aws/log/s3_task_handler.py", line 123, in _read_remote_logs
    keys = self.hook.list_keys(bucket_name=bucket, prefix=prefix)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/amazon/aws/hooks/s3.py", line 89, in wrapper
    return func(*bound_args.args, **bound_args.kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/amazon/aws/hooks/s3.py", line 820, in list_keys
    for page in response:
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/paginate.py", line 269, in __iter__
    response = self._make_request(current_kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/paginate.py", line 357, in _make_request
    return self._method(**current_kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/client.py", line 535, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/client.py", line 963, in _make_api_call
    http, parsed_response = self._make_request(
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/client.py", line 986, in _make_request
    return self._endpoint.make_request(operation_model, request_dict)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/endpoint.py", line 119, in make_request
    return self._send_request(request_dict, operation_model)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/endpoint.py", line 199, in _send_request
    success_response, exception = self._get_response(
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/endpoint.py", line 241, in _get_response
    success_response, exception = self._do_get_response(
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/endpoint.py", line 308, in _do_get_response
    parsed_response = parser.parse(
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/parsers.py", line 252, in parse
    parsed = self._do_parse(response, shape)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/parsers.py", line 883, in _do_parse
    self._add_modeled_parse(response, shape, final_parsed)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/parsers.py", line 893, in _add_modeled_parse
    self._parse_payload(response, shape, member_shapes, final_parsed)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/parsers.py", line 934, in _parse_payload
    original_parsed = self._initial_body_parse(response['body'])
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/parsers.py", line 1033, in _initial_body_parse
    return self._parse_xml_string_to_dom(xml_string)
  File "/home/airflow/.local/lib/python3.8/site-packages/botocore/parsers.py", line 508, in _parse_xml_string_to_dom
    raise ResponseParserError(
botocore.parsers.ResponseParserError: Unable to parse response (not well-formed (invalid token): line 1, column 0), invalid XML received. Further retries may succeed:
b'{"bucket":"airflow_remote_logs","keyCount":1,"maxKeys":100,"versions":false,"prefix":"dag_id=prepare_test_dag\\/run_id=manual__2023-11-17T06:47:08.743392+00:00\\/task_id=prepare_tests_task\\/attempt=1.log","truncated":false,"results":[{"size":2361,"etag":"\\"93B2FB522B02F5C7C298902AD4299074\\"","lastModified":1700203636054,"version":"253baf60-8515-11ee-bcf5-d8c497a42413","key":"dag_id=prepare_test_dag\\/run_id=manual__2023-11-17T06:47:08.743392+00:00\\/task_id=prepare_tests_task\\/attempt=1.log"}]}'

[Taragolis]

I'm having the same issue, and i'm not using SQS

All signs here that is not the same issue

[potiuk]

Yes. Generally in such cases, you should ratehr open a new issue @r4um and explain all the details and circumstances there. You can link to that discussion/issue there, but if we find it's a duplicate - we mark it so,

[Taragolis]

I've just add that output of pip freeze would be nice, I guess this might be yet another dependency conflicts. Especially if we take in account recent changes in botocore