kaxil commented on code in PR #62790: URL: https://github.com/apache/airflow/pull/62790#discussion_r3118814317
########## providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py: ########## @@ -0,0 +1,404 @@ +# +# 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 = "mq" + 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 not self.open_options: + 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. + """ + backoff = _BACKOFF_BASE + stop_event = threading.Event() Review Comment: **Critical: trigger gets permanently wedged after any transient failure.** `stop_event` is created once here, outside the `while True:` retry loop, and the `finally:` at line 268 (`stop_event.set()`) fires on every iteration. So after the first time `consume()` returns (whether it returned `None`, raised an exception, or succeeded), `stop_event.is_set()` is `True` forever on this `aconsume` invocation. On the next retry, `consume()` enters its `while not stop_event.is_set():` loop at line 339 and exits immediately -- it never calls `q.get()`. It just returns `None`, which `aconsume` treats as "retry with backoff". The trigger now loops forever sleeping and logging `"returned no event"`, never actually reading from the queue. Real messages are lost and the AssetWatcher is effectively dead. The user-visible symptom is a silently broken trigger after any transient MQ blip. The existing test `test_aconsume_retries_on_none_then_succeeds` misses this because it patches `self.hook.consume` directly, bypassing the real `consume()` body that checks `stop_event`. **Fix:** either create a fresh `stop_event = threading.Event()` at the top of each loop iteration, or move the `stop_event.set()` out of the retry `finally` and only set it when `aconsume` itself is canceled/returns. The former is simpler. The `stop_event` contract is "tell the worker thread to stop for *this* call", not "mark this hook instance as stopped forever". ########## 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( + scheme="mq://mq_default/MY.QUEUE.NAME", Review Comment: **Critical: this example raises `ValueError` on every run.** `MessageQueueTrigger(scheme="mq://mq_default/MY.QUEUE.NAME")` passes a full URI to the `scheme` kwarg. In `common.messaging`, the provider match is `scheme_matches = (self.scheme == scheme)`, and this provider declares `scheme = "mq"`. The full URI is not equal to `"mq"`, so `MessageQueueTrigger.__init__` raises `ValueError("scheme '...' is not recognized")` before the DAG ever parses. The same snippet is copied into `docs/message-queues.rst`, so the documented onboarding path is also broken. The docstring in `queues/mq.py:51` shows the correct form: `queue="mq://mq_default/MY.QUEUE.NAME"`. Please pick one form and apply it consistently across the example DAG, the RST docs, and the provider docstring. I'd verify by running `pytest providers/ibm/mq/tests/system/...` with a mocked `MessageQueueTrigger` import path before merge. ########## providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py: ########## @@ -0,0 +1,404 @@ +# +# 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 = "mq" + 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 not self.open_options: Review Comment: **`MQOO_NONE = 0` is a valid bitmask that this branch treats as unset.** `if not self.open_options:` is truthy for both `None` and `0`. `MQOO_NONE` is a legitimate IBM MQ value (used for server-side overrides), so a user who explicitly sets `open_options=0` silently gets overwritten with the default. Use `if self.open_options is None:` instead. Separately, this block mutates `self.open_options` as a side effect of `get_conn()`. That's surprising -- calling `get_conn()` changes instance state. Consider resolving to a local variable (e.g. `resolved_open_options`) and threading it through `consume()` / `produce()` explicitly, so the hook stays idempotent across calls. ########## providers/ibm/mq/src/airflow/providers/ibm/mq/queues/mq.py: ########## @@ -0,0 +1,86 @@ +# +# 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] +QUEUE_REGEXP = r"^mq://" +# [END queue_regexp] + + +class IBMMQMessageQueueProvider(BaseMessageQueueProvider): + """ + Configuration for IBM MQ integration with common-messaging. + + [START ibmmq_message_queue_provider_description] + + * It uses ``mq`` 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="mq://mq_default/MY.QUEUE.NAME", Review Comment: **Docstring form and system-test DAG form disagree.** This docstring shows `MessageQueueTrigger(queue="mq://...")`, which is the form that actually works (the `queue` kwarg is what `common.messaging` parses). The system test DAG at `providers/ibm/mq/tests/system/ibm/mq/example_dag_message_queue_trigger.py:25` and the RST docs use `scheme="mq://..."`, which raises `ValueError` -- see separate comment on the example DAG. Please align all three (docstring, example DAG, RST) on a single form. ########## providers/ibm/mq/tests/unit/ibm/mq/hooks/test_mq.py: ########## @@ -0,0 +1,248 @@ +# 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 +from unittest import mock +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from airflow.models import Connection +from airflow.providers.ibm.mq.hooks.mq import ( + _BACKOFF_BASE, + _BACKOFF_FACTOR, + _BACKOFF_MAX, + IBMMQHook, +) + +MQ_PAYLOAD = """RFH x"MQSTR <mcd><Msd>jms_map</Msd></mcd> <jms><Dst>topic://localhost/topic</Dst><Tms>1772121947476</Tms><Dlv>2</Dlv><Uci dt='bin.hex'>414D5143514D4941303054202020202069774D7092F81057</Uci></jms>L<usr><XMSC_CLIENT_ID>local</XMSC_CLIENT_ID><release>26.01.00</release></usr> 4<mqps><Top>topic</Top></mqps> {}""" + + +def mq_connection(): + """Create a test MQ connection object.""" + return Connection( + conn_id="mq_conn", + conn_type="mq", + host="mq.example.com", + login="user", + password="pass", + port=1414, + extra='{"queue_manager": "QM1", "channel": "DEV.APP.SVRCONN"}', + ) + + [email protected] +def mock_get_connection(): + """Fixture that mocks BaseHook.get_connection to return a test connection.""" + with patch("airflow.providers.ibm.mq.hooks.mq.BaseHook.get_connection") as mock_conn: + mock_conn.return_value = mq_connection() + yield mock_conn + + [email protected] +def patch_sync_to_async(): + """Patch sync_to_async to call the wrapped function directly for testing.""" + + def sync_to_async(func): + """Wrap a sync function so it can be awaited directly.""" + + async def wrapper(*args, **kwargs): + return func(*args, **kwargs) + + return wrapper + + with patch("airflow.providers.ibm.mq.hooks.mq.sync_to_async", side_effect=sync_to_async): + yield + + +def fake_get(*args, **kwargs): + import ibmmq + + raise ibmmq.MQMIError("connection broken", reason=ibmmq.CMQC.MQRC_CONNECTION_BROKEN) + + [email protected] +class TestIBMMQHook: + """Tests for the IBM MQ hook.""" + + @pytest.fixture(autouse=True) + def setup_connections(self, create_connection_without_db): + create_connection_without_db( + Connection( + conn_id="mq_conn", + conn_type="mq", + host="mq.example.com", + login="user", + password="pass", + port=1414, + extra='{"queue_manager": "QM1", "channel": "DEV.APP.SVRCONN"}', + ) + ) + self.hook = IBMMQHook("mq_conn") + + @patch("ibmmq.connect") + @patch("ibmmq.Queue") + async def test_aconsume_message( + self, mock_queue_class, mock_connect, mock_get_connection, patch_sync_to_async + ): + """Test consuming a single message.""" + + mock_qmgr = MagicMock() + mock_connect.return_value = mock_qmgr + + mock_queue = MagicMock() + mock_queue_class.return_value = mock_queue + mock_queue.get.return_value = MQ_PAYLOAD.format("test message").encode() + + result = await self.hook.aconsume(queue_name="QUEUE1", poll_interval=0.1) + assert isinstance(result, str) + assert "test message" in result + + mock_connect.assert_called_once() + mock_queue_class.assert_called_once_with( + mock_qmgr, + mock.ANY, + mock.ANY, + ) + + @patch("ibmmq.connect") + @patch("ibmmq.Queue") + async def test_aproduce_message( + self, mock_queue_class, mock_connect, mock_get_connection, patch_sync_to_async + ): + """Test producing a message to the queue.""" + + mock_qmgr = MagicMock() + mock_connect.return_value = mock_qmgr + + mock_queue = MagicMock() + mock_queue_class.return_value = mock_queue + + await self.hook.aproduce(queue_name="QUEUE1", payload="payload") + + mock_connect.assert_called_once() + mock_queue_class.assert_called_once_with( + mock_qmgr, + mock.ANY, + mock.ANY, + ) + mock_queue.put.assert_called_once() + + @patch("ibmmq.connect") + @patch("ibmmq.Queue") + async def test_aconsume_connection_broken( + self, mock_queue_class, mock_connect, mock_get_connection, patch_sync_to_async, caplog + ): + """Test that consume logs a warning on connection broken.""" + + mock_qmgr = MagicMock() + mock_connect.return_value = mock_qmgr + mock_queue = MagicMock() + mock_queue_class.return_value = mock_queue + mock_queue.get.side_effect = fake_get + + # aconsume() retries on None, so we need to cancel after the first attempt + with patch("airflow.providers.ibm.mq.hooks.mq.asyncio.sleep", new_callable=AsyncMock) as mock_sleep: + mock_sleep.side_effect = asyncio.CancelledError + with pytest.raises(asyncio.CancelledError): + await self.hook.aconsume(queue_name="QUEUE1", poll_interval=0.1) + + assert "MQ connection broken on queue 'QUEUE1'; will reconnect" in caplog.text + + @patch("airflow.providers.ibm.mq.hooks.mq.asyncio.sleep", new_callable=AsyncMock) + async def test_aconsume_retries_on_none_then_succeeds(self, mock_sleep, patch_sync_to_async): + """When consume returns None, aconsume retries with backoff until a message arrives.""" + with patch.object(self.hook, "consume", side_effect=[None, None, "payload after retries"]): Review Comment: **This test can't catch the `stop_event` reuse bug.** It patches `self.hook.consume` with `side_effect=[None, None, "payload after retries"]`, which bypasses the real `consume()` body -- so the `while not stop_event.is_set():` loop never runs, and the bug described on `hooks/mq.py:251` is invisible. Please add an integration-style test that exercises the real `consume()` at least once: patch `ibmmq.connect` / `ibmmq.Queue.get` to return `None` on the first call (simulating broken connection) then a real payload on the second, and assert `aconsume` ultimately returns the payload. That test fails today on the current `stop_event` placement, and passing it becomes the acceptance criterion for the fix. (Good that the existing retry-logic tests are in place -- they catch the happy path. This is about adding the one test that would have caught the core bug.) ########## providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py: ########## @@ -0,0 +1,404 @@ +# +# 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 = "mq" Review Comment: **`conn_type = "mq"` is too generic for a vendor-specific hook.** "mq" is a category, not a product. Airflow already has several MQ-family providers (Kafka, SQS, Redis streams, NATS in draft), and they all namespace to the vendor/protocol: `conn_type = "kafka"`, `"aws"`, `"redis"`. If a future Rabbit / ActiveMQ / Solace provider lands, they can't take a natural `conn_type` without colliding with this one. Suggest `conn_type = "ibm_mq"` (matching the package path `airflow.providers.ibm.mq`). The `hook_name = "IBM MQ"` already implies the right UI label. This is much cheaper to change now than after the first release. ########## providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py: ########## @@ -0,0 +1,404 @@ +# +# 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 = "mq" + 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 not self.open_options: + 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. + """ + backoff = _BACKOFF_BASE + stop_event = threading.Event() + while True: + try: + result = await sync_to_async(self.consume)(queue_name, poll_interval, stop_event) + except asyncio.CancelledError: + raise + except Exception: Review Comment: **Unbounded retry on `except Exception` masks permanent failures.** This catches everything non-`CancelledError` and keeps retrying with backoff capped at `_BACKOFF_MAX=60s`. That's right for transient failures (connection blip, queue-manager restart), but it's wrong for permanent ones: - `MQRC_NOT_AUTHORIZED` (2035) -- bad credentials; will never succeed - `MQRC_UNKNOWN_OBJECT_NAME` (2085) -- queue doesn't exist; will never succeed - `MQRC_Q_MGR_NAME_ERROR` (2058) -- misconfigured queue manager - Programming bugs (`TypeError`, `AttributeError`) from the hook itself Today, all of these produce a WARNING log every 60 seconds forever. The AssetWatcher looks alive but is silently broken, and there's no signal to ops other than watching logs. **Fix:** either classify MQ reason codes into retryable vs permanent (the common MQ clients all do this) and raise immediately on permanent ones, or add a `max_retries` / `max_retry_duration` parameter with a sensible default (e.g. 5 minutes). The current behavior trades one silent-failure mode (swallowed exception) for another (infinite retry on a doomed operation). ########## providers/ibm/mq/src/airflow/providers/ibm/mq/queues/mq.py: ########## @@ -0,0 +1,86 @@ +# +# 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] +QUEUE_REGEXP = r"^mq://" Review Comment: **`QUEUE_REGEXP = r"^mq://"` matches any `mq://` URI with no structural check.** Combined with the generic `conn_type = "mq"` (see separate comment), this means a future MQ-family provider can't coexist: both would claim every `mq://...` URI and `common.messaging` would pick one by import order. Also the regexp doesn't actually validate what `trigger_kwargs` later requires (`mq://<conn_id>/<queue_name>`). A user passing `"mq://"` or `"mq:///queue"` gets through `queue_matches` and then fails later with a less helpful error. **Fix:** tighten to something like `r"^mq://[^/]+/.+"` and namespace the scheme (e.g. `ibm-mq://` or `ibmmq://`). The tighter regex also lets `queue_matches` do useful negative matching instead of a prefix check. ########## providers/ibm/mq/src/airflow/providers/ibm/mq/hooks/mq.py: ########## @@ -0,0 +1,404 @@ +# +# 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 = "mq" + 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 not self.open_options: + 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. + """ + backoff = _BACKOFF_BASE + stop_event = threading.Event() + while True: + try: + result = await sync_to_async(self.consume)(queue_name, poll_interval, stop_event) + except asyncio.CancelledError: + raise + except Exception: + self.log.warning( + "IBM MQ consume encountered an error for queue '%s'; retrying in %.1fs", + queue_name, + backoff, + exc_info=True, + ) + await asyncio.sleep(backoff) + backoff = min(backoff * _BACKOFF_FACTOR, _BACKOFF_MAX) + continue + finally: + stop_event.set() + + if result is not None: + return result + + self.log.warning( + "IBM MQ consume returned no event for queue '%s'; retrying in %.1fs", Review Comment: **WARNING-level log every `poll_interval` on quiet queues.** When `consume()` returns `None`, this warning fires and the hook sleeps+retries. But `consume()` returns `None` both on genuine connection-broken (line ~336) *and* on clean `MQMIError`/`PYIFError` exits (line 357). A quiet but healthy queue that hits `MQRC_CONNECTION_BROKEN` and reconnects will spam WARNING every few seconds, drowning real signal. Also, from the caller's perspective the two failure modes are indistinguishable: "consume returned no event" could mean "broken connection" or "swallowed MQ error". Operators can't tell whether to page. **Fix options:** (a) log the two cases differently inside `consume()` (broken-connection stays WARNING, unexpected MQ error stays ERROR, and `aconsume` drops to DEBUG for the generic retry message); or (b) return a small enum/tuple from `consume()` so `aconsume` can pick the right log level. Either is better than a uniform WARNING. -- 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]
