Draft Common Message Queue

[vikramkoka]

Here is a very early draft PR to introduce and socialize the concept of a "common message queue" abstraction similar to the "Common SQL" and "Common IO" abstractions in Airflow.

This will be a provider package similar to those and is intended to be an abstraction over Apache Kafka, Amazon SQL, and Google PubSub to begin with. It can then be expanded to other messaging systems based on community adoption.

The initial goal with this is to provide a simple abstraction for integrating Event Driven Scheduling coming with Airflow 3 to message notification systems such as Kafka, currently being used to publish data availability.

At this stage, this is very much a WIP draft intended to solicit input from the community.

[vincbeck]

Implementation wise, here is my thinking. I am starting by MessageQueueTrigger.

Given msg_queue, MessageQueueTrigger needs to figure which hook it will use to poll/pop a message from the queue. Example: if msg_queue.starts_with("https://sqs."): hook = SqsHook(...). Then we can use the hook to retrieve the message. The hook will contain the logic for each provider (AWS, Google, Kafka, ...). This means, this new provider will have a dependency with all these providers. Do you think this is an issue? Did you have something else in mind?

[vikramkoka]

Implementation wise, here is my thinking. I am starting by MessageQueueTrigger.

Given msg_queue, MessageQueueTrigger needs to figure which hook it will use to poll/pop a message from the queue. Example: if msg_queue.starts_with("https://sqs."): hook = SqsHook(...). Then we can use the hook to retrieve the message. The hook will contain the logic for each provider (AWS, Google, Kafka, ...). This means, this new provider will have a dependency with all these providers. Do you think this is an issue? Did you have something else in mind?

You are right Vincent. I did think about the "Composition vs. Inheritance" approach tradeoff.

The composition style interface as defined here is easier for the DAG author, but more maintenance for us.
I talked about this with Ash and Jed as well and because of the underlying plumbing already present in Airflow for finding connections, et al, this seemed reasonable as an approach in order to make the end-user experience better.

[jscheffl]

In general looks good. Some more nit that I would have on the Python code/Interface but we can leave this until it is in real review.

Would be great to add an example DAG as well for the showcase.

[vincbeck]

I am iterating on that PR but the new provider is not recognized. I get:

ModuleNotFoundError: No module named 'airflow.providers.common.msgq'

With the new restructure, what is the process to add a new provider? Do I just need to create provider.yaml and pyproject.toml and it will be automatically detected/indexed? @potiuk

[vincbeck]

I updated the PR. I focused only on the trigger side. Please let me know if this is what you had in mind in terms of implementation regarding the trigger. I really see it as a proxy of the provider triggers. I could not test it because the new provider is not recognized but once that solved I should be able to test it.

[potiuk]

I am iterating on that PR but the new provider is not recognized. I get:

ModuleNotFoundError: No module named 'airflow.providers.common.msgq'

With the new restructure, what is the process to add a new provider? Do I just need to create provider.yaml and pyproject.toml and it will be automatically detected/indexed? @potiuk

You need to look at the main pyproject.toml and add the provider in the same way as others (there are few places). Then uv sync or image build should work after that.

And yes I updated https://github.com/apache/airflow/blob/main/providers/MANAGING_PROVIDERS_LIFECYCLE.rst#creating-a-new-community-provider - with the new structure and how to add a new provider, but that part is likely missing so after you figure it out, PRs there are most welcome.

BTW. It will likely slightly change in the future as we will move airflow-core and others, but still it would be great to keep it updated.

[potiuk]

Generally @vincbeck -> look at everything below [dependency-groups] in the root pyproject.toml - the provider should be added in all those places.

[vincbeck]

Thank you :D

[vincbeck]

I am not sure what is going on 👀 check-providers-subpackages-init-file-exist is failing because it detects missing __init__ file in /home/runner/work/airflow/airflow/providers/common/messaging/src/airflow/__init__.py but this is the old provider directory. It no longer exist.

Added missing __init__.py file: /home/runner/work/airflow/airflow/providers/common/messaging/src/airflow/__init__.py
Added missing __init__.py file: /home/runner/work/airflow/airflow/providers/common/messaging/src/airflow/providers/__init__.py
Added missing __init__.py file: /home/runner/work/airflow/airflow/providers/common/messaging/src/airflow/providers/common/__init__.py
Added missing path extension to __init__.py file /home/runner/work/airflow/airflow/providers/common/messaging/src/airflow/__init__.py

[jedcunningham]

This doesn't match the 3.0+ requirement.

[jedcunningham]

Suggested change

	task = EmptyOperator(task_id="task")
	chain(task)
	EmptyOperator(task_id="task")

No need for a var and chain if there is a single task.

[jedcunningham]

This feels a little weird being in the test dir?