kaxil commented on code in PR #62790:
URL: https://github.com/apache/airflow/pull/62790#discussion_r3171505679
##########
airflow-core/docs/extra-packages-ref.rst:
##########
@@ -343,6 +343,8 @@ Some of those enable Airflow to use executors to run tasks
with them - other tha
+---------------------+-----------------------------------------------------+-----------------------------------------------------------------+----------------------------------------------+
| influxdb | ``pip install 'apache-airflow[influxdb]'`` |
Influxdb operators and hook |
|
+---------------------+-----------------------------------------------------+-----------------------------------------------------------------+----------------------------------------------+
+| ibm-mq | ``pip install 'apache-airflow[ibm-mq]'`` |
IBM MQ hooks and operators |
|
Review Comment:
**"hooks and operators" overstates what the provider ships.**
This row says "IBM MQ hooks and operators" but the provider currently
exposes hooks, triggers, and a queue provider -- there are no operators. Users
reading this table will install the package expecting `IBMMQOperator` /
`IBMMQSensor` and find neither.
Suggest: "IBM MQ hook and trigger" (matches what's actually in
`provider.yaml`).
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/queues/mq.py:
##########
@@ -0,0 +1,87 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+from airflow.providers.common.messaging.providers.base_provider import
BaseMessageQueueProvider
+from airflow.providers.ibm.mq.triggers.mq import AwaitMessageTrigger
+
+if TYPE_CHECKING:
+ from airflow.triggers.base import BaseEventTrigger
+
+# [START queue_regexp]
+# Matches ibmmq://<conn_id>/<queue_name> (conn_id required, queue_name
required)
+QUEUE_REGEXP = r"^ibmmq://[^/]+/.+"
+# [END queue_regexp]
+
+
+class IBMMQMessageQueueProvider(BaseMessageQueueProvider):
+ """
+ Configuration for IBM MQ integration with common-messaging.
+
+ [START ibmmq_message_queue_provider_description]
+
+ * It uses ``ibmmq`` as scheme for identifying IBM MQ queues.
+ * For parameter definitions take a look at
+ :class:`~airflow.providers.ibm.mq.triggers.mq.AwaitMessageTrigger`.
+
+ .. code-block:: python
+
+ from airflow.providers.common.messaging.triggers.msg_queue import
MessageQueueTrigger
+ from airflow.sdk import Asset, AssetWatcher
+
+ trigger = MessageQueueTrigger(
+ queue="ibmmq://mq_default/MY.QUEUE.NAME",
+ )
+
+ asset = Asset("mq_topic_asset",
watchers=[AssetWatcher(name="mq_watcher", trigger=trigger)])
+
+ [END ibmmq_message_queue_provider_description]
+ """
+
+ scheme = "ibmmq"
+
+ def queue_matches(self, queue: str) -> bool:
+ return bool(re.match(QUEUE_REGEXP, queue))
+
+ def trigger_class(self) -> type[BaseEventTrigger]:
+ return AwaitMessageTrigger # type: ignore[return-value]
+
+ def trigger_kwargs(self, queue: str, **kwargs) -> dict[str, Any]:
+ # Parse the queue URI
+ parsed = urlparse(queue)
+
+ if not parsed.netloc:
+ raise ValueError(
+ "MQ URI must contain connection id. Expected format:
mq://<conn_id>/<queue_name>"
Review Comment:
**Scheme-rename leftover.** This error message still tells users the format
is `mq://<conn_id>/<queue_name>`, but the provider scheme was renamed to
`ibmmq` (see `scheme = "ibmmq"` on line 60 and `QUEUE_REGEXP =
r"^ibmmq://[^/]+/.+"` on line 32). A user who pastes `mq://...` to follow this
error will fail `queue_matches()` and never reach this code path. Update to
`ibmmq://<conn_id>/<queue_name>`.
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/queues/mq.py:
##########
@@ -0,0 +1,87 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+from airflow.providers.common.messaging.providers.base_provider import
BaseMessageQueueProvider
+from airflow.providers.ibm.mq.triggers.mq import AwaitMessageTrigger
+
+if TYPE_CHECKING:
+ from airflow.triggers.base import BaseEventTrigger
+
+# [START queue_regexp]
+# Matches ibmmq://<conn_id>/<queue_name> (conn_id required, queue_name
required)
+QUEUE_REGEXP = r"^ibmmq://[^/]+/.+"
+# [END queue_regexp]
+
+
+class IBMMQMessageQueueProvider(BaseMessageQueueProvider):
+ """
+ Configuration for IBM MQ integration with common-messaging.
+
+ [START ibmmq_message_queue_provider_description]
+
+ * It uses ``ibmmq`` as scheme for identifying IBM MQ queues.
+ * For parameter definitions take a look at
+ :class:`~airflow.providers.ibm.mq.triggers.mq.AwaitMessageTrigger`.
+
+ .. code-block:: python
+
+ from airflow.providers.common.messaging.triggers.msg_queue import
MessageQueueTrigger
+ from airflow.sdk import Asset, AssetWatcher
+
+ trigger = MessageQueueTrigger(
+ queue="ibmmq://mq_default/MY.QUEUE.NAME",
+ )
+
+ asset = Asset("mq_topic_asset",
watchers=[AssetWatcher(name="mq_watcher", trigger=trigger)])
+
+ [END ibmmq_message_queue_provider_description]
+ """
+
+ scheme = "ibmmq"
+
+ def queue_matches(self, queue: str) -> bool:
+ return bool(re.match(QUEUE_REGEXP, queue))
+
+ def trigger_class(self) -> type[BaseEventTrigger]:
+ return AwaitMessageTrigger # type: ignore[return-value]
+
+ def trigger_kwargs(self, queue: str, **kwargs) -> dict[str, Any]:
+ # Parse the queue URI
+ parsed = urlparse(queue)
+
+ if not parsed.netloc:
+ raise ValueError(
+ "MQ URI must contain connection id. Expected format:
mq://<conn_id>/<queue_name>"
+ )
+
+ conn_id = parsed.netloc
+
+ queue_name = parsed.path.lstrip("/")
+ if not queue_name:
+ raise ValueError("MQ URI must contain queue name. Expected format:
mq://<conn_id>/<queue_name>")
Review Comment:
Same scheme-rename leftover as the message above (line 74). Update to
`ibmmq://<conn_id>/<queue_name>` so the error message matches the actual scheme.
##########
providers/ibm/mq/provider.yaml:
##########
@@ -0,0 +1,61 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+---
+package-name: apache-airflow-providers-ibm-mq
+name: IBM MQ
+
+state: ready
Review Comment:
**`state: ready` + `Development Status :: 5 - Production/Stable` are
premature for an initial 0.1.0 incubation release.**
`pyproject.toml:41` declares `"Development Status :: 5 - Production/Stable"`
while this file declares `state: ready` and `lifecycle: incubation`. The trio
is internally inconsistent: a 0.1.0 release in `incubation` lifecycle shouldn't
be Production/Stable yet.
Recently-added providers in this repo (e.g. `pydantic-ai`) start at `state:
not-ready` / `Development Status :: 4 - Beta` and graduate via a follow-up PR
once they've shipped at least one minor release with real-world usage. Suggest
matching that pattern for the initial release.
##########
providers/ibm/mq/tests/system/ibm/mq/example_dag_message_queue_trigger.py:
##########
@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+# [START howto_trigger_message_queue]
+from airflow.providers.common.messaging.triggers.msg_queue import
MessageQueueTrigger
+from airflow.sdk import DAG, Asset, AssetWatcher, task
+
+# Define a trigger that listens to an external message queue (IBM MQ in this
case)
+trigger = MessageQueueTrigger(
+ queue="ibmmq://mq_default/MY.QUEUE.NAME",
+)
+
+mq_topic_asset = Asset(
+ "mq_topic_asset",
+ watchers=[AssetWatcher(name="mq_watcher", trigger=trigger)],
+)
+
+with DAG(dag_id="example_ibm_mq_watcher", schedule=[mq_topic_asset]) as dag:
+
+ @task
+ def process_message(**context):
+ for event in context["triggering_asset_events"][mq_topic_asset]:
+ # Get the message from the TriggerEvent payload
+ print("Processing event: ", event)
+ payload = event["payload"]
Review Comment:
**Crashes on every triggered run.**
`triggering_asset_events[mq_topic_asset]` yields `AssetEvent` ORM objects,
not dicts. There is no `["payload"]` key on `AssetEvent`, so this line raises
`TypeError: 'AssetEvent' object is not subscriptable` (or a `KeyError`
depending on context) before the print runs.
The trigger yields `TriggerEvent(event)` where `event` is a `str` (see
`triggers/mq.py:71`). To make this example actually run, either:
1. Yield a structured payload from the trigger: `TriggerEvent({"payload":
event})`, and access via `event.extra["payload"]` from the asset-events side, or
2. Mirror the Kafka `tests/system/kafka` example which iterates
`triggering_asset_events` and uses the documented attribute access.
Right now anyone who copies this DAG to try out the provider hits a runtime
failure on the first message.
##########
providers/ibm/mq/docs/connections/mq.rst:
##########
@@ -0,0 +1,30 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/connection:mq:
Review Comment:
**Sphinx label and title don't reflect the new conn_type.**
The label `_howto/connection:mq:` and the title "MQ connection" predate the
rename to `connection-type: ibmmq` (see `provider.yaml:52`). Anyone trying the
natural cross-ref `:ref:`howto/connection:ibmmq`` from another doc will get a
broken link, and the page title "MQ connection" no longer matches the
vendor-specific naming the rest of the provider uses.
Suggest:
```rst
.. _howto/connection:ibmmq:
IBM MQ connection
=================
```
Also worth renaming the file from `mq.rst` to `ibmmq.rst` for consistency,
though that's a wider rename.
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py:
##########
@@ -0,0 +1,423 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+from contextlib import contextmanager, suppress
+from typing import Any
+
+from asgiref.sync import sync_to_async
+
+from airflow.providers.common.compat.sdk import BaseHook
+
+# Backoff parameters for transient consume failures
+_BACKOFF_BASE: float = 1.0
+_BACKOFF_MAX: float = 60.0
+_BACKOFF_FACTOR: float = 2.0
+
+
+class IBMMQHook(BaseHook):
+ """
+ Interact with IBM MQ queue managers to consume and produce messages.
+
+ This hook wraps the ``ibmmq`` C client and manages connection
+ lifecycle, queue open/close, and message serialization. Both synchronous
+ (context-manager) and asynchronous (``consume`` / ``produce``) interfaces
+ are provided.
+
+ Connection parameters (host, port, login, password) are read from the
+ Airflow connection identified by *conn_id*. ``queue_manager``,
``channel``,
+ and ``open_options`` can be supplied either as constructor arguments or via
+ the connection's *extra* JSON — constructor arguments take precedence.
+
+ :param conn_id: Airflow connection ID for the IBM MQ instance.
+ Defaults to ``"mq_default"``.
+ :param queue_manager: Name of the IBM MQ queue manager to connect to.
+ If not provided, the value is read from the ``queue_manager`` key in
+ the connection's *extra* JSON.
+ :param channel: MQ channel name used for the connection.
+ If not provided, the value is read from the ``channel`` key in the
+ connection's *extra* JSON.
+ :param open_options: Integer bitmask of ``MQOO_*`` open options passed
+ when opening a queue (e.g.,
+ ``ibmmq.CMQC.MQOO_INPUT_SHARED | ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING``).
+ If not provided, the value is resolved from the ``open_options`` key
+ in the connection's *extra* JSON, falling back to
+ ``MQOO_INPUT_SHARED``.
+ """
+
+ conn_name_attr = "conn_id"
+ default_conn_name = "mq_default"
+ conn_type = "ibmmq"
+ hook_name = "IBM MQ"
+ default_open_options = "MQOO_INPUT_SHARED"
+
+ def __init__(
+ self,
+ conn_id: str = default_conn_name,
+ queue_manager: str | None = None,
+ channel: str | None = None,
+ open_options: int | None = None,
+ ):
+ super().__init__()
+ self.conn_id = conn_id
+ self.queue_manager = queue_manager
+ self.channel = channel
+ self.open_options = open_options
+ self._conn = None
+
+ @classmethod
+ def get_ui_field_behaviour(cls) -> dict[str, Any]:
+ """Return custom UI field behaviour for IBM MQ Connection."""
+ return {
+ "hidden_fields": ["schema"],
+ "placeholders": {
+ "host": "mq.example.com",
+ "port": "1414",
+ "login": "app_user",
+ "extra": json.dumps(
+ {
+ "queue_manager": "QM1",
+ "channel": "DEV.APP.SVRCONN",
+ "open_options": cls.default_open_options,
+ },
+ indent=2,
+ ),
+ },
+ }
+
+ @classmethod
+ def get_open_options_flags(cls, open_options: int) -> list[str]:
+ """
+ Return the symbolic MQ open option flags set in a given bitmask.
+
+ Each flag corresponds to a constant in ``ibmmq.CMQC`` that starts with
``MQOO_``.
+
+ :param open_options: The integer bitmask used when opening an MQ queue
+ (e.g., ``MQOO_INPUT_EXCLUSIVE |
MQOO_FAIL_IF_QUIESCING``).
+
+ :return: A list of the names of the MQ open flags that are set in the
bitmask.
+ For example, ``['MQOO_INPUT_EXCLUSIVE',
'MQOO_FAIL_IF_QUIESCING']``.
+
+ Example:
+ >>> open_options = ibmmq.CMQC.MQOO_INPUT_SHARED |
ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING
+ >>> cls.get_open_options_flags(open_options)
+ ['MQOO_INPUT_SHARED', 'MQOO_FAIL_IF_QUIESCING']
+ """
+ import ibmmq
+
+ return [
+ name
+ for name, value in vars(ibmmq.CMQC).items()
+ if name.startswith("MQOO_") and (open_options & value)
+ ]
+
+ @staticmethod
+ def _connect(queue_manager: str, channel: str, conn_info: str, csp):
+ """
+ Connect to the IBM MQ queue manager.
+
+ Connection errors from the C client are caught and re-raised as a
+ :class:`ConnectionError` with a human-readable message.
+
+ :return: IBM MQ connection object
+ """
+ import ibmmq
+
+ try:
+ return ibmmq.connect(queue_manager, channel, conn_info, csp=csp)
+ except (ibmmq.MQMIError, ibmmq.PYIFError) as e:
+ raise ConnectionError(
+ f"Failed to connect to IBM MQ queue manager '{queue_manager}' "
+ f"at {conn_info} on channel '{channel}': {e}"
+ ) from e
+
+ @contextmanager
+ def get_conn(self):
+ """
+ Sync context manager for IBM MQ connection lifecycle.
+
+ Must be called from the executor thread (not the event loop thread).
+ Retrieves the Airflow connection, extracts MQ parameters, and manages
+ the IBM MQ connection lifecycle.
+
+ :yield: IBM MQ connection object
+ """
+ import ibmmq
+
+ connection = BaseHook.get_connection(self.conn_id)
+ config = connection.extra_dejson
+ queue_manager = self.queue_manager or config.get("queue_manager")
+ channel = self.channel or config.get("channel")
+
+ if not queue_manager:
+ raise ValueError("queue_manager must be set in Connection extra
config or hook init")
+ if not channel:
+ raise ValueError("channel must be set in Connection extra config
or hook init")
+
+ if self.open_options is None:
+ self.open_options = getattr(
+ ibmmq.CMQC,
+ config.get("open_options", self.default_open_options),
+ ibmmq.CMQC.MQOO_INPUT_SHARED,
+ )
+
+ csp = ibmmq.CSP()
+ csp.CSPUserId = connection.login
+ csp.CSPPassword = connection.password
+
+ conn_info = f"{connection.host}({connection.port})"
+ conn = self._connect(queue_manager, channel, conn_info, csp)
+ try:
+ yield conn
+ finally:
+ with suppress(Exception):
+ conn.disconnect()
+
+ def _process_message(self, message: bytes) -> str:
+ """
+ Process a raw MQ message.
+
+ If the message contains an RFH2 header, the header is unpacked and the
+ payload following the header is returned. If unpacking fails, the raw
+ message is returned decoded as UTF-8.
+
+ :param message: Raw message received from IBM MQ.
+ :return: Decoded message payload.
+ """
+ import ibmmq
+
+ try:
+ rfh2 = ibmmq.RFH2()
+ rfh2.unpack(message)
+
+ payload_offset = rfh2.get_length()
+ payload = message[payload_offset:]
+
+ decoded = payload.decode("utf-8", errors="ignore")
+ self.log.debug("Message received from MQ (RFH2 decoded): %s",
decoded)
+ return decoded
+ except ibmmq.PYIFError as error: # RFH2 header not present or unpack
failed
+ self.log.warning(
+ "Failed to unpack RFH2 header (%s). Returning raw message
payload: %s", error, message
+ )
+ return message.decode("utf-8", errors="ignore")
+
+ async def aconsume(self, queue_name: str, poll_interval: float = 5) -> str:
+ """
+ Asynchronous version of :meth:`consume`.
+
+ Wait for a single message from the specified IBM MQ queue and return
its decoded payload.
+
+ The method retries with exponential back-off whenever the underlying
+ :meth:`consume` returns ``None`` (connection broken, timeout) or raises
+ an unexpected exception, so that an AssetWatcher is never silently
killed
+ by a transient failure.
+
+ All blocking IBM MQ operations ('connect', 'open', 'get', 'close',
'disconnect') run in a
+ separate thread via 'sync_to_async' to satisfy the IBM MQ C client's
thread-affinity
+ requirement — every operation on a connection must happen from the
thread that created it.
+
+ A :class:`threading.Event` stop signal is passed to the worker so
that, when this
+ coroutine is canceled (e.g. because the Airflow triggerer reassigns
the watcher to
+ another pod), the background thread exits cleanly after the current
'q.get' call
+ times out (at most 'poll_interval' seconds). Without this, an
orphaned thread could
+ silently consume a message after cancellation, causing the event to be
lost and the
+ DAG never to run.
+
+ :param queue_name: Name of the IBM MQ queue to consume messages from.
+ :param poll_interval: Interval in seconds used to wait for messages
and to control
+ how long the underlying MQ 'get' operation blocks before checking
again.
+ :return: The decoded message payload.
+ """
+ import ibmmq
+
+ backoff = _BACKOFF_BASE
+ while True:
+ stop_event = threading.Event()
+ try:
+ result = await sync_to_async(self.consume)(queue_name,
poll_interval, stop_event)
+ except asyncio.CancelledError:
+ stop_event.set()
+ raise
+ except ibmmq.MQMIError as e:
+ if e.reason == ibmmq.CMQC.MQRC_CONNECTION_BROKEN:
Review Comment:
**Retry treats only `MQRC_CONNECTION_BROKEN` (2009) as transient; other
recoverable codes kill the watcher.**
Several MQ reason codes are routinely transient and should retry:
- `MQRC_Q_MGR_QUIESCING` (2161) -- queue manager shutting down for restart
- `MQRC_Q_MGR_NOT_AVAILABLE` (2059) -- QM not yet up after a restart
- `MQRC_HOST_NOT_AVAILABLE` (2538) -- network blip
- `MQRC_CONNECTION_QUIESCING` (2202) -- channel restart
These currently hit the `raise` on line 280 (`-- not retrying`) and kill the
AssetWatcher even though they recover within seconds. Suggest:
```python
_TRANSIENT_REASONS = frozenset({
ibmmq.CMQC.MQRC_CONNECTION_BROKEN,
ibmmq.CMQC.MQRC_Q_MGR_QUIESCING,
ibmmq.CMQC.MQRC_Q_MGR_NOT_AVAILABLE,
ibmmq.CMQC.MQRC_HOST_NOT_AVAILABLE,
ibmmq.CMQC.MQRC_CONNECTION_QUIESCING,
})
...
if e.reason in _TRANSIENT_REASONS:
...retry...
```
Defining the set as a module-level constant also makes it easy for users to
extend via subclass if they hit something exotic.
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py:
##########
@@ -0,0 +1,423 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+from contextlib import contextmanager, suppress
+from typing import Any
+
+from asgiref.sync import sync_to_async
+
+from airflow.providers.common.compat.sdk import BaseHook
+
+# Backoff parameters for transient consume failures
+_BACKOFF_BASE: float = 1.0
+_BACKOFF_MAX: float = 60.0
+_BACKOFF_FACTOR: float = 2.0
+
+
+class IBMMQHook(BaseHook):
+ """
+ Interact with IBM MQ queue managers to consume and produce messages.
+
+ This hook wraps the ``ibmmq`` C client and manages connection
+ lifecycle, queue open/close, and message serialization. Both synchronous
+ (context-manager) and asynchronous (``consume`` / ``produce``) interfaces
+ are provided.
+
+ Connection parameters (host, port, login, password) are read from the
+ Airflow connection identified by *conn_id*. ``queue_manager``,
``channel``,
+ and ``open_options`` can be supplied either as constructor arguments or via
+ the connection's *extra* JSON — constructor arguments take precedence.
+
+ :param conn_id: Airflow connection ID for the IBM MQ instance.
+ Defaults to ``"mq_default"``.
+ :param queue_manager: Name of the IBM MQ queue manager to connect to.
+ If not provided, the value is read from the ``queue_manager`` key in
+ the connection's *extra* JSON.
+ :param channel: MQ channel name used for the connection.
+ If not provided, the value is read from the ``channel`` key in the
+ connection's *extra* JSON.
+ :param open_options: Integer bitmask of ``MQOO_*`` open options passed
+ when opening a queue (e.g.,
+ ``ibmmq.CMQC.MQOO_INPUT_SHARED | ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING``).
+ If not provided, the value is resolved from the ``open_options`` key
+ in the connection's *extra* JSON, falling back to
+ ``MQOO_INPUT_SHARED``.
+ """
+
+ conn_name_attr = "conn_id"
+ default_conn_name = "mq_default"
+ conn_type = "ibmmq"
+ hook_name = "IBM MQ"
+ default_open_options = "MQOO_INPUT_SHARED"
+
+ def __init__(
+ self,
+ conn_id: str = default_conn_name,
+ queue_manager: str | None = None,
+ channel: str | None = None,
+ open_options: int | None = None,
+ ):
+ super().__init__()
+ self.conn_id = conn_id
+ self.queue_manager = queue_manager
+ self.channel = channel
+ self.open_options = open_options
+ self._conn = None
+
+ @classmethod
+ def get_ui_field_behaviour(cls) -> dict[str, Any]:
+ """Return custom UI field behaviour for IBM MQ Connection."""
+ return {
+ "hidden_fields": ["schema"],
+ "placeholders": {
+ "host": "mq.example.com",
+ "port": "1414",
+ "login": "app_user",
+ "extra": json.dumps(
+ {
+ "queue_manager": "QM1",
+ "channel": "DEV.APP.SVRCONN",
+ "open_options": cls.default_open_options,
+ },
+ indent=2,
+ ),
+ },
+ }
+
+ @classmethod
+ def get_open_options_flags(cls, open_options: int) -> list[str]:
+ """
+ Return the symbolic MQ open option flags set in a given bitmask.
+
+ Each flag corresponds to a constant in ``ibmmq.CMQC`` that starts with
``MQOO_``.
+
+ :param open_options: The integer bitmask used when opening an MQ queue
+ (e.g., ``MQOO_INPUT_EXCLUSIVE |
MQOO_FAIL_IF_QUIESCING``).
+
+ :return: A list of the names of the MQ open flags that are set in the
bitmask.
+ For example, ``['MQOO_INPUT_EXCLUSIVE',
'MQOO_FAIL_IF_QUIESCING']``.
+
+ Example:
+ >>> open_options = ibmmq.CMQC.MQOO_INPUT_SHARED |
ibmmq.CMQC.MQOO_FAIL_IF_QUIESCING
+ >>> cls.get_open_options_flags(open_options)
+ ['MQOO_INPUT_SHARED', 'MQOO_FAIL_IF_QUIESCING']
+ """
+ import ibmmq
+
+ return [
+ name
+ for name, value in vars(ibmmq.CMQC).items()
+ if name.startswith("MQOO_") and (open_options & value)
+ ]
+
+ @staticmethod
+ def _connect(queue_manager: str, channel: str, conn_info: str, csp):
+ """
+ Connect to the IBM MQ queue manager.
+
+ Connection errors from the C client are caught and re-raised as a
+ :class:`ConnectionError` with a human-readable message.
+
+ :return: IBM MQ connection object
+ """
+ import ibmmq
+
+ try:
+ return ibmmq.connect(queue_manager, channel, conn_info, csp=csp)
+ except (ibmmq.MQMIError, ibmmq.PYIFError) as e:
+ raise ConnectionError(
+ f"Failed to connect to IBM MQ queue manager '{queue_manager}' "
+ f"at {conn_info} on channel '{channel}': {e}"
+ ) from e
+
+ @contextmanager
+ def get_conn(self):
+ """
+ Sync context manager for IBM MQ connection lifecycle.
+
+ Must be called from the executor thread (not the event loop thread).
+ Retrieves the Airflow connection, extracts MQ parameters, and manages
+ the IBM MQ connection lifecycle.
+
+ :yield: IBM MQ connection object
+ """
+ import ibmmq
+
+ connection = BaseHook.get_connection(self.conn_id)
+ config = connection.extra_dejson
+ queue_manager = self.queue_manager or config.get("queue_manager")
+ channel = self.channel or config.get("channel")
+
+ if not queue_manager:
+ raise ValueError("queue_manager must be set in Connection extra
config or hook init")
+ if not channel:
+ raise ValueError("channel must be set in Connection extra config
or hook init")
+
+ if self.open_options is None:
+ self.open_options = getattr(
+ ibmmq.CMQC,
+ config.get("open_options", self.default_open_options),
+ ibmmq.CMQC.MQOO_INPUT_SHARED,
+ )
+
+ csp = ibmmq.CSP()
+ csp.CSPUserId = connection.login
+ csp.CSPPassword = connection.password
+
+ conn_info = f"{connection.host}({connection.port})"
+ conn = self._connect(queue_manager, channel, conn_info, csp)
+ try:
+ yield conn
+ finally:
+ with suppress(Exception):
+ conn.disconnect()
+
+ def _process_message(self, message: bytes) -> str:
+ """
+ Process a raw MQ message.
+
+ If the message contains an RFH2 header, the header is unpacked and the
+ payload following the header is returned. If unpacking fails, the raw
+ message is returned decoded as UTF-8.
+
+ :param message: Raw message received from IBM MQ.
+ :return: Decoded message payload.
+ """
+ import ibmmq
+
+ try:
+ rfh2 = ibmmq.RFH2()
+ rfh2.unpack(message)
+
+ payload_offset = rfh2.get_length()
+ payload = message[payload_offset:]
+
+ decoded = payload.decode("utf-8", errors="ignore")
+ self.log.debug("Message received from MQ (RFH2 decoded): %s",
decoded)
+ return decoded
+ except ibmmq.PYIFError as error: # RFH2 header not present or unpack
failed
+ self.log.warning(
+ "Failed to unpack RFH2 header (%s). Returning raw message
payload: %s", error, message
+ )
+ return message.decode("utf-8", errors="ignore")
+
+ async def aconsume(self, queue_name: str, poll_interval: float = 5) -> str:
+ """
+ Asynchronous version of :meth:`consume`.
+
+ Wait for a single message from the specified IBM MQ queue and return
its decoded payload.
+
+ The method retries with exponential back-off whenever the underlying
+ :meth:`consume` returns ``None`` (connection broken, timeout) or raises
+ an unexpected exception, so that an AssetWatcher is never silently
killed
+ by a transient failure.
+
+ All blocking IBM MQ operations ('connect', 'open', 'get', 'close',
'disconnect') run in a
+ separate thread via 'sync_to_async' to satisfy the IBM MQ C client's
thread-affinity
+ requirement — every operation on a connection must happen from the
thread that created it.
+
+ A :class:`threading.Event` stop signal is passed to the worker so
that, when this
+ coroutine is canceled (e.g. because the Airflow triggerer reassigns
the watcher to
+ another pod), the background thread exits cleanly after the current
'q.get' call
+ times out (at most 'poll_interval' seconds). Without this, an
orphaned thread could
+ silently consume a message after cancellation, causing the event to be
lost and the
+ DAG never to run.
+
+ :param queue_name: Name of the IBM MQ queue to consume messages from.
+ :param poll_interval: Interval in seconds used to wait for messages
and to control
+ how long the underlying MQ 'get' operation blocks before checking
again.
+ :return: The decoded message payload.
+ """
+ import ibmmq
+
+ backoff = _BACKOFF_BASE
+ while True:
+ stop_event = threading.Event()
+ try:
+ result = await sync_to_async(self.consume)(queue_name,
poll_interval, stop_event)
+ except asyncio.CancelledError:
+ stop_event.set()
+ raise
+ except ibmmq.MQMIError as e:
+ if e.reason == ibmmq.CMQC.MQRC_CONNECTION_BROKEN:
+ self.log.warning(
+ "Transient MQ error on queue '%s': completion_code=%s
reason_code=%s (%s); retrying in %.1fs",
+ queue_name,
+ e.comp,
+ e.reason,
+ e,
+ backoff,
+ )
+ await asyncio.sleep(backoff)
+ backoff = min(backoff * _BACKOFF_FACTOR, _BACKOFF_MAX)
+ continue
+ self.log.error(
+ "Permanent MQ error on queue '%s': completion_code=%s
reason_code=%s (%s) -- not retrying",
+ queue_name,
+ e.comp,
+ e.reason,
+ e,
+ )
+ raise
+ except ibmmq.PYIFError as e:
+ self.log.error(
+ "PYIFError on queue '%s': %s -- not retrying",
+ queue_name,
+ e,
+ )
+ raise
+ except Exception as e:
+ # Programming errors should not be retried
+ self.log.error(
+ "Unexpected error in IBM MQ consume for queue '%s': %s --
not retrying",
+ queue_name,
+ e,
+ exc_info=True,
+ )
+ raise
+
+ if result is not None:
+ stop_event.set()
+ return result
+
+ self.log.debug(
+ "IBM MQ consume returned no event for queue '%s'; queue may be
quiet. Retrying in %.1fs",
+ queue_name,
+ backoff,
+ )
+ await asyncio.sleep(backoff)
+ backoff = min(backoff * _BACKOFF_FACTOR, _BACKOFF_MAX)
+
+ def consume(
+ self,
+ queue_name: str,
+ poll_interval: float,
+ stop_event: threading.Event,
+ ) -> str | None:
+ """
+ Blocking implementation that consumes a single message from the given
IBM MQ queue.
+
+ All IBM MQ handles (queue manager connection, queue) are created **and
used** within
+ this method, satisfying the thread-affinity requirement of the IBM MQ
C client library.
+ The 'stop_event' is checked between 'q.get' calls so the thread
terminates promptly
+ after the coroutine side is canceled.
+
+ For an asynchronous interface see :meth:`aconsume`.
+ """
+ import ibmmq
+
+ od = ibmmq.OD()
+ od.ObjectName = queue_name
+
+ md = ibmmq.MD()
+ md.Format = ibmmq.CMQC.MQFMT_STRING
+ md.CodedCharSetId = 1208
+ md.Encoding = ibmmq.CMQC.MQENC_NATIVE
+
+ gmo = ibmmq.GMO()
+ gmo.Options = ibmmq.CMQC.MQGMO_WAIT | ibmmq.CMQC.MQGMO_NO_SYNCPOINT |
ibmmq.CMQC.MQGMO_CONVERT
Review Comment:
**`MQGMO_NO_SYNCPOINT` autocommits the get; cancellation can lose messages.**
With `NO_SYNCPOINT`, the moment `q.get` returns the message is committed off
the queue. If `aconsume` is cancelled (triggerer reassigning the watcher to
another pod, or a `CancelledError` fired during `asyncio.sleep`) while `q.get`
is mid-flight or just after it returns but before `_process_message` yields,
the message is gone from the queue but no `TriggerEvent` is ever emitted. The
DAG never runs for that message.
This is exactly the silent-loss failure mode the `stop_event` machinery in
`aconsume` is supposed to prevent, but `NO_SYNCPOINT` short-circuits it.
Options:
1. Use `MQGMO_SYNCPOINT` and commit only after the `TriggerEvent` is yielded
(strict at-least-once).
2. Keep `NO_SYNCPOINT` but document the at-most-once guarantee prominently
in the trigger / hook docstrings so users can decide whether MQ is appropriate
for their workload.
Either is acceptable, but the current behaviour is undocumented at-most-once
with an inviting `stop_event` API that suggests otherwise.
##########
providers/ibm/mq/src/airflow/providers/ibm/mq/queues/mq.py:
##########
@@ -0,0 +1,87 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+from airflow.providers.common.messaging.providers.base_provider import
BaseMessageQueueProvider
+from airflow.providers.ibm.mq.triggers.mq import AwaitMessageTrigger
+
+if TYPE_CHECKING:
+ from airflow.triggers.base import BaseEventTrigger
+
+# [START queue_regexp]
+# Matches ibmmq://<conn_id>/<queue_name> (conn_id required, queue_name
required)
+QUEUE_REGEXP = r"^ibmmq://[^/]+/.+"
+# [END queue_regexp]
+
+
+class IBMMQMessageQueueProvider(BaseMessageQueueProvider):
+ """
+ Configuration for IBM MQ integration with common-messaging.
+
+ [START ibmmq_message_queue_provider_description]
+
+ * It uses ``ibmmq`` as scheme for identifying IBM MQ queues.
+ * For parameter definitions take a look at
+ :class:`~airflow.providers.ibm.mq.triggers.mq.AwaitMessageTrigger`.
+
+ .. code-block:: python
+
+ from airflow.providers.common.messaging.triggers.msg_queue import
MessageQueueTrigger
+ from airflow.sdk import Asset, AssetWatcher
+
+ trigger = MessageQueueTrigger(
+ queue="ibmmq://mq_default/MY.QUEUE.NAME",
+ )
+
+ asset = Asset("mq_topic_asset",
watchers=[AssetWatcher(name="mq_watcher", trigger=trigger)])
+
+ [END ibmmq_message_queue_provider_description]
+ """
+
+ scheme = "ibmmq"
+
+ def queue_matches(self, queue: str) -> bool:
+ return bool(re.match(QUEUE_REGEXP, queue))
+
+ def trigger_class(self) -> type[BaseEventTrigger]:
+ return AwaitMessageTrigger # type: ignore[return-value]
+
+ def trigger_kwargs(self, queue: str, **kwargs) -> dict[str, Any]:
+ # Parse the queue URI
+ parsed = urlparse(queue)
+
+ if not parsed.netloc:
+ raise ValueError(
+ "MQ URI must contain connection id. Expected format:
mq://<conn_id>/<queue_name>"
+ )
+
+ conn_id = parsed.netloc
+
+ queue_name = parsed.path.lstrip("/")
+ if not queue_name:
+ raise ValueError("MQ URI must contain queue name. Expected format:
mq://<conn_id>/<queue_name>")
+
+ return {
+ "mq_conn_id": conn_id,
+ "queue_name": queue_name,
+ "poll_interval": kwargs.get("poll_interval", 5),
+ }
Review Comment:
**`open_options` not forwarded to the trigger.**
`trigger_kwargs` extracts `mq_conn_id`, `queue_name`, and `poll_interval`,
but swallows `open_options`. So a user who wants to consume in
`MQOO_INPUT_EXCLUSIVE` mode via the unified
`MessageQueueTrigger(queue="ibmmq://...")` form has no way to pass it -- they
have to drop down to instantiating `AwaitMessageTrigger` directly, which
defeats the point of the `common.messaging` abstraction.
Either forward `kwargs.get("open_options")` or document explicitly that
`open_options` must be set via the connection extras (and that
`MessageQueueTrigger` kwargs are not propagated).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
