How to ensure reliable submission of a request from the database to the message queue?

To reliably submit a request from a database to a message queue, it is necessary to ensure the atomicity of the operation and guaranteed delivery processing. This is usually achieved through transactions and acknowledgments.

An approximate approach:

1. A request record is created in the database with a "not sent" flag.
2. Within a transaction:
   • The request record is created.
   • A message is sent to the queue (e.g., RabbitMQ, Kafka).
   • If the sending is successful, the request flag is updated to "sent".
3. If message sending fails, the transaction is rolled back so that the request does not remain in the database without a message.

If the queue supports acknowledgments (ack), you need to wait for delivery confirmation.

Example in Python using the pika library for RabbitMQ and SQLAlchemy:

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
import pika

Base = declarative_base()

class Request(Base):
    __tablename__ = 'requests'
    id = Column(Integer, primary_key=True)
    data = Column(String)
    sent = Column(Boolean, default=False)

engine = create_engine('sqlite:///requests.db')
Session = sessionmaker(bind=engine)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)

session = Session()
try:
    new_request = Request(data='some data')
    session.add(new_request)
    session.flush()  # get id

    message = new_request.data
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(delivery_mode=2)  # persistent message
    )

    new_request.sent = True
    session.commit()
except Exception as e:
    session.rollback()
    print('Error sending request:', e)
finally:
    session.close()
    connection.close()

Thus, either the request and message are created together, or not at all, ensuring reliability.