SQS - Failure to restore unacked messages

[YahelOpp]

Versions:

• Python 3.10
• Kombu 5.3.4
• Celery 5.3.6
• Boto3 1.33.7

After seeing messages stuck in NotVisible state at our production queues, I've set out to verify the warm shutdown mechanism works properly. My test setup:

• Simple task: sleeps for a desired amount of seconds, then retries with a countdown of 10.
• Max retries: 3
• Single worker, threads pool, concurrency 5

What I see happens:

• A task is run
• It raises the retry -> putting a new message in the queue
• The message is consumed by the same worker directly (message received)
• During the countdown period, the message is in NotVisible state
• During the countdown period, I shutdown the worker with SIGTERM
• The worker successfully starts a Warm shutdown
• The worker notes that it is restoring 1 unacked message
• The message stays in NotVisible state

Expected behavior: the message should move to Visible state.

A possible explanation I've found:
By manually changing the code in kombu/transport/SQS.py at the _put function in the case message.get('redelivered'), adding a duplicate call to change_message_visibility - the problem seemed to go away. I've tested this against real live SQS and against Localstack. In both the problem was consistent and solved once I added the duplicate call.

To sum up, the problem might be on boto3's side and not in kombu, but I'm unsure about this.
I will be happy to assist and make a PR / provide a simple testing scenario.

Thanks

[YahelOpp]

The patch I've done that makes it work as expected is:

# Patch a fix for SQS channel
original_put = Transport.Channel._put

def patched_put(self, queue, message, **kwargs):
    original_put(self, queue, message, **kwargs)
    if message.get("redelivered"):
        q_url = self._new_queue(queue)
        c = self.sqs(queue=self.canonical_queue_name(queue))
        c.change_message_visibility(
            QueueUrl=q_url,
            ReceiptHandle=message['properties']['delivery_tag'],
            VisibilityTimeout=0
        )


Transport.Channel._put = patched_put