Add MskHook

[arnoldmr01]

What is this PR for?

Integrate Amazon MSK for user to have variety way to implement Event-Driven Scheduling feature which has been introduced in Airflow 3.0
related: #52712

What is changed in this PR?

This PR add a hook for Amazon MSK. We only make this hook as an interface for creating an authenticated MSK boto3 client so far.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
• Codex

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[o-nikolas]

Most of the methods look like single line wrappers around the boto API. It seems like we don't have much of a usecase for this? Curious to hear what @vincbeck and @ferruzzi have to say about this one.

[o-nikolas]

The usage of this is optional in the code as you've written it, but it's included as a mandatory dependency here. Let's move this below to be an optional dependency

[arnoldmr01]

Ok, I will modify it later

[o-nikolas]

We'd probably have to bump the botocore minimum since these are new APIs?

[github-actions]

uv.lock on main just moved via #68933 ("[main] Upgrade important CI environment"), commit e0a1726 and this PR currently conflicts.

Quickest fix:

git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease

Automated nudge — ignore if you're not ready to rebase. This comment is updated in place on future uv.lock bumps.

[ferruzzi]

Agreed with Niko. Looks like the hooks are generally replacing one-liners with functions, which we generally avoid.

[jason810496]

Thanks, LGTM overall.

Most of the methods look like single line wrappers around the boto API. It seems like we don't have much of a usecase for this?

I agreed with this point as well. We could make the MSK Hook itself just the interface to retrieve the authenticated MSK client with given conn_id.

Same as the AwaitMessageTrigger (the Kafka concusmer in Airlfow trigger) implementation, we will leverage the SDK client that retrieved via connection directly without additional wrapper.

airflow/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py

Lines 111 to 118 in 81dbef9

	asyncdefrun(self):
	consumer_hook=KafkaConsumerHook(topics=self.topics, kafka_config_id=self.kafka_config_id)
	async_get_consumer=sync_to_async(consumer_hook.get_consumer)
	self._consumer=awaitasync_get_consumer()
	async_poll=sync_to_async(self._consumer.poll)
	async_commit=sync_to_async(self._consumer.commit)

[vincbeck]

I agree, I dont think these methods are needed. They do not add much value, in the task/operator you can directly invoke boto3 the same way you do it in the hook

[arnoldmr01]

Thanks for reviewing. I agree that these methods are not necessary. I will follow @jason810496 comment and modify the PR and make it focus on retrieving authenticated MSK client

[jason810496]

LGTM overall, thanks.

[arnoldmr01]

Thanks for your review

[arnoldmr01]

Add confluent_token for confluent_kafka to connect with Amazon MSK

[eladkal]

I thought the task is to also add a trigger. Did I miss something?

[arnoldmr01]

No, you're right. I will add the trigger in the follow-up PR