Celery Worker with Amazon SQS not working

[nathanielDuro]

Hi all. I'm using Apache Superset app and I want to utilize SQS for queueing. Celery is integrated with Apache Superset. Their documentation about setting up celery is not up to date with the latest version of Celery and their is not much information. I asked in their slack community and only referred me to the Celery docs. This is how I've setup celery config.
Celery Config:

class CeleryConfig:  # pylint: disable=too-few-public-methods
    broker_url = "sqs://"
    broker_transport_options = {
        "region": "us-west-2",
        "visibility_timeout": 700,
        "polling_interval": 5,
        'predefined_queues': {
            'celery': {
                'url': '<sqs url>',
                'access_key_id': '<access key id from iam user',
                'secret_access_key': '<secret access key generated from the iam user>',
                'backoff_policy': {1: 10, 2: 20, 3: 40, 4: 80, 5: 320, 6: 640},
                'backoff_tasks': ['svc.tasks.tasks.task1']
            }
        }
    }
    imports = ("superset.sql_lab",)
    result_backend = "db+sqlite:///celery_results.sqlite"
    worker_prefetch_multiplier = 1
    #task_acks_late = False
    worker_enable_remote_control = False
    #task_default_queue = "celery"
    task_annotations = {
        "sql_lab.get_sql_results": {"rate_limit": "100/s"},
        "email_reports.send": {
            "rate_limit": "1/s",
            "time_limit": int(timedelta(seconds=120).total_seconds()),
            "soft_time_limit": int(timedelta(seconds=150).total_seconds()),
            "ignore_result": True,
        },
    }
    beat_schedule = {
        "email_reports.schedule_hourly": {
            "task": "email_reports.schedule_hourly",
            "schedule": crontab(minute=1, hour="*"),
        },
        "reports.scheduler": {
            "task": "reports.scheduler",
            "schedule": crontab(minute="*", hour="*"),
        },
        "reports.prune_log": {
            "task": "reports.prune_log",
            "schedule": crontab(minute=0, hour=0),
        },
    }


CELERY_CONFIG = CeleryConfig  # pylint: disable=invalid-name`

This the command I use to run the celery worker.

celery --app=superset.tasks.celery_app:app worker --pool=prefork --max-tasks-per-child=128 -O fair -c 4

The SQS Queue is not receiving any messages when I'm doing something in Apache Superset app. Any help in this regard would be helpful and appreciated. Thank you!

[auvipy]

are you using kombu 5.3.4?

[nathanielDuro]

Yes. I'm also using kombu 5.3.4.
I also tried changing the config. Instead using predefined queue the class will create a new queue. And YES it does create a new queue with the name "superset_task_queue" which is defined task_default_queue = "superset_task_queue" as shown in the config below. Meaning superset celery config does have a permission to the Amazon SQS. But in terms of receiving messages, Amazon SQS doesn't receive any messages even doing some tasks inside the Apache Superset application.

class CeleryConfig:  # pylint: disable=too-few-public-methods
    broker_url = "sqs://"
    broker_transport_options = {'region': 'us-west-2'}
    imports = ("superset.sql_lab",)
    result_backend = "db+sqlite:///celery_results.sqlite"
    worker_prefetch_multiplier = 1
    worker_enable_remote_control = False
    task_default_queue = "superset_task_queue"
    task_acks_late = False
    task_annotations = {
        "sql_lab.get_sql_results": {"rate_limit": "100/s"},
        "email_reports.send": {
            "rate_limit": "1/s",
            "time_limit": int(timedelta(seconds=120).total_seconds()),
            "soft_time_limit": int(timedelta(seconds=150).total_seconds()),
            "ignore_result": True,
        },
    }
    beat_schedule = {
        "email_reports.schedule_hourly": {
            "task": "email_reports.schedule_hourly",
            "schedule": crontab(minute=1, hour="*"),
        },
        "reports.scheduler": {
            "task": "reports.scheduler",
            "schedule": crontab(minute="*", hour="*"),
        },
        "reports.prune_log": {
            "task": "reports.prune_log",
            "schedule": crontab(minute=0, hour=0),
        },
    }


CELERY_CONFIG = CeleryConfig  # pylint: disable=invalid-name

[nathanielDuro]

Hi. I tried switching broker from SQS to Redis. But I still got the same results.

ubuntu@ip-172-31-119-224:~$ systemctl status celery.service
● celery.service - Celery Service
     Loaded: loaded (/etc/systemd/system/celery.service; disabled; vendor preset: enabled)
     Active: activating (start) since Wed 2023-11-22 05:18:39 UTC; 12s ago
Cntrl PID: 12314 (celery)
      Tasks: 6 (limit: 2292)
     Memory: 453.8M
     CGroup: /system.slice/celery.service
             ├─12314 /home/ubuntu/venv/bin/python3 /home/ubuntu/venv/bin/celery --app=superset.tasks.celery_app:app worker --pool=prefork -O fair -c 4 -E
             ├─12319 /home/ubuntu/venv/bin/python3 /home/ubuntu/venv/bin/celery --app=superset.tasks.celery_app:app worker --pool=prefork -O fair -c 4 -E
             ├─12320 /home/ubuntu/venv/bin/python3 /home/ubuntu/venv/bin/celery --app=superset.tasks.celery_app:app worker --pool=prefork -O fair -c 4 -E
             ├─12321 /home/ubuntu/venv/bin/python3 /home/ubuntu/venv/bin/celery --app=superset.tasks.celery_app:app worker --pool=prefork -O fair -c 4 -E
             └─12322 /home/ubuntu/venv/bin/python3 /home/ubuntu/venv/bin/celery --app=superset.tasks.celery_app:app worker --pool=prefork -O fair -c 4 -E

Nov 22 05:33:35 ip-172-31-119-224 celery[12540]: logging was configured successfully
Nov 22 05:33:35 ip-172-31-119-224 celery[12540]: 2023-11-22 05:33:35,075:INFO:superset.utils.logging_configurator:logging was configured successfully
Nov 22 05:33:35 ip-172-31-119-224 celery[12540]: 2023-11-22 05:33:35,078:INFO:root:Configured event logger of type <class 'superset.utils.log.DBEventLogger'>
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]:  
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]:  -------------- celery@ip-172-31-119-224 v5.3.5 (emerald-rush)
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: --- ***** -----
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: -- ******* ---- Linux-5.15.0-1050-aws-x86_64-with-glibc2.29 2023-11-22 05:33:36
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: - *** --- * ---
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: - ** ---------- [config]
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: - ** ---------- .> app:         __main__:0x7fdbf79a0340
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: - ** ---------- .> transport:   rediss://:**@cache.amazonaws.com:6379/0
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: - ** ---------- .> results:     rediss://:**@cache.amazonaws.com:6379/0
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: - *** --- * --- .> concurrency: 4 (prefork)
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: -- ******* ---- .> task events: ON
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]: --- ***** -----
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]:  -------------- [queues]
Nov 22 05:33:36 ip-172-31-119-224 celery[12540]:                 .> celery           exchange=celery(direct) key=celery

[auvipy]

I don't have any knowledge about apache superset

[auvipy]

but they seem to pin to another version https://github.com/apache/superset/blob/master/requirements/base.txt#L43

[nathanielDuro]

Thank you for your reply and trying to help me out. Unfortunately it still didn't work. I need to pause on troubleshooting this for now. Because there are still other matters I need to work on. I'll update this in the future if I found a fix. Have a great day!