Skip to content

AIP-82: Add KinesisMessageQueueProvider - #73509

Open
aaron-y-chen wants to merge 3 commits into
apache:mainfrom
aaron-y-chen:aip-82-kinesis-common-message-queue
Open

aaron-y-chen wants to merge 3 commits into
apache:mainfrom
aaron-y-chen:aip-82-kinesis-common-message-queue

Conversation

@aaron-y-chen

Copy link
Copy Markdown
Contributor

related: #52712

Summary

Add Amazon Kinesis Data Streams integration to the common messaging interface (AIP-82), connecting the existing KinesisTrigger to MessageQueueTrigger using scheme="kinesis".

  • Implements KinesisMessageQueueProvider with scheme-based dispatch (scheme="kinesis").
  • Registers the queue provider in provider.yaml and updates provider metadata.
  • Adds user documentation in message-queues/index.rst and an illustrative example Dag (example_kinesis_message_queue.py).

Testing

  • Live AWS E2E Verification:
    Verified against a live Amazon Kinesis Data Stream (airflow-kinesis-cmq-e2e in us-west-2) with a running Airflow Triggerer:
    1. A test record (kinesis-e2e-payload-001) was sent to Kinesis via AWS CLI put-record.
    2. MessageQueueTrigger(scheme="kinesis") detected the record and emitted an AssetEvent (kinesis://airflow-kinesis-cmq-e2e).
    3. The Scheduler automatically triggered an asset_triggered Dag run (asset_triggered__2026-09-21T21:46:23.928742+00:00_I3mBFdjv), and the task successfully decoded and verified the record data.

E2E Screenshots

  1. AWS Kinesis Console (Data viewer showing the put record):
    kinesis_1
  2. Airflow Web UI (asset_triggered Dag run and decoded log):
    kinesis_2

Was generative AI tooling used to co-author this PR?
  • Yes (please specify the tool below)

Generated-by: [Codex-6] following the guidelines


  • 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.

@aaron-y-chen
aaron-y-chen marked this pull request as ready for review September 21, 2026 23:35
Comment thread providers/amazon/src/airflow/providers/amazon/aws/queues/kinesis.py Outdated
@aaron-y-chen

Copy link
Copy Markdown
Contributor Author

Thanks for the reviews, they're really helpful 😄

@vincbeck vincbeck left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM :)

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants