Refactor Common Queue Interface

[jason810496]

related: #52712
related discussion: #53556 (comment)

Why

Here are some advantages of using "scheme+passing keyword parameters directly from MessageQueueTrigger" instead of the "Queue URI" interface:

1. It is more secure than explicitly including the host, port, and URI at the Dag code level.
2. It is less confusing for users, as specifying the "host" and "port" in the URI could conflict with the current Airflow Connection feature and create ambiguity.

For more details, please refer to the discussion at: #53556 (comment)

What

• The queue URI parameter should be replaced by the scheme parameter for matching the provider's queue.
  • The remaining parameters required by the provider's trigger (e.g., AwaitMessageTrigger, SqsSensorTrigger) should be passed directly from MessageQueueTrigger.
• Provide a backward-compatible interface for using the queue URI.
  • This is mainly for AWS SQS and Apache Kafka.
  • Since Redis Queue has not been released yet, we can simply replace the "queue" logic with the "scheme" logic.
• Add a deprecation warning for the queue parameter in both logging and documentation.

[jason810496]

Hi @vincbeck, @potiuk, could you please take a look when you have a moment? Thank you!
The Google Pub/Sub and Azure Message Bus implementations are still depends on this one.

[potiuk]

I will take a look in a day or two. Some priority things are creeping in.

[jason810496]

I think we need to discuss and approach a deprecation mechanism for it. I would be for a rather "clean cut" with this one than a long deprecation. I see no particular reason to stick to the old mechanism if we convert all our message queue providers to the new versions.

Agree. I will vote for "Sudden death" approach in this case and have same though as this Those are still very early days for message queue, and i bet not many people have already built their own queues, and even if they did, switching to the new API is very straightforward. part. Additionally, having "Sudden death" is more clean because replacing Queue URI with Scheme is more like a fix but also a breaking change IMO.

If we decide to go with "Sudden death", does it mean we need to remove the compatible interface for common queue?
Like the following part can be directly removed? ( Since there will be no more queue parameter in 2.0.0 )

https://github.com/apache/airflow/pull/54651/files#diff-9eba0f966907dc13a12fae086fff750db6a51fffc60275fdddbea41227fb6dd8R68-R80

you will not be able to downgrad amazon, kafka, redis selectively

By the way, only SQS and Kafka will be impacted, since Queue Provider of Redis haven't been released before. It was excluded in last provider release.

[vincbeck]

Nice!

[vincbeck]

To avoid having new queue provider using the deprecated mechanism, I vote for the "sudden death" solution too

[Jasperora]

Thanks @jason810496 !

[potiuk]

Yeah. I propose sudden death too :)

[potiuk]

cc: @eladkal . This is yet another example where we have to (when you prepare providers) do the following:

a) update common.messaging to 2.0.0
b) update all providers that use common.messaging to be >= 2.0.0

There are two options for it:

1. This could be done now -> but this will cause the problematic changelog behaviours you complained about - i.e. the changelog type selection will not be taken into account.

2. You need to remember to do it when you prepare release documentation. Currently we have no information stored anywhere except this discussion here what to do with those providers. Seems that current consensus is that we should do "sudden death" - i.e what I described above (2.0.0 in common.messaging and >=2.0.0 in dependencies of providers that depend on common.messaging).

I am not sure if there is another way of storing it - or just remembering that you have to do it is enough. Or maybe we can figure out a way how to replace "remembering" with some information stored in the repo. Currently we have no such information stored, what we should do is literally stored in this discussion.

Maybe we might come up with some good ideas how to handle those cases better @gopidesupavan @amoghrajesh as well?

[amoghrajesh]

@potiuk @eladkal I might have a simple proposal here without too much complexity. What if we add a simple coordination block at the top of provider changelogs to store actionable information that release tooling can process during the release?

In short this is what I propose, we define a block at the top of a provider's changelog (since this will be mostly be per provider granularity)

Format

