I'm using Azure service bus library for python to read messages from queues. After x period of time I get the following error:
Traceback (most recent call last):
File "/opt/anaconda3/lib/python3.7/site-packages/uamqp/authentication/cbs_auth.py", line 76, in create_authenticator
self._connection.container_id)
File "./src/cbs.pyx", line 73, in uamqp.c_uamqp.CBSTokenAuth.__cinit__
ValueError: Unable to open CBS link.
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/opt/anaconda3/lib/python3.7/site-packages/azure/servicebus/receive_handler.py", line 309, in open
self._handler.open(connection=self.connection)
File "/opt/anaconda3/lib/python3.7/site-packages/uamqp/client.py", line 259, in open
self._build_session()
File "/opt/anaconda3/lib/python3.7/site-packages/uamqp/client.py", line 214, in _build_session
on_attach=self._on_attach)
File "/opt/anaconda3/lib/python3.7/site-packages/uamqp/authentication/cbs_auth.py", line 82, in create_authenticator
"Please confirm target hostname exists: {}".format(connection.container_id, connection.hostname))
uamqp.errors.AMQPConnectionError: Unable to open authentication session on connection b'SBReceiver-00000000-0000-0000-0000-000000000000'.
Please confirm target hostname exists: b'myhostname.servicebus.windows.net'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/opt/anaconda3/lib/python3.7/runpy.py", line 193, in _run_module_as_main
"__main__", mod_spec)
File "/opt/anaconda3/lib/python3.7/runpy.py", line 85, in _run_code
exec(code, run_globals)
File "/path/to/main.py", line 648, in <module>
main()
File "/path/to/main.py", line 631, in main
run_service_bus()
File "/path/to/main.py", line 482, in run_service_bus
with my_client.get_receiver() as queue_receiver:
File "/opt/anaconda3/lib/python3.7/site-packages/azure/servicebus/base_handler.py", line 57, in __enter__
self.open()
File "/opt/anaconda3/lib/python3.7/site-packages/azure/servicebus/receive_handler.py", line 318, in open
self._handle_exception(e)
File "/opt/anaconda3/lib/python3.7/site-packages/azure/servicebus/base_handler.py", line 131, in _handle_exception
raise ServiceBusConnectionError(message, exception)
azure.servicebus.common.errors.ServiceBusConnectionError: Failed to open handler: Unable to open authentication session on connection b'SBReceiver-00000000-0000-0000-0000-000000000000'.
Please confirm target hostname exists: b'myhostname.servicebus.windows.net'
I think what's happening here is that after a period of time the token I have expires. What is the proper way to handle this?
The code I'm using is the following:
sb_client = ServiceBusClient.from_connection_string("primary_connection_string_here")
my_client = sb_client.get_queue("queue_name_here")
with my_client.get_receiver() as queue_receiver:
messages = queue_receiver.fetch_next(timeout=3)
for message in messages:
message.complete()
As discussed over comments, the issue in this case is most likely to be due to a network transient error which is quite normal in distributed environment. A transient error most of the times can be recovered by retry. Unfortunately, in the legacy python service bus SDK of v0.50.x, there was no out of the box retry feature. Exponential back off retry has been added in the latest V7 SDK (currently in preview, soon will be GA). You can refer the migration guide from v0.50 to v7 for details. Below is an example of the receiver code you have using V7 SDK (NOTE: a sync variation, there is also asyncio support which you can check in the extensive list of samples).
V7 SDK now allows you pass the retry parameters for the client. Though default values should be enough in general.
retry_total : Total number of retries to allow. Takes precedence over other counts. Default value is 10.
retry_backoff_factor : A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a second try without a delay). In fixed mode, retry policy will alwasy sleep for {backoff factor}. In 'exponential' mode, retry policy will sleep for: {backoff factor} * (2 ** ({number of total retries} - 1)) seconds. If the backoff_factor is 0.1, then the retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8.
retry_backoff_max : The maximum back off time. Default value is 120 (in seconds).
servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR, retry_total=10, retry_backoff_factor=1, retry_backoff_max=30)
with servicebus_client:
receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
with receiver:
received_msgs = receiver.receive_messages(max_message_count=10, max_wait_time=5)
for msg in received_msgs:
print(str(msg))
msg.complete()