updated to handle exceptions in queue

Author: susrisha

src/example.py

@@ -40,8 +40,15 @@ def publish_messages(topic_name):
     print('Message Published')
 
 
+random_exec = 3
+exec_count = 0
 def long_running_task(sleep_time):
     # Simulate a long-running task
+    global exec_count
+    exec_count += 1 
+    if exec_count % random_exec == 0:
+        # throw an exception
+        raise Exception('Random Exception')
     time.sleep(sleep_time)
     sender_obj = MessageSender(topic_name=respond_topic)
     sender_obj.send_message({'message': 'Task Completed'})

src/python_ms_core/core/topic/azure_topic.py

@@ -48,7 +48,7 @@ def __init__(self, config: TopicConfig=None, topic_name=None, max_concurrent_mes
         self.publisher = self.client.get_topic_sender(topic_name=topic_name)
         self.executor = ThreadPoolExecutor(max_workers=max_concurrent_messages)
         self.internal_count = 0
-        self.lock_renewal = AutoLockRenewer(max_workers=4)
+        self.lock_renewal = AutoLockRenewer(max_workers=max_concurrent_messages)
         self.max_renewal_duration = 86400 # Renew the message upto 1 day
         self.wait_time_for_message = 5
 
@@ -99,24 +99,29 @@ def internal_callback(self, message, callbackfn):
             ServiceBusMessage: The processed message.
         """
         try:
-            self.internal_count += 1
+            self.internal_count += 1 # thread safe.
             queue_message = QueueMessage.data_from(str(message))
             callbackfn(queue_message)
+            return [True,message]
         except Exception as e:
             logger.error(f'Error in processing message: {e}')
-        finally: 
-            return message                 
-
+            return [False,message]
+                             
 
     def settle_message(self, x: cf.Future):
         """
         Sets the message as completed and updates the internal count.
         Args:
             x (cf.Future): The future object representing the message processing.
         """
-        self.internal_count -= 1
-        incoming_message = x.result()
-        self.receiver.complete_message(incoming_message)
+        self.internal_count -= 1 # thread safe.
+        # Check if the future has an exception
+        [is_success,incoming_message] = x.result()
+        if is_success:
+            self.receiver.complete_message(incoming_message)
+        else:
+            print(f'Abandoning message: {incoming_message}')
+            self.receiver.abandon_message(incoming_message) # send back to the topic
         return