dabla commented on code in PR #62790: URL: https://github.com/apache/airflow/pull/62790#discussion_r3121683864
########## 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: Refactored test to make sure stop_event is being taken into account to avoid regression in the future ########## 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: Done -- 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]
