Removed unused code

Author: sujata-m

src/python_ms_core/core/topic/azure_topic.py

@@ -1,5 +1,4 @@
 import json
-import gc
 import logging
 import multiprocessing as mp
 import os
@@ -13,15 +12,8 @@
 from azure.servicebus import AutoLockRenewer
 import threading
 
-try:
-    import psutil
-except ImportError:  # pragma: no cover - dependency exists in the package, but keep logging resilient.
-    psutil = None
 
-
-logging.basicConfig(format='%(asctime)s %(levelname)-8s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
 logger = logging.getLogger('AzureTopic')
-logger.setLevel(logging.INFO)
 
 
 """
@@ -59,14 +51,8 @@ def __init__(self, config: TopicConfig=None, topic_name=None, max_concurrent_mes
         self.callback_process_start_method = self._get_process_start_method()
         self.process_context = self._get_process_context()
         self.internal_count = 0
-        self.max_renewal_duration = self._get_positive_int_from_env(
-            'TOPIC_MAX_LOCK_RENEWAL_DURATION_SECONDS',
-            86400,
-        )  # Renew the message upto 1 day
-        self.lock_renewal_margin = self._get_positive_int_from_env(
-            'TOPIC_LOCK_RENEWAL_MARGIN_SECONDS',
-            60,
-        )
+        self.max_renewal_duration = 86400  # Renew the message upto 1 day
+        self.lock_renewal_margin = 60
         renewer_max_workers = max(max_concurrent_messages, 2)
         self.lock_renewal = AutoLockRenewer(
             max_lock_renewal_duration=self.max_renewal_duration,
@@ -76,22 +62,9 @@ def __init__(self, config: TopicConfig=None, topic_name=None, max_concurrent_mes
         # The SDK default renews only in the last 10 seconds of the lock window.
         # Start earlier so long-running jobs have more headroom for scheduler jitter.
         self.lock_renewal._renew_period = min(self.lock_renewal_margin, self.max_renewal_duration)
-        self.lock_renew_receiver = _LockRenewLoggingReceiver(self)
         self.wait_time_for_message = 5
         self.thread_lock = threading.Lock()
         self.pending_tasks = []
-        self._process = None
-        self._prime_runtime_samplers()
-        logger.info(
-            'Configured AzureTopic %s: execution_mode=%s, process_start_method=%s, '
-            'max_lock_renewal_duration=%s seconds, renew_margin=%s seconds, renewer_max_workers=%s',
-            self.topic_name,
-            self.callback_execution_mode,
-            self.callback_process_start_method,
-            self.max_renewal_duration,
-            self.lock_renewal_margin,
-            renewer_max_workers,
-        )
 
 
     def publish(self, data: QueueMessage):
@@ -135,7 +108,7 @@ def subscribe(self, subscription: str, callback, max_receivable_messages=-1):
                         for message in messages: 
                             execution_task = self._submit_processing_task(message, callback)
                             self.lock_renewal.register(
-                                self.lock_renew_receiver,
+                                self.receiver,
                                 message,
                                 max_lock_renewal_duration=self.max_renewal_duration,
                                 on_lock_renew_failure=self._handle_lock_renew_failure,
@@ -283,88 +256,13 @@ def _handle_lock_renew_failure(self, renewable, error):
         failure_reason = error or getattr(renewable, 'auto_renew_error', None) or 'lock expired before renewal could complete'
         logger.error(
             f'Error renewing lock for message {message_id}: {failure_reason}; '
-            f'locked_until_utc={getattr(renewable, "locked_until_utc", None)}; '
-            f'runtime_snapshot={self._get_runtime_snapshot()}'
+            f'locked_until_utc={getattr(renewable, "locked_until_utc", None)}'
         )
 
     @staticmethod
     def _get_message_id(message):
         return getattr(message, 'message_id', None) or getattr(message, 'messageId', 'unknown')
 
-    def _prime_runtime_samplers(self):
-        if psutil is None:
-            return
-        try:
-            self._process = psutil.Process(os.getpid())
-            self._process.cpu_percent(interval=None)
-            psutil.cpu_percent(interval=None)
-        except Exception:  # pragma: no cover - best effort diagnostics
-            self._process = None
-
-    def _get_runtime_snapshot(self):
-        return f'{self._get_memory_snapshot()}, {self._get_cpu_snapshot()}, {self._get_gc_snapshot()}'
-
-    def _get_memory_snapshot(self):
-        if self._process is None:
-            return 'memory=psutil-unavailable'
-        try:
-            memory_info = self._process.memory_info()
-            rss_mb = memory_info.rss / (1024 * 1024)
-            vms_mb = memory_info.vms / (1024 * 1024)
-            return f'memory=rss_mb={rss_mb:.2f}, vms_mb={vms_mb:.2f}, num_threads={self._process.num_threads()}'
-        except Exception as exc:  # pragma: no cover - diagnostic fallback
-            return f'memory=unavailable({exc})'
-
-    def _get_cpu_snapshot(self):
-        if self._process is None:
-            return 'cpu=psutil-unavailable'
-        try:
-            process_cpu_percent = self._process.cpu_percent(interval=None)
-            system_cpu_percent = psutil.cpu_percent(interval=None)
-            return (
-                f'cpu=process_percent={process_cpu_percent:.2f}, '
-                f'system_percent={system_cpu_percent:.2f}'
-            )
-        except Exception as exc:  # pragma: no cover - diagnostic fallback
-            return f'cpu=unavailable({exc})'
-
-    @staticmethod
-    def _get_gc_snapshot():
-        try:
-            gc_counts = gc.get_count()
-            gc_thresholds = gc.get_threshold()
-            gc_stats = gc.get_stats()
-            generation_summaries = []
-            for generation, stats in enumerate(gc_stats):
-                generation_summaries.append(
-                    'gen{generation}[collections={collections}, collected={collected}, uncollectable={uncollectable}]'.format(
-                        generation=generation,
-                        collections=stats.get('collections', 0),
-                        collected=stats.get('collected', 0),
-                        uncollectable=stats.get('uncollectable', 0),
-                    )
-                )
-            return (
-                f'gc=enabled={gc.isenabled()}, counts={gc_counts}, thresholds={gc_thresholds}, '
-                f'stats={"; ".join(generation_summaries)}'
-            )
-        except Exception as exc:  # pragma: no cover - diagnostic fallback
-            return f'gc=unavailable({exc})'
-
-    @staticmethod
-    def _get_positive_int_from_env(name, default):
-        value = os.environ.get(name)
-        if value is None:
-            return default
-        try:
-            parsed = int(value)
-            if parsed > 0:
-                return parsed
-        except (TypeError, ValueError):
-            pass
-        logger.warning(f'Invalid value for {name}: {value}. Using default {default}.')
-        return default
-
     @staticmethod
     def _get_callback_execution_mode():
         value = os.environ.get('TOPIC_CALLBACK_EXECUTION_MODE', 'process')
@@ -405,19 +303,6 @@ def _get_process_context(self):
             self.callback_execution_mode = 'thread'
             return None
 
-class _LockRenewLoggingReceiver:
-    def __init__(self, topic):
-        self._topic = topic
-
-    def renew_message_lock(self, renewable):
-        logger.info(
-            'Attempting lock renewal for message %s; locked_until_utc=%s; runtime_snapshot=%s',
-            self._topic._get_message_id(renewable),
-            getattr(renewable, 'locked_until_utc', None),
-            self._topic._get_runtime_snapshot(),
-        )
-        return self._topic.receiver.renew_message_lock(renewable)
-
 
 def _run_callback_in_subprocess(message_payload, callbackfn, result_connection):
     try:

tests/unit_tests/test_topic/test_azure_topic.py

@@ -163,7 +163,7 @@ def test_subscribe_settles_completed_tasks_on_receiver_loop(
 
         topic._submit_processing_task.assert_called_once_with(mock_message, callback)
         mock_auto_lock_renewer.return_value.register.assert_called_once_with(
-            topic.lock_renew_receiver,
+            mock_receiver,
             mock_message,
             max_lock_renewal_duration=topic.max_renewal_duration,
             on_lock_renew_failure=topic._handle_lock_renew_failure,
@@ -369,74 +369,13 @@ def test_handle_lock_renew_failure_logs_when_sdk_returns_no_error(
         mock_message.auto_renew_error = None
 
         topic = AzureTopic(config=mock_config, topic_name='mock-topic', max_concurrent_messages=1)
-        topic._get_runtime_snapshot = MagicMock(
-            return_value=(
-                'memory=rss_mb=128.00, vms_mb=256.00, num_threads=4, '
-                'cpu=process_percent=80.00, system_percent=91.00, '
-                'gc=enabled=True, counts=(1, 2, 3), thresholds=(700, 10, 10), '
-                'stats=gen0[collections=1, collected=2, uncollectable=0]'
-            )
-        )
 
         topic._handle_lock_renew_failure(mock_message, None)
 
         mock_logger.error.assert_called_once_with(
             'Error renewing lock for message message-1: lock expired before renewal could complete; '
-            'locked_until_utc=2026-03-17T09:39:28Z; '
-            'runtime_snapshot=memory=rss_mb=128.00, vms_mb=256.00, num_threads=4, '
-            'cpu=process_percent=80.00, system_percent=91.00, '
-            'gc=enabled=True, counts=(1, 2, 3), thresholds=(700, 10, 10), '
-            'stats=gen0[collections=1, collected=2, uncollectable=0]'
-        )
-
-    @patch.dict(os.environ, {}, clear=True)
-    @patch('src.python_ms_core.core.topic.azure_topic.logger')
-    @patch('src.python_ms_core.core.topic.azure_topic.mp.get_context')
-    @patch('src.python_ms_core.core.topic.azure_topic.mp.get_all_start_methods', return_value=['fork', 'spawn'])
-    @patch('src.python_ms_core.core.topic.azure_topic.AutoLockRenewer')
-    @patch('src.python_ms_core.core.topic.azure_topic.ServiceBusClient')
-    def test_lock_renew_attempt_logs_memory_snapshot(
-        self,
-        mock_service_bus_client,
-        mock_auto_lock_renewer,
-        mock_get_all_start_methods,
-        mock_get_context,
-        mock_logger,
-    ):
-        mock_client = MagicMock()
-        mock_config = MagicMock(connection_string='Endpoint=sb://test/')
-        mock_service_bus_client.from_connection_string.return_value = mock_client
-        mock_client.get_topic_sender.return_value = MagicMock()
-        mock_auto_lock_renewer.return_value = MagicMock()
-        mock_get_context.return_value = MagicMock()
-
-        topic = AzureTopic(config=mock_config, topic_name='mock-topic', max_concurrent_messages=1)
-        topic._get_runtime_snapshot = MagicMock(
-            return_value=(
-                'memory=rss_mb=64.00, vms_mb=128.00, num_threads=3, '
-                'cpu=process_percent=55.00, system_percent=70.00, '
-                'gc=enabled=True, counts=(4, 5, 6), thresholds=(700, 10, 10), '
-                'stats=gen0[collections=3, collected=10, uncollectable=0]'
-            )
-        )
-
-        mock_message = MagicMock()
-        mock_message.message_id = 'message-2'
-        mock_message.locked_until_utc = '2026-03-17T09:39:33Z'
-        topic.receiver = MagicMock()
-
-        topic.lock_renew_receiver.renew_message_lock(mock_message)
-
-        mock_logger.info.assert_any_call(
-            'Attempting lock renewal for message %s; locked_until_utc=%s; runtime_snapshot=%s',
-            'message-2',
-            '2026-03-17T09:39:33Z',
-            'memory=rss_mb=64.00, vms_mb=128.00, num_threads=3, '
-            'cpu=process_percent=55.00, system_percent=70.00, '
-            'gc=enabled=True, counts=(4, 5, 6), thresholds=(700, 10, 10), '
-            'stats=gen0[collections=3, collected=10, uncollectable=0]',
+            'locked_until_utc=2026-03-17T09:39:28Z'
         )
-        topic.receiver.renew_message_lock.assert_called_once_with(mock_message)
 
 
 if __name__ == '__main__':