You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jason810496 (via GitHub)" <gi...@apache.org> on 2025/04/29 08:57:30 UTC
[PR] [WIP] AIP-82: Add KafkaMessageQueueProvider [airflow]
jason810496 opened a new pull request, #49938:
URL: https://github.com/apache/airflow/pull/49938
## Why
Several users have pointed out that the Event-Driven Scheduling feature introduced in Airflow 3 currently supports only Amazon SQS.
For example, in [Airflow 3.0: The Good, The DAG, and the Ugly (and the Future!)](https://codyrich.medium.com/airflow-3-0-the-good-the-dag-and-the-ugly-and-the-future-a0c60b5dafce), the author notes:
> “AssetWatchers, in my opinion, is where TRUE data-aware scheduling comes into play. However, Amazon SQS is the only message queue currently supported. While a great start, I’m excited to see how Airflow’s massive community builds on this, supporting various systems like Snowflake staging tables.”
## What
This PR introduces support for **Apache Kafka** as a `MessageQueueProvider`, expanding the capabilities of Event-Driven Scheduling.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2872344798
Very 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] AIP-82: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2866416255
Hi @vincbeck do you mind take a look at PR after moving to Kafka-specific directory when you are free? Thanks.
If there isn't other concern, I will keep working on other common Message Queue providers to enrich Event-Driven Scheduling feature.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844958304
> As for the JSON format—perhaps we could have a `queue_type` for matching and an `extra` field for provider-specific kwargs?
> Just a rough idea, but JSON format also works for me.
Yeah. We can iterate on that in the future versions - I think :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] AIP-82: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069050039
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/kafka.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
My proposal would be to implement similar feature as we have in DBAPI Hook where we have provider's manager and exposing "Hooks" available in provider - just expose "MessageQueue" implementations - that would be very similar code and it would work in a similar way => we have all the discovery code already in place. The added benefit is that we could also automatically show it as output of `airflow providers` CLI, and also our documentation here https://airflow.apache.org/docs/apache-airflow-providers/core-extensions/index.html
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2847628574
> OK. Everything for common.messaging is merged and even backported for Airflow 3.0.1 @jason810496 -> you can follow the `sqs` implementation and do similar for Kafka. I think for Kafka you should even be able to have an integration test implemented.
Thanks again! Sure, I will follow #50057 for refactoring Kafka one. I see, yes, there should be more appropriate to have an integration test for kafkaMessageQueueProvider.
Make the PR as draft for refactoring.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069017281
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/__init__.py:
##########
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+from airflow.providers.common.messaging.providers.kafka import KafkaMessageQueueProvider
from airflow.providers.common.messaging.providers.sqs import SqsMessageQueueProvider
-MESSAGE_QUEUE_PROVIDERS = [SqsMessageQueueProvider()]
+MESSAGE_QUEUE_PROVIDERS = [SqsMessageQueueProvider(), KafkaMessageQueueProvider()]
Review Comment:
This is what we would have to do to make it happens - we would have to discover the available providers from the packages installed.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844716502
This is how I envision it (updated my https://github.com/apache/airflow/pull/50057 with SQS to show how it should look like IMHO (and how we should communicate to avoid confusion):
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2842850128
> Thank you for pointing that out @potiuk
>
> I intended for it to be decoupled with the original messaging abstraction. I didn't notice the "coupling issue" earlier. That was definitely unintentional.
BTW. We can even make it backwards compatible when we switch to the "in-provider" abstraction
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2843406760
There you go - https://github.com/apache/airflow/pull/50057/ I think it's already pretty complete change.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069050039
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/kafka.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
My proposal would be to implement similar feature as we have in DBAPI Hook where we have provider's manager and exposing "Hooks" available in provider - just expose "MessageQueue" implementations - that would be very similar code and it would work in a similar way => we have all the discovery code already in place. The added benefit is that we could also automatically show it as output of `airflow providers` CLI, and also our documentation here would show the list of implemented "MessageQueues" https://airflow.apache.org/docs/apache-airflow-providers/core-extensions/index.html
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069452569
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/kafka.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Will do!
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2850839265
Just moved the Kafka Queue Provider to the Apache Kafka-specific directory and it's ready for review.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "vikramkoka (via GitHub)" <gi...@apache.org>.
vikramkoka commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2842805587
Thank you for pointing that out @potiuk
I intended for it to be decoupled with the original messaging abstraction.
I didn't notice the "coupling issue" earlier. That was definitely unintentional.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069053249
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/kafka.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Happy to help with the implementation if we agree this is the right way. Should be rather straightforward
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "cmarteepants (via GitHub)" <gi...@apache.org>.
cmarteepants commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2898614119
This is awesome! Which version are we targeting for this?
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069036049
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/kafka.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Just to explain what's currently problematic here.
If you look at `pyproject.toml` - > both `kafka` and `sqs` are optional dependencies. But in fact - they are not.
Currently in `__init__.py` we import **BOTH** Kafka and SQS Queue Provider - and transitively each of the Queue Provider implementions imports their respective provider.
Which means that if someone wants to use just Kafka provider but not SQS provider, they will have to install amazon provider.
That is very inconvenient for a generic `common.messaging` to have to install all providers that have "potential" implementations.
Surely we can do some try/catches to catch import errors an do turn it into really optional provider features, but it seems brittle and introduces unnecessary "other providers" code in common.messaging that should be very generic.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069096040
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/kafka.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
I agree, that's something I missed when I implemented it. We should definitely change that. I like your proposal Jarek, we should probably do that. If you're up to, yes please create a PR :) Happy to review!
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844589605
> I'd also like to open a discussion about the end-user interface for `MessageQueueTrigger`. While working on the `KafkaMessageQueueProvider`, I noticed that the constructor arguments for different provider-specific trigger classes vary significantly.
>
> e.g. Kafka: `AwaitMessageTrigger`, Google Pub/Sub: `PubsubPullTrigger`
>
> https://github.com/apache/airflow/blob/c37e6eb482548d913d71c3ac023b7d74c05d5b5e/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py#L59-L68
>
> https://github.com/apache/airflow/blob/c37e6eb482548d913d71c3ac023b7d74c05d5b5e/providers/google/src/airflow/providers/google/cloud/triggers/pubsub.py#L54-L63
>
> Additionally, the `queue` URI argument in the current interface is used solely to match the appropriate provider, and not necessarily tied to an Airflow Connection. This might be confusing to users.
>
> Should we consider replacing the `queue` argument in the `MessageQueueTrigger` constructor with something more explicit, like `type` or `provider`?
>
> For example:
>
> ```python
> AVAILABLE_MESSAGE_QUEUE_PROVIDERS = Literal[
> "sqs",
> "kafka",
> ]
>
> class MessageQueueTrigger(BaseEventTrigger):
>
> def __init__(self, *, provider: AVAILABLE_MESSAGE_QUEUE_PROVIDERS, **kwargs: Any) -> None:
> self.provider = provider
> self.kwargs = kwargs
>
> @cached_property
> def trigger(self) -> BaseEventTrigger:
> # get trigger_class based on provider attribute
> ```
>
> cc @potiuk, @vincbeck, @vikramkoka
Good question. I really wanted to avoid that and totally abstract the notion of provider to the user, that's why I came up with the parameter `queue`. But I dont think it is possible to know from a Kafka topic name or a Google pubsub project_id the identity of the provider. One solution would be to "create" a format/protocol where we could identity the provider and extract the queue identifier:
- Kafka: `kafka://{topic_name}`
- Google pubsub: `google://{project_id}`
But this would create a new complexity specific to Airflow so I dont think this is a good idea. To be honest I am not a big fan of introducing this new parameter `provider` but I do not see other solutions. 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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844687861
Wrong button pressed :) . sorry for closing
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844727402
And even better where we include the regexp directly from sources to the documentation:
--
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] [WIP] AIP-82: Add KafkaMessageQueueProvider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2838017702
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] AIP-82: Add KafkaMessageQueueProvider [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2073365480
##########
providers/common/messaging/tests/system/common/messaging/kafka_message_queue_trigger.py:
##########
Review Comment:
Should this be in common.messaging or in kafka provider?
I think we may want to have index in common.messaging that leads to examples in the various providers that holds the logic (amazon sqs, kafka)
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #49938:
URL: https://github.com/apache/airflow/pull/49938
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2069008941
##########
providers/common/messaging/src/airflow/providers/common/messaging/providers/kafka.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Hmm. I have some doubts.. I already had it already but this one actually makes it more visible.
Why are we adding the `common.messaging` implementation to common.messaging?
They shoud be in "Amazon" (sqs) and `Kafka` provider (kafka`) - NOT in common.messaging I think.
Similarly to common.sql - where implementation of teh DBAPIHook is in the respective provider.
OR am I wrong ?
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844668124
Why don't we add additonal `*trigger_kwargs` that we could pass to the message queue and use as constructor args ?
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844668797
We have very similar approach for connections already where we pass hook kwargs
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844782435
> What would happen if we want to add a new provider to support common.messaging? Should we release a new common.messaging provider version? That makes no sense and is completely against the "genericness" of it - also it prevents others from developing their `common.messaging` capable provider independently.
I see—I overlooked the general-purpose of the common provider. In that case, having a `type` or `provider` argument in the constructor might not be a good idea.
> We just need to be very clear - for each implementation - what is the format of the `queue` URI. I don't think this will be universally standard - because basically there is no universal standard for globally recognized URI format for message queues. Or maybe there are 5 competing standards or smth.. I think - similarly as in case of connections (where we have AIRFLOW connection URI form - for good or better) this should be done this way.
> And even better where we include the regexp directly from sources to the documentation:
Looks great!
The detailed documentation really helps clarify the ambiguity around the `queue` URI, especially with the inclusion of regex examples and how the parsed values are passed to the trigger constructor.
> we might consider supporting "json" form of the queue on top of the URI. URI has certain limitations (especially around encoding) and eventually we settled on both URI and JSON formats supported for connection for that reason.
As for the JSON format—perhaps we could have a `queue_type` for matching and an `extra` field for provider-specific kwargs?
Just a rough idea, but JSON format also works for me.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #49938: AIP-82: Add KafkaMessageQueueProvider
URL: https://github.com/apache/airflow/pull/49938
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2899960121
> This is awesome! Which version are we targeting for this? 3.0.x or 3.1?
This targets 3.0.x and will be included in the next provider release.
However, I haven't tested the RC version yet 😅
https://github.com/apache/airflow/issues/50599
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844210132
I'd also like to open a discussion about the end-user interface for `MessageQueueTrigger`. While working on the `KafkaMessageQueueProvider`, I noticed that the constructor arguments for different provider-specific trigger classes vary significantly.
e.g. Kafka: `AwaitMessageTrigger`, Google Pub/Sub: `PubsubPullTrigger`
https://github.com/apache/airflow/blob/c37e6eb482548d913d71c3ac023b7d74c05d5b5e/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py#L59-L68
https://github.com/apache/airflow/blob/c37e6eb482548d913d71c3ac023b7d74c05d5b5e/providers/google/src/airflow/providers/google/cloud/triggers/pubsub.py#L54-L63
Additionally, the `queue` URI argument in the current interface is used solely to match the appropriate provider, and not necessarily tied to an Airflow Connection. This might be confusing to users.
Should we consider replacing the `queue` argument in the `MessageQueueTrigger` constructor with something more explicit, like `type` or `provider`?
For example:
```python
AVAILABLE_MESSAGE_QUEUE_PROVIDERS = Literal[
"sqs",
"kafka",
]
class MessageQueueTrigger(BaseEventTrigger):
def __init__(self, *, provider: AVAILABLE_MESSAGE_QUEUE_PROVIDERS, **kwargs: Any) -> None:
self.provider = provider
self.kwargs = kwargs
@cached_property
def trigger(self) -> BaseEventTrigger:
# get trigger_class based on provider attribute
```
cc @potiuk, @vincbeck, @vikramkoka
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844207010
Thanks for pointing that out @potiuk, and for the PR to address it.
Moving the provider-specific message queue implementations to the `providers` directory definitely makes more sense!
I'll rebase and relocate my changes to the appropriate provider path once your PR is merged.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844687589
Ah we already have the kwargs..... ... bad me .. I see the point now.
First of all: - no, absolutely we should not have list of avaliable providers in `common.messaging` . That completely defeats the purpose of common.messaging being a generic interface if we want to make list of providers in the common.messaging.
What would happen if we want to add a new provider to support common.messaging? Should we release a new common.messaging provider version? That makes no sense and is completely against the "genericness" of it - also it prevents others from developing their `common.messaging` capable provider independently.
Other than optional extras that indicate the list of providers that we have in community that support common.messaging - there should be ABSOLUTELY NO DIFFERENCE between our providers, and 3rd-party providers implementing common.messaging. So list of "possible" providers in the code of common.messaging is generally out-of-the-question.
I think the way it works now is not really too confusing and pretty good
The `queue` should be enough and should be interpreted by the particular MessageQueue implementation - i.e. if the queue matches SQS, it's up to SQSMessageQueue to interpret the queue + kwargs. Ideally it will derive some basic parameters that can be passed to the Trigger without passing extra args. But yes - there might be other MessageQueue implementations that expect more. And it should be up to the implemenation how to interpret - both queue and kwargs.
For example I can imagine a PubSubMessageQueue implementation that takes such a queue (not sure what is the right form - it's more of a logical representation
```
queue="pubsub://project:subscription", kwargs = {"gcp_conn_id" = "connection" }
```
In this case the queue could be parsed by the PubSubMessageQueue and converted into appropriate trigger parameters, while allowing additional trigger parameters to be passed by kwargs if they cannot be easily translated into `queue` URI or where we would like to use `some` standardized form.
We just need to be very clear - for each implementation - what is the format of the `queue` URI. I don't think this will be universally standard - because basically there is no universal standard for globally recognized URI format for message queues. Or maybe there are 5 competing standards or smth.. I think - similarly as in case of connections (where we have AIRFLOW connection URI form - for good or better) this should be done this way.
Two learning from connections though:
* we have to be VERY clear that queue is not universally standard and each message queue should precisely describe what kind of URI and kwargs it expects
* we might consider supporting "json" form of the `queue` on top of the URI. URI has certain limitations (especially around encoding) and eventually we settled on both URI and JSON formats supported for connection for that reason.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2844740308
Indeed, the documentation helps. I am okay with that
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #49938:
URL: https://github.com/apache/airflow/pull/49938#discussion_r2073464442
##########
providers/common/messaging/tests/system/common/messaging/kafka_message_queue_trigger.py:
##########
Review Comment:
Bass on discussion above, we should place in Kafka provider.
I overlooked the system test, thanks for the catch.
--
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: Add KafkaMessageQueueProvider [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49938:
URL: https://github.com/apache/airflow/pull/49938#issuecomment-2847379176
OK. Everything for common.messaging is merged and even backported for Airflow 3.0.1 @jason810496 -> you can follow the `sqs` implementation and do similar for Kafka. I think for Kafka you should even be able to have an integration test implemented.
--
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