You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dejii (via GitHub)" <gi...@apache.org> on 2025/08/14 06:40:39 UTC
[PR] AIP-82: implement Google Pub/Sub message queue provider [airflow]
dejii opened a new pull request, #54494:
URL: https://github.com/apache/airflow/pull/54494
This PR implements `PubsubMessageQueueProvider` as part of the AIP-82 https://github.com/apache/airflow/issues/52712 initiative to expand Event-Driven Scheduling, enabling Google Cloud Pub/Sub subscriptions as message queues in Airflow's common messaging framework.
usage:
```python
trigger = MessageQueueTrigger(
queue="projects/my-project-id/subscriptions/my-subcription"
)
asset = Asset(
"pubsub_queue_asset", watchers=[AssetWatcher(name="pubsub_watcher", trigger=trigger)]
)
@dag(
schedule=[asset],
catchup=False,
)
...
```
<!--
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.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
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](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "VladaZakharova (via GitHub)" <gi...@apache.org>.
VladaZakharova commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3205609957
hi there :)
yes, there is this PR which is in draft: https://github.com/apache/airflow/pull/54684
and maybe makes sense to close it to not duplicate...
also I have some questions: are we sure we want to keep this name? it can be confusing for users why we have queue (as a separate service in Google provider), but inside there is still pubsub operators.
And are you planning to add system tests for this? if yes, can you please explain what should be configured in order to run it? in general in Google provider we are trying to keep every test atomic, so all the configuration and connections that should be used inside the system tests should be created and then deleted in the system test
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "dejii (via GitHub)" <gi...@apache.org>.
dejii commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3199983740
> checks failing :(
Fixed. All checks passed now
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3370753806
> @jason810496 Hi there! can we please merge this change for now? From Google side we want to review better this change before releasing it :)
No problem! I just merged 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
Re: [PR] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3374660795
Sorry, so that @dejii might need to raise PR again. 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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "VladaZakharova (via GitHub)" <gi...@apache.org>.
VladaZakharova commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3370765711
@jason810496
I messed up a bit, i am sorry, wanted to say NOT merge it :D
is there a way to roll back it if it is not late?
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "VladaZakharova (via GitHub)" <gi...@apache.org>.
VladaZakharova commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3370778449
yes, please 😿
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3201105203
> Or is it decided ad hoc based on the queue type, as long as we make sure it doesn’t collide with existing ones?
Yes, as long as the `scheme` identifier is different from existing one, the naming should be fine.
> If not, inspired by the redis+pubsub scheme, would it make sense to adopt a consistent pattern like <provider>+<queue>? For this PR, that would mean using google+pubsub as the scheme.
Sure, I named the Redis one as `redis+pubsub` because there are multiple ways to leverage Redis as a message queue (using List, Stream, and the Redis PubSub as in the current Redis provider implementation). Therefore, it's better to maintain flexibility by using `+pubsub` as a suffix in Redis case.
There might also be other services in GCP that provide message queue-like functionality, so naming it `google+pubsub` works well for me.
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "jroachgolf84 (via GitHub)" <gi...@apache.org>.
jroachgolf84 commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3319559997
I also had a draft PR open for this quite a while ago. Please feel free to use this as reference if needed: https://github.com/apache/airflow/pull/54684
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "dejii (via GitHub)" <gi...@apache.org>.
dejii commented on code in PR #54494:
URL: https://github.com/apache/airflow/pull/54494#discussion_r2284154665
##########
providers/google/src/airflow/providers/google/cloud/queues/pubsub.py:
##########
@@ -0,0 +1,73 @@
+# 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
+
+import re
+from typing import TYPE_CHECKING
+
+from airflow.exceptions import AirflowOptionalProviderFeatureException
+from airflow.providers.google.cloud.triggers.pubsub import PubsubPullTrigger
+
+try:
+ from airflow.providers.common.messaging.providers.base_provider import BaseMessageQueueProvider
+except ImportError:
+ raise AirflowOptionalProviderFeatureException(
+ "This feature requires the 'common.messaging' provider to be installed in version >= 1.0.1."
+ )
+
+if TYPE_CHECKING:
+ from airflow.triggers.base import BaseEventTrigger
+
+# [START queue_regexp]
+QUEUE_REGEXP = (
+ r"^projects/(?P<project_id>[^/]+)/subscriptions/"
+ r"(?P<subscription>(?!goog)[A-Za-z][A-Za-z0-9\-_.~+%]{2,254})$"
Review Comment:
It works as long as the subscription name starts with a letter. For example, `abc-%zzz` is valid (I confirmed by creating one). However, if you need to interact directly with the REST APIs, the % must be encoded e.g. `abc-%25zzz`.
For this provider implementation, URL encoding isn’t required since we interact with the Pub/Sub client libraries.
I was able to use:
```python
trigger = MessageQueueTrigger(
queue="projects/my-project/subscriptions/abc-%zzz"
)
```
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #54494:
URL: https://github.com/apache/airflow/pull/54494
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "dejii (via GitHub)" <gi...@apache.org>.
dejii commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3200157646
> Nice! Thanks for the PR!
>
> However, we may need to hold off on merging this until [Refactor Common Queue Interface #54651](https://github.com/apache/airflow/pull/54651) is merged.
>
> As we discuss in [#53556 (comment)](https://github.com/apache/airflow/pull/53556#discussion_r2265242503), there are some drawbacks to using Queue URI, and we should replace Queue URI with just "scheme".
>
> In the meantime, it would be nice to take [000b6c3](https://github.com/apache/airflow/commit/000b6c301eaf874ff239cbf980a9b0e4f13f9b07) as a guide for replacing the "Queue" logic with "Scheme" logic.
>
> Thanks!
Thanks! I just went through the `scheme` logic and it looks solid. One question I had - do we have a standard for generating scheme names? Or is it decided ad hoc based on the queue type, as long as we make sure it doesn’t collide with existing ones?
If not, inspired by the `redis+pubsub` scheme, would it make sense to adopt a consistent pattern like `<provider>+<queue>`? For this PR, that would mean using `google+pubsub` as the scheme.
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "dejii (via GitHub)" <gi...@apache.org>.
dejii commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3205857649
Hi :)
> also I have some questions: are we sure we want to keep this name? it can be confusing for users why we have queue (as a separate service in Google provider), but inside there is still pubsub operators.
This structure comes from the implementation details of AIP-82 - it’s how the `common.messaging` provider discovers the available queues.
Reference:
https://github.com/apache/airflow/blob/b340e8c101c1ed12fb097f9aa5b41bb9c2599275/airflow-core/src/airflow/providers_manager.py#L1097-L1103
https://github.com/apache/airflow/blob/b340e8c101c1ed12fb097f9aa5b41bb9c2599275/providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py#L38
> And are you planning to add system tests for this? if yes, can you please explain what should be configured in order to run it? in general in Google provider we are trying to keep every test atomic, so all the configuration and connections that should be used inside the system tests should be created and then deleted in the system test
This can be looked into. A Pub/Sub subscription (and implicitly a topic) would be required, but given the way event-driven scheduling works, I’m not sure it can be created and torn down entirely within the scope of a system test as the subscription is what triggers the DAG.
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3192514581
checks failing :(
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3370775507
> @jason810496 I messed up a bit, i am sorry, wanted to say NOT merge it :D is there a way to roll back it if it is not late?
I can revert it right now on GitHub. WDYT?
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "abhiips07 (via GitHub)" <gi...@apache.org>.
abhiips07 commented on code in PR #54494:
URL: https://github.com/apache/airflow/pull/54494#discussion_r2282026191
##########
providers/google/src/airflow/providers/google/cloud/queues/pubsub.py:
##########
@@ -0,0 +1,73 @@
+# 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
+
+import re
+from typing import TYPE_CHECKING
+
+from airflow.exceptions import AirflowOptionalProviderFeatureException
+from airflow.providers.google.cloud.triggers.pubsub import PubsubPullTrigger
+
+try:
+ from airflow.providers.common.messaging.providers.base_provider import BaseMessageQueueProvider
+except ImportError:
+ raise AirflowOptionalProviderFeatureException(
+ "This feature requires the 'common.messaging' provider to be installed in version >= 1.0.1."
+ )
+
+if TYPE_CHECKING:
+ from airflow.triggers.base import BaseEventTrigger
+
+# [START queue_regexp]
+QUEUE_REGEXP = (
+ r"^projects/(?P<project_id>[^/]+)/subscriptions/"
+ r"(?P<subscription>(?!goog)[A-Za-z][A-Za-z0-9\-_.~+%]{2,254})$"
Review Comment:
I was reading the GCP docs about subscription naming. Suppose I put a % in the name, but it’s not a valid UTF-8 encoding (like %zzz). What happens in that case?
> You can use the special characters in the preceding list in resource names without URL-encoding. However, you must ensure that any other special characters are properly encoded or decoded when you use them in URLs. For example, mi-tópico is an invalid ID. However, mi-t%C3%B3pico is valid. This format is important when you make REST calls.
https://cloud.google.com/pubsub/docs/pubsub-basics#resource_names
--
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] AIP-82: implement Google Pub/Sub message queue provider [airflow]
Posted by "VladaZakharova (via GitHub)" <gi...@apache.org>.
VladaZakharova commented on PR #54494:
URL: https://github.com/apache/airflow/pull/54494#issuecomment-3370717117
@jason810496
Hi there! can we please merge this change for now? From Google side we want to review better this change before releasing 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