Add support for fair queue in SQS

docs/reference/kombu.transport.SQS.rst

@@ -70,4 +70,30 @@ Message Attributes
 
 SQS supports sending message attributes along with the message body.
 To use this feature, you can pass a 'message_attributes' as keyword argument
-to `basic_publish` method.
+to `basic_publish` method.
+
+Fair Queue Support
+------------------------
+
+Kombu supports Amazon SQS Fair Queues, which provide improved message processing fairness by ensuring that messages from different message groups 
+are processed in a balanced manner.
+
+Fair Queues are designed to prevent a single message group (or tenant) from monopolizing consumer resources, which can happen with standard queues 
+that handle multi-tenant workloads with unbalanced message distribution.
+
+When publishing messages to a Fair Queue, you must provide a `MessageGroupId`. This can be done by passing it as a 
+keyword argument to the `publish` method::
+
+    producer.publish(
+        message,
+        routing_key='my-fair-queue',
+        MessageGroupId='customer-123'  # Required for FIFO and Fair queues
+    )
+
+Benefits of using Fair Queues with Kombu:
+- Improved message processing fairness across message groups
+- Better workload distribution among consumers
+- Eliminates noisy neighbor problem
+
+For more information, refer to the AWS documentation on Fair Queues: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-fair-queues.html
+

kombu/transport/SQS.py

@@ -463,11 +463,13 @@ def _put(self, queue, message, **kwargs):
                 # we don't want to want to have the attribute in the body
                 kwargs['MessageAttributes'] = \
                     message['properties'].pop('message_attributes')
+            # support SQS fair queue system.
+            if 'MessageGroupId' in message['properties']:
+                kwargs['MessageGroupId'] = \
+                    message['properties']['MessageGroupId']
+            # Support FIFO queues.
             if queue.endswith('.fifo'):
-                if 'MessageGroupId' in message['properties']:
-                    kwargs['MessageGroupId'] = \
-                        message['properties']['MessageGroupId']
-                else:
+                if 'MessageGroupId' not in kwargs:
                     kwargs['MessageGroupId'] = 'default'
                 if 'MessageDeduplicationId' in message['properties']:
                     kwargs['MessageDeduplicationId'] = \

t/unit/transport/test_SQS.py

@@ -1149,6 +1149,29 @@ def test_predefined_queues_put_to_fifo_queue(self):
         assert 'MessageDeduplicationId' in \
             sqs_queue_mock.send_message.call_args[1]
 
+    def test_predefined_queues_put_with_message_group_id(self):
+        connection = Connection(transport=SQS.Transport, transport_options={
+            'predefined_queues': example_predefined_queues,
+        })
+        channel = connection.channel()
+
+        queue_name = 'queue-1'
+
+        exchange = Exchange('test_SQS', type='direct')
+        p = messaging.Producer(channel, exchange, routing_key=queue_name)
+
+        queue = Queue(queue_name, exchange, queue_name)
+        queue(channel).declare()
+
+        channel.sqs = Mock()
+        sqs_queue_mock = Mock()
+        channel.sqs.return_value = sqs_queue_mock
+        p.publish('message', MessageGroupId='test-group-id')
+
+        sqs_queue_mock.send_message.assert_called_once()
+        assert 'MessageGroupId' in sqs_queue_mock.send_message.call_args[1]
+        assert sqs_queue_mock.send_message.call_args[1]['MessageGroupId'] == 'test-group-id'
+
     def test_predefined_queues_put_to_queue(self):
         connection = Connection(transport=SQS.Transport, transport_options={
             'predefined_queues': example_predefined_queues,