Pass socket_keepalive_options to redis client for result_backend

celery/backends/redis.py

@@ -228,6 +228,7 @@ def __init__(self, host=None, port=None, db=None, password=None,
         socket_connect_timeout = _get('redis_socket_connect_timeout')
         retry_on_timeout = _get('redis_retry_on_timeout')
         socket_keepalive = _get('redis_socket_keepalive')
+        socket_keepalive_options = self._transport_options['socket_keepalive_options']
         health_check_interval = _get('redis_backend_health_check_interval')
 
         self.connparams = {
@@ -259,6 +260,8 @@ def __init__(self, host=None, port=None, db=None, password=None,
         # absent in redis.connection.UnixDomainSocketConnection
         if socket_keepalive:
             self.connparams['socket_keepalive'] = socket_keepalive
+            if socket_keepalive_options:
+                self.connparams['socket_keepalive_options'] = socket_keepalive_options
 
         # "redis_backend_use_ssl" must be a dict with the keys:
         # 'ssl_cert_reqs', 'ssl_ca_certs', 'ssl_certfile', 'ssl_keyfile'

t/unit/backends/test_redis.py

@@ -5,6 +5,7 @@
 from contextlib import contextmanager
 from datetime import timedelta
 from pickle import dumps, loads
+from socket import TCP_KEEPCNT, TCP_KEEPIDLE, TCP_KEEPINTVL
 from unittest.mock import ANY, Mock, call, patch
 
 import pytest
@@ -17,7 +18,6 @@
 from celery.utils.collections import AttributeDict
 from t.unit import conftest
 
-
 def raise_on_second_call(mock, exc, *retval):
     def on_first_call(*args, **kwargs):
         mock.side_effect = exc
@@ -452,8 +452,28 @@ def test_socket_url(self):
         assert x.connparams['socket_timeout'] == 30.0
         assert 'socket_connect_timeout' not in x.connparams
         assert 'socket_keepalive' not in x.connparams
+        assert 'socket_keepalive_options' not in x.connparams
         assert x.connparams['db'] == 3
 
+    def test_socket_keepalive_options(self):
+        pytest.importorskip('redis') 
+
+        self.app.conf.redis_socket_keepalive = True
+        self.app.conf.result_backend_transport_options = {
+            'socket_keepalive_options': {
+                TCP_KEEPIDLE: 300,
+                TCP_KEEPCNT: 9,
+                TCP_KEEPINTVL: 45
+            }
+        }
+
+        x = self.Backend(
+            'socket:///tmp/redis.sock?virtual_host=/3', app=self.app,
+        )
+
+        assert x.connparams['socket_keepalive'] == True
+        assert x.connparams['socket_keepalive_options'] == {4: 300, 6: 9, 5: 45}
+
     def test_backend_ssl(self):
         pytest.importorskip('redis')