# providers/common/messaging/docs/changelog.rst

.. COORDINATION-PENDING::
   
   **Breaking Version:** 2.0.0
   
   **Update Required:**
   
   * apache-airflow-providers-amazon: ">=2.0.0"
   * apache-airflow-providers-google: ">=2.0.0" 
   * apache-airflow-providers-azure: ">=2.0.0"

``apache-airflow-providers-common-messaging``

Changelog
---------

2.0.0
.....

Workflow

1. Developer adds coordination block when introducing breaking changes
2. Tooling scans for COORDINATION-PENDING blocks during preparation
3. Tooling automatically updates affected provider dependencies
4. Block is removed/marked complete when coordination is done

Potential Implementation

We will have to define some sort of parsing block to automatically updating provider dependencies during release preparation.

Like define a python class like this --

Block:

class CoordinationRequirement:
    breaking_version: str
    affected_providers: Dict[str, str]  # provider_name -> version_constraint

Parsing:

def parse_coordination_block(content: str) -> CoordinationRequirement:
    # Find the coordination block
    block_match = re.search(r'\.\.\ COORDINATION-PENDING::(.*?)(?=^\S|\Z)', content, re.MULTILINE | re.DOTALL)
    if not block_match:
        return None
    
    block_content = block_match.group(1)
    
    version_match = re.search(r'\*\*Breaking Version:\*\*\s+(\S+)', block_content)
    if not version_match:
        return None
    breaking_version = version_match.group(1)
    
    affected_providers = {}
    provider_lines = re.findall(r'\*\s+([^:]+):\s+"([^"]+)"', block_content)
    
    for provider, version_constraint in provider_lines:
        clean_provider = provider.replace('apache-airflow-providers-', '').strip()
        affected_providers[clean_provider] = version_constraint
    
    return CoordinationRequirement(
        breaking_version=breaking_version,
        affected_providers=affected_providers
    )

Example:

content = """
.. COORDINATION-PENDING::
   **Breaking Version:** 2.0.0
   **Update Required:**
   * apache-airflow-providers-amazon: ">=2.0.0"
   * apache-airflow-providers-google: ">=2.0.0" 
   * apache-airflow-providers-azure: ">=2.0.0"
"""
req = parse_coordination_block(content)
# req.breaking_version = "2.0.0"
# req.affected_providers = {
#     "amazon": ">=2.0.0",
#     "google": ">=2.0.0", 
#     "azure": ">=2.0.0"
# }
req
Out[6]: CoordinationRequirement(breaking_version='2.0.0', affected_providers={'* amazon': '>=2.0.0', 'google': '>=2.0.0', 'azure': '>=2.0.0'})

[potiuk]

@potiuk @eladkal I might have a simple proposal here without too much complexity. What if we add a simple coordination block at the top of provider changelogs to store actionable information that release tooling can process during the release?

Yep. I thought about something similar. I think it is important to have it as simple as possible and natural for the developer to do.

I think maybe a better idea though could be comments in pyproject.toml of the affected providers. That's a bit more localized to the places that should be updated, and we will not have to remove much more than a comment:

Amazon pyproject.toml:

[project.optional-dependencies]
"common.messaging" = [
    "apache-airflow-providers-common-messaging>=1.0.3"  # BUMP

Redis pyproject.toml:

[project.optional-dependencies]
"common.messaging" = [
    "apache-airflow-providers-common-messaging>=1.0.3"  # BUMP
]

Maybe not exactly this comment, but smth like that should be more than enough to actually do the bump. It would be "post-processing" after prepare providers documentation is done.

[potiuk]

Also - I think this is a bit more "accurate" - becasue we do not know exactly what will be the bump. For example if we decided that common.messaging is not going to have a major release (i.e. non-sudden-death scenario) the # BUMP would mean >=1.1.0 not >=2.0.0 - but we only know it at the moment release manager reviews common.messaging and decides whether we have "feature" or "breaking" change.