You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "vikramkoka (via GitHub)" <gi...@apache.org> on 2025/02/12 18:21:29 UTC
[PR] Draft Common Message Queue [airflow]
vikramkoka opened a new pull request, #46694:
URL: https://github.com/apache/airflow/pull/46694
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.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2684548277
> 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.
No. That's new directory. And yes the recent change in the way how we treat namespace packages (in order to make vscode package inspection and navigation works) actually made us to change the approach to use legacy rather than implicit packages for all providers - see https://github.com/apache/airflow/pull/47064 -> so yes, those `__init__.py` files are needed and the static check is very right about adding them - they should be added and committed.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2684663321
I wrote a more detailed message about it - while the whole packaging (and re-packaging :) ) is not complete - the `__init__.py` / legacy namespace is worth a message on it's own to not get people confused https://lists.apache.org/thread/8sh0ksodcd6kr7tkpf7dgbshlom36yvs
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vikramkoka (via GitHub)" <gi...@apache.org>.
vikramkoka commented on code in PR #46694:
URL: https://github.com/apache/airflow/pull/46694#discussion_r1955322629
##########
providers/common/msgq/README.rst:
##########
@@ -0,0 +1,101 @@
+
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ .. NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE OVERWRITTEN!
+
+ .. IF YOU WANT TO MODIFY TEMPLATE FOR THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+ `PROVIDER_README_TEMPLATE.rst.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+
+
+Package ``apache-airflow-providers-common-msgq``
+
+Release: ``0.1.0``
+
+
+
+Provider package
+----------------
+
+This is a provider package for ``common.msgq`` provider. All classes for this provider package
+are in ``airflow.providers.common.msgq`` python package.
+
+This provider package is intended to serve as a common abstraction on top of popular message queue
+providers such as Apache Kafka, Amazon SQS, and Google PubSub.
+
+The expectation is that a common provider abstraction for message queues is especially useful for
+Event Driven Scheduling, which is being introduced as part of Airflow 3.0. Based on conversations
+with users of Apache Airflow, these Events for publishing of Data Assets are very often broadcast
+over a publish and subscribe mechanism. The underlying technology used for this publish and subscribe
+mechanism varies by environment, but Apache Kafka, Amazon SQS, and Google PubSub are commonly used.
+
+
+Expected usage would be something on the lines below::
+
+ trigger = = MsgQueueSensorTrigger(msg_queue="https://sqs.us-east-1.amazonaws.com/722404908466/Test")
Review Comment:
Yes, you are right. Thanks for pointing that out.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2685089513
I see, thanks for the details, I'll add them
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2688909097
I dont understand what Sphinx is complaining about: `/opt/airflow/docs/apache-airflow-providers-common-messaging/_api/airflow/providers/common/messaging/providers/base_provider/index.rst: WARNING: document isn't included in any
toctree`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2688995566
> I dont understand what Sphinx is complaining about: `/opt/airflow/docs/apache-airflow-providers-common-messaging/_api/airflow/providers/common/messaging/providers/base_provider/index.rst: WARNING: document isn't included in any toctree`
Sphinx is - as usual - speaking riddles :) . It means that there is index.rst file generated by autoapi (base_provider module) and that index is not mentioned anywhere. This means that you have to add it to some index file and refer to it - because otherwise that file is not reachable from anywhere. And it means some documentation is missing to explain what it is - usually a reference doc (see in other providers)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2689163360
> It is already there: https://github.com/apache/airflow/pull/46694/files#diff-f54feaaca8fd8ecfad946ef2cc5b389e082660ba53d305843bab44f5a014d582R36
So if the index to the "__init__.py" is not linked (and does not need to be linked) from anywhere - it should be excluded in "docs/conf.py" explicitly for provider package builds.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #46694:
URL: https://github.com/apache/airflow/pull/46694#discussion_r1971990016
##########
providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py:
##########
Review Comment:
Fair enough, I guess I spend significantly more time looking at core example dags than provider ones.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #46694:
URL: https://github.com/apache/airflow/pull/46694#discussion_r1971685710
##########
providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py:
##########
Review Comment:
Why? It is a system test. All system tests are in tests directory
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #46694:
URL: https://github.com/apache/airflow/pull/46694
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2689003751
Likely some documentation about base_provider should be added here https://github.com/apache/airflow/blob/f4fd6fd5ae45cd20924149aa0201d2da08a63112/providers/common/messaging/docs/providers.rst
> You need to extend base_provider ble ble ble....
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2690858656
All green :) I also tested it manually and triggered few DAGs using `MessageQueueTrigger` and it works nicely. I really like the user experience, users do not have to worry about the actual implementation of the queue provider
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2689175055
> docs/conf.py
Yeah I did not want to do that but I think I'll do that, I really cannot find a solution. What I do not understand is there are a lot of modules in others providers that are not documented (for good reasons like the module `utils` in `amazon`) but Sphinx is not complaining about that. I do not know why it does like this new provider.
But anyway, thanks for your help and I'll add these 2 paths to `exclude_patterns`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2682867740
> 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.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2689184985
> Yeah I did not want to do that but I think I'll do that, I really cannot find a solution. What I do not understand is there are a lot of modules in others providers that are not documented (for good reasons like the module utils in amazon) but Sphinx is not complaining about that. I do not know why it does like this new provider.
They are likely referred to in class docstrings or others. The thing is that if your class or module is not referred ANYWHERE - the only way you can reach it is by direct URL. And this is what Sphinx complains about.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2689200970
> > Yeah I did not want to do that but I think I'll do that, I really cannot find a solution. What I do not understand is there are a lot of modules in others providers that are not documented (for good reasons like the module utils in amazon) but Sphinx is not complaining about that. I do not know why it does like this new provider.
>
> They are likely referred to in class docstrings or others. The thing is that if your class or module is not referred ANYWHERE - the only way you can reach it is by direct URL. And this is what Sphinx complains about.
It is probably that! Makes a bit more sense in all that Sphinx dialect :) Thanks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2675205990
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
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2682907593
Thank you :D
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #46694:
URL: https://github.com/apache/airflow/pull/46694#discussion_r1970861110
##########
providers/common/messaging/docs/changelog.rst:
##########
@@ -0,0 +1,35 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. NOTE TO CONTRIBUTORS:
+ Please, only add notes to the Changelog just below the "Changelog" header when there are some breaking changes
+ and you want to add an explanation to the users on how they are supposed to deal with them.
+ The changelog is updated and maintained semi-automatically by release manager.
+
+``apache-airflow-providers-common-messaging``
+
+Changelog
+---------
+
+1.0.0
+.....
+
+.. note::
+ This release of provider is only available for Airflow 2.9+ as explained in the
Review Comment:
This doesn't match the 3.0+ requirement.
##########
providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.models.baseoperator import chain
+from airflow.models.dag import DAG
+from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk.definitions.asset import Asset, AssetWatcher
+
+# [START howto_trigger_message_queue]
+trigger = MessageQueueTrigger(queue="https://sqs.us-east-1.amazonaws.com/0123456789/my-queue")
+asset = Asset("sqs_asset", watchers=[AssetWatcher(name="sqs_asset_watcher", trigger=trigger)])
+
+with DAG(
+ dag_id="example_msgq_watcher",
+ schedule=[asset],
+ catchup=False,
+):
+ task = EmptyOperator(task_id="task")
+
+ chain(task)
Review Comment:
```suggestion
EmptyOperator(task_id="task")
```
No need for a var and chain if there is a single task.
##########
providers/common/messaging/tests/system/common/messaging/example_message_queue_trigger.py:
##########
Review Comment:
This feels a little weird being in the test dir?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #46694:
URL: https://github.com/apache/airflow/pull/46694#discussion_r1957119478
##########
providers/common/msgq/README.rst:
##########
@@ -0,0 +1,101 @@
+
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ .. NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE OVERWRITTEN!
+
+ .. IF YOU WANT TO MODIFY TEMPLATE FOR THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+ `PROVIDER_README_TEMPLATE.rst.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+
+
+Package ``apache-airflow-providers-common-msgq``
+
+Release: ``0.1.0``
+
+
+
+Provider package
+----------------
+
+This is a provider package for ``common.msgq`` provider. All classes for this provider package
+are in ``airflow.providers.common.msgq`` python package.
+
+This provider package is intended to serve as a common abstraction on top of popular message queue
+providers such as Apache Kafka, Amazon SQS, and Google PubSub.
+
+The expectation is that a common provider abstraction for message queues is especially useful for
+Event Driven Scheduling, which is being introduced as part of Airflow 3.0. Based on conversations
+with users of Apache Airflow, these Events for publishing of Data Assets are very often broadcast
+over a publish and subscribe mechanism. The underlying technology used for this publish and subscribe
+mechanism varies by environment, but Apache Kafka, Amazon SQS, and Google PubSub are commonly used.
+
+
+Expected usage would be something on the lines below::
+
+ trigger = = MsgQueueTrigger(msg_queue="https://sqs.us-east-1.amazonaws.com/722404908466/Test")
Review Comment:
Typo?
```suggestion
trigger = MsgQueueTrigger(msg_queue="https://sqs.us-east-1.amazonaws.com/722404908466/Test")
```
##########
providers/common/msgq/src/airflow/providers/common/msgq/operators/msg_queue.py:
##########
@@ -0,0 +1,109 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import re
+from collections.abc import Sequence
+from functools import cached_property
+from typing import Any, NoReturn
+
+from airflow.exceptions import AirflowException, AirflowFailException
+from airflow.hooks.base import BaseHook
+from airflow.models import BaseOperator
+from airflow.providers.common.msgq.hooks.msg_queue import MsgQueueHook
+
+_PROVIDERS_MATCHER = re.compile(r"airflow\.providers\.(.*?)\.hooks.*")
+
+_MIN_SUPPORTED_PROVIDERS_VERSION = {
+ "amazon": "4.1.0",
+ "apache.kafka": "2.1.0",
+ "google": "8.2.0",
+}
Review Comment:
I am with Vincent here - we should not have a reverse dependency to the providers using this feature.
##########
providers/common/msgq/README.rst:
##########
@@ -0,0 +1,101 @@
+
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ .. NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE OVERWRITTEN!
+
+ .. IF YOU WANT TO MODIFY TEMPLATE FOR THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+ `PROVIDER_README_TEMPLATE.rst.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+
+
+Package ``apache-airflow-providers-common-msgq``
+
+Release: ``0.1.0``
+
+
+
+Provider package
+----------------
+
+This is a provider package for ``common.msgq`` provider. All classes for this provider package
+are in ``airflow.providers.common.msgq`` python package.
+
+This provider package is intended to serve as a common abstraction on top of popular message queue
+providers such as Apache Kafka, Amazon SQS, and Google PubSub.
+
+The expectation is that a common provider abstraction for message queues is especially useful for
+Event Driven Scheduling, which is being introduced as part of Airflow 3.0. Based on conversations
+with users of Apache Airflow, these Events for publishing of Data Assets are very often broadcast
+over a publish and subscribe mechanism. The underlying technology used for this publish and subscribe
+mechanism varies by environment, but Apache Kafka, Amazon SQS, and Google PubSub are commonly used.
+
+
+Expected usage would be something on the lines below::
+
+ trigger = = MsgQueueTrigger(msg_queue="https://sqs.us-east-1.amazonaws.com/722404908466/Test")
+
+ data_asset = Asset("incoming_asset", watchers=[
Review Comment:
Spacing
```suggestion
data_asset = Asset("incoming_asset", watchers=[
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2689024201
> Likely some documentation about base_provider should be added here https://github.com/apache/airflow/blob/f4fd6fd5ae45cd20924149aa0201d2da08a63112/providers/common/messaging/docs/providers.rst
>
> > You need to extend base_provider ble ble ble....
It is already there: https://github.com/apache/airflow/pull/46694/files#diff-f54feaaca8fd8ecfad946ef2cc5b389e082660ba53d305843bab44f5a014d582R36
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2692166051
NICE!
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2682874292
Generally @vincbeck -> look at everything below `[dependency-groups]` in the root `pyproject.toml` - the provider should be added in all those places.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2657003285
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?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vikramkoka (via GitHub)" <gi...@apache.org>.
vikramkoka commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2660125707
> 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.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "shubham22 (via GitHub)" <gi...@apache.org>.
shubham22 commented on code in PR #46694:
URL: https://github.com/apache/airflow/pull/46694#discussion_r1953998635
##########
providers/common/msgq/README.rst:
##########
@@ -0,0 +1,101 @@
+
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ .. NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE OVERWRITTEN!
+
+ .. IF YOU WANT TO MODIFY TEMPLATE FOR THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+ `PROVIDER_README_TEMPLATE.rst.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+
+
+Package ``apache-airflow-providers-common-msgq``
+
+Release: ``0.1.0``
+
+
+
+Provider package
+----------------
+
+This is a provider package for ``common.msgq`` provider. All classes for this provider package
+are in ``airflow.providers.common.msgq`` python package.
+
+This provider package is intended to serve as a common abstraction on top of popular message queue
+providers such as Apache Kafka, Amazon SQS, and Google PubSub.
+
+The expectation is that a common provider abstraction for message queues is especially useful for
+Event Driven Scheduling, which is being introduced as part of Airflow 3.0. Based on conversations
+with users of Apache Airflow, these Events for publishing of Data Assets are very often broadcast
+over a publish and subscribe mechanism. The underlying technology used for this publish and subscribe
+mechanism varies by environment, but Apache Kafka, Amazon SQS, and Google PubSub are commonly used.
+
+
+Expected usage would be something on the lines below::
+
+ trigger = = MsgQueueSensorTrigger(msg_queue="https://sqs.us-east-1.amazonaws.com/722404908466/Test")
Review Comment:
I believe this should be `MessageQueueTrigger(msg_queue...)` as the trigger class name is MessageQueueTrigger.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2683560880
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
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2688997241
I think it would be great to somehow explain that "toctree" better :)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2689164901
And yes - I reverse engineered it having similar issues. Likely it should be done bettter, so we do not have to do it manually.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft Common Message Queue [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #46694:
URL: https://github.com/apache/airflow/pull/46694#issuecomment-2675792198
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 to 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.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org