Использование kafka-python с флягой для отправки уведомления

На диаграмме выше показана текущая архитектура, которую я пытаюсь реализовать. В настоящее время я запускаю zookeeper и kakfa (новинка для обоих) на удаленном сервере, и это приложение службы уведомлений построено с использованием колбы, которая действует как потребитель (kafka-python), а другое приложение App1..Appn будет действовать как производитель. Что ж, я могу независимо управлять потребителем и производителем, которые работают нормально. Но я попытался интегрировать потребителя в службу уведомлений, ничего не произошло.

app.py

def send_notification_from_consumer(payload):
  # sending notification using firebase


if __name__ == '__main__':
  app.run(host='127.0.0.1', port=8000, debug=True)

consumer.py

from kafka import KafkaConsumer
import json
from app import send_notification_from_consumer as snc

def notification_consumer():
  print("---inside consumer-->")
  consumer = KafkaConsumer(
    'send-notification',
    bootstrap_servers=['remote_server_ip:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='my-group',
    value_deserializer=lambda m: json.loads(m.decode("utf-8")))

  for message in consumer:
    print("----->",message.value)
    snc(message.value)

producer.py

from kafka import KafkaProducer
import json

producer = KafkaProducer(bootstrap_servers=['remote_server_ip:9092'],
                         value_serializer=lambda m: json.dumps(m).encode('utf-8'))


def on_send_success(record_metadata):
    print(record_metadata.topic)
    print(record_metadata.partition)
    print(record_metadata.offset)


def on_send_error(excp):
    log.error('I am an errback', exc_info=excp)


for i in range(1):
    name = {"phone_number_list": ["1234567890", "1234567891"],
            "payload": {"title": "Hello from another ",
                        "description": "Test notification by notification service",
                        "banner_image": "", "id": 123, "action": {"type": "web", "button_label": "", "link": ""},
                        "app_name": "app_name"}
            }

    producer.send('send-notification', name).add_callback(on_send_success).add_errback(on_send_error)

Можно ли вызвать потребителя без использования HTTP-запроса, потому что я просто вызываю send_notification_from_consumer внутри потребителя kafka. Как правильно это делать? Как мне запустить потребителя в колбе.

Я также пробовал это решение, но я получил эту ошибку. Можно ли масштабировать это решение?

RuntimeError: Working outside of request context.