AIP-82: Add RedisPubSubMessageQueueProvider

[Jasperora]

Why

Related issue: #52712.
Related PR: #52917.

What

Introduce support for Redis as a MessageQueueProvider, extending Event-Driven Scheduling.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[potiuk]

Question. Would it be possible to add an integration test for that one? We already have integration tests with redis run in our CI and for me it would be cool if we can run the tests of message queue with "real" readis rather than based on unit tests.

Some pointers for you:

• https://github.com/apache/airflow/blob/main/contributing-docs/testing/integration_tests.rst
• https://github.com/apache/airflow/tree/main/providers/redis/tests/integration. -> where all redis integration tests are

[Jasperora]

Thanks for pointing out! I'll give it a try.

[vincbeck]

One nit but overall it looks good! Very nice to see other providers onboarding 🚀 I agree with Jarek's comment too, if we can have integration tests, we should do it :)

[Jasperora]

Hi @jason810496, during the period of adding integration test, I found that a bug exists in my unit test.
From pytest importorskip doc, the parameter should be module name instead of class name.
I tried using class name (airflow.providers.common.messaging.providers.base_provider.BaseMessageQueueProvider) in redis, and the test would be skipped since the class cannot be imported.
I noticed that same issue exists in Apache Kafka and Amazon SQS.

I could address the issue if my understanding is correct. However, I would appreciate hearing your suggestions first. Thank you!

[vincbeck]

Hi @jason810496, during the period of adding integration test, I found that a bug exists in my unit test. From pytest importorskip doc, the parameter should be module name instead of class name. I tried using class name (airflow.providers.common.messaging.providers.base_provider.BaseMessageQueueProvider) in redis, and the test would be skipped since the class cannot be imported. I noticed that same issue exists in Apache Kafka and Amazon SQS.

I could address the issue if my understanding is correct. However, I would appreciate hearing your suggestions first. Thank you!

You're entirely correct! Thanks for the catch! I created #54135 to fix it in SQS and Kafka

[jason810496]

Thanks for the update.
I just noticed an issue regarding URI usage, and we might need to have further discussion on this.