Add KafkaMessageQueueTrigger as a Convenience Wrapper

[dingo4dev]

Description

Add a KafkaMessageQueueTrigger class inherit from MessageQueueTrigger that provides a more intuitive API for Kafka-specific use cases while leveraging the existing architecture.

Use case/motivation

Currently need to dig deep into the provider function and found out what kwargs can pass to the Trigger.

It will be more intuitive if provides:

• Clear parameter names for Kafka-specific configurations
• Better type hints and validation

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[jroachgolf84]

@dingo4dev, thanks for opening up this issue. Take a peek at this PR: #49938. Does this provide the functionality that you're looking for?

[dingo4dev]

Thank you for your response. I would like to configure Kafka in the context of message queue trigger. However, it seems that there is no explicit type hint available for this unless we inspect the code for both the Kafka provider and the message queue trigger.

https://github.com/jason810496/airflow/blob/9da23244ba0b3ce110033b804928e98304d505ae/providers/apache/kafka/src/airflow/providers/apache/kafka/queues/kafka.py#L79-L100

https://github.com/jason810496/airflow/blob/fa3def0010c7e0589d6d7a8391a33c49b52bc55a/providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py#L74-L76

I think it is better to provide a class for Kafka provider for convenience:

class KafkaMessageQueueTrigger(MessageQueueTrigger):
    def __init__(
        self,
        topics: Sequence[str],
        apply_function: str,
        kafka_config_id: str = "kafka_default",
        apply_function_args: Sequence[Any] | None = None,
        apply_function_kwargs: dict[Any, Any] | None = None,
        poll_timeout: float = 1,
        poll_interval: float = 5,
    ) -> None:        
        # Any validation
        queue_topics= ...
        kafka_boostrap= ...
        super().__init__(
            queue=f"kafka://{kafka_bootstrap}/{topics}",
            apply_function=apply_function,
            apply_function_args=apply_function_args,
            poll_timeout=poll_timeout,
            poll_interval: float = 5
        )

trigger = KafkaMessageQueueTrigger(
    topics=["test"],
    kafka_config_id="kafka_airflow",
    apply_function="kafka_message_queue_trigger.apply_function",
)
asset = Asset("kafka_queue_asset", watchers=[AssetWatcher(name="kafka_watcher", trigger=trigger)])

[jroachgolf84]

Happy to assign this issue to you if you'd like to contribute it!

[dingo4dev]

Thanks, I'd love to try it!