openstack/oslo.messaging
Code Issues Proposed changes

[rabbit] Ignore stream offset header when stream queue is not used

Browse Source 
Consumer callback should look for x-stream-offset header only if
the stream mode is used. Otherwise the message headers attribute could
be None and the error appears when using get method.

Closes-Bug: #2084168
Change-Id: I7f9c742f8f557d9faae2cd749d34dcb15d8005c0
This commit is contained in:
Vladimir Kozhukalov 2024-10-09 18:51:35 -05:00 committed by Takashi Kajinami
parent b3d7270b3a
commit 930d97599c
1 changed files with 7 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
oslo_messaging/_drivers/impl_rabbit.py
View File

@ -631,12 +631,12 @@ class Consumer(object):
Messages that are processed and ack'ed. Messages that are processed and ack'ed.
""" """
if self.rabbit_stream_fanout:
offset = message.headers.get("x-stream-offset") offset = message.headers.get("x-stream-offset")
if offset is not None: if offset is not None:
LOG.debug("Stream for %s current offset: %s", self.queue_name, LOG.debug("Stream for %s current offset: %s",
offset) self.queue_name, offset)
self.next_stream_offset = offset + 1 self.next_stream_offset = offset + 1
m2p = getattr(self.queue.channel, 'message_to_python', None) m2p = getattr(self.queue.channel, 'message_to_python', None)
if m2p: if m2p:
@ -1099,7 +1099,7 @@ class Connection(object):
info = {'err_str': exc, 'sleep_time': interval} info = {'err_str': exc, 'sleep_time': interval}
info.update(self._get_connection_info(conn_error=True)) info.update(self._get_connection_info(conn_error=True))
if 'Basic.cancel' in str(exc): if self.rabbit_stream_fanout and 'Basic.cancel' in str(exc):
# This branch allows for consumer offset reset # This branch allows for consumer offset reset
# in the unlikely case consumers are cancelled. This may # in the unlikely case consumers are cancelled. This may
# happen, for example, when we delete the stream queue. # happen, for example, when we delete the stream queue.