Using Kombu for long running tasks

[amCap1712]

Hi! I am trying to use Kombu for long running tasks (the longest one can take as long as 1hr) but am facing intermittent issues. I see OSError: Server unexpectedly closed connection occuring ocassionlly in the client code. The server logs show that the connection was closed due to missed heartbeats. I am not sure why that happens becauase I am offloading the long running task to a background thread and keeping the main thread free for kombu to communicate heartbeats with rabbitmq. Any thoughts on how to fix the issue and make kombu play well with long running tasks.

Here is my consumer code,

import config
from queue import Queue as qq

from kombu import Queue, Exchange
from kombu.mixins import ConsumerProducerMixin
from threading import Thread

class RequestConsumer(ConsumerProducerMixin):

    def __init__(self):
        self.connection = Connection(
                hostname=config.RABBITMQ_HOST,
                userid=config.RABBITMQ_USERNAME,
                port=config.RABBITMQ_PORT,
                password=config.RABBITMQ_PASSWORD,
                virtual_host=config.RABBITMQ_VHOST,
                heartbeat=2*60*60,
                transport_options={"client_properties": {"connection_name": "spark-request-consumer"}}
            )

        self.request_exchange = Exchange(config.SPARK_REQUEST_EXCHANGE, type="fanout")
        self.request_queue = Queue(config.SPARK_REQUEST_QUEUE, self.request_exchange)
        self.result_exchange = Exchange(config.SPARK_RESULT_EXCHANGE, type="fanout")
        self.result_queue = Queue(config.SPARK_RESULT_QUEUE, self.result_exchange)
        self.task_queue = qq()
        Thread(target=self.calculate_results).start()

    def calculate_results(self):
        while True:
            try:
                body, message = self.task_queue.get()
                request = json.loads(body)
                results = calculate(request)  # long running task
                if results:
                    self.push_to_result_queue(results)
                message.ack()
                logger.info("Request done!")
            except Exception:
                logger.error("Error while calculating results:", exc_info=True)

    def push_to_result_queue(self, messages):
        logger.debug("Pushing result to RabbitMQ...")
        for message in messages:
            body = json.dumps(message)
            if message:
                self.producer.publish(
                    body,
                    exchange=self.result_exchange,
                    routing_key="",
                    delivery_mode=2,
                    content_type="application/json",
                    content_encoding="utf-8",
                )

    def callback(self, body, message):
        logger.info("Received a request!")
        self.task_queue.put((body, message))
        logger.info("Request put in internal queue!")

    def get_consumers(self, Consumer, channel):
        return [Consumer(queues=[self.request_queue], callbacks=[self.callback], prefetch_count=1)]