const { Kafka, Partitioners } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['localhost:9092']
});

const producer = kafka.producer({
  // Modern safe partitioning rule
  createPartitioner: Partitioners.DefaultPartitioner
});

async function publishOrder(orderId, orderData) {
  await producer.connect();
  try {
    await producer.send({
      topic: 'orders-v1',
      acks: -1, // Guarantee absolute durability (acks=all)
      messages: [
        {
          key: orderId, // Keys guarantee sequential ordering per order
          value: JSON.stringify(orderData)
        }
      ]
    });
    console.log(`Order ${orderId} successfully published to Kafka.`);
  } catch (error) {
    console.error('Kafka Producer Error:', error);
    // Trigger retry logic or write to internal recovery log
  } finally {
    await producer.disconnect();
  }
}

from confluent_kafka import Consumer, KafkaError, KafkaException
import json
import sys

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-fulfillment-group',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,  # Manual commits for safety!
}

consumer = Consumer(conf)

def process_messages():
    try:
        consumer.subscribe(['orders-v1'])
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    sys.stderr.write(f'%% {msg.topic()} [{msg.partition()}] reached end at offset {msg.offset()}\n')
                else:
                    raise KafkaException(msg.error())
            else:
                # Valid message received
                order_id = msg.key().decode('utf-8')
                order_data = json.loads(msg.value().decode('utf-8'))
                
                print(f"Processing order {order_id} at partition {msg.partition()} offset {msg.offset()}")
                
                # Fulfil order database mutations here...
                
                # Commit offset synchronously to ensure At-Least-Once Delivery
                consumer.commit(asynchronous=False)
                
    except KeyboardInterrupt:
        print("Stopping consumer...")
    finally:
        consumer.close()

if __name__ == "__main__":
    process_messages()

const amqp = require('amqplib');

async function sendTask(taskId, taskPayload) {
  let connection;
  try {
    // Establish physical TCP connection
    connection = await amqp.connect('amqp://localhost');
    
    // Create virtual multiplexed channel
    const channel = await connection.createChannel();
    
    const exchangeName = 'tasks-exchange';
    const routingKey = 'tasks.priority.high';
    
    // Assert exchange exists (durable & safe)
    await channel.assertExchange(exchangeName, 'topic', { durable: true });
    
    const message = JSON.stringify(taskPayload);
    
    // Publish message with persistence
    channel.publish(exchangeName, routingKey, Buffer.from(message), {
      deliveryMode: 2, // Persistent message (writes payload to physical disk)
      messageId: taskId
    });
    
    console.log(`Task ${taskId} published to exchange.`);
    
  } catch (error) {
    console.error('RabbitMQ Publisher Error:', error);
  } finally {
    if (connection) {
      setTimeout(() => connection.close(), 500);
    }
  }
}

import pika
import sys
import json
import time

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()

    # Assert exchange and queue are durable
    channel.exchange_declare(exchange='tasks-exchange', exchange_type='topic', durable=True)
    channel.queue_declare(queue='high-priority-jobs', durable=True)
    
    # Bind queue to exchange with wildcard routing key
    channel.queue_bind(exchange='tasks-exchange', queue='high-priority-jobs', routing_key='tasks.priority.*')

    # VERY IMPORTANT: Define Prefetch limit (Backpressure)
    # Never push more than 1 unacknowledged message to this consumer
    channel.basic_qos(prefetch_count=1)

    def callback(ch, method, properties, body):
        try:
            task = json.loads(body.decode('utf-8'))
            print(f" [x] Processing task {properties.message_id} on thread")
            
            # Simulate work
            time.sleep(2)
            
            print(f" [x] Task {properties.message_id} complete")
            
            # Acknowledge completion manually
            ch.basic_ack(delivery_tag=method.delivery_tag)
            
        except Exception as e:
            print(f"Error processing task: {e}")
            # Requeue=True: Puts it back in queue. Requeue=False: DLQ routing
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

    # Register consumer with manual ACKs (no auto_ack=True!)
    channel.basic_consume(queue='high-priority-jobs', on_message_callback=callback, auto_ack=False)

    print(' [*] Waiting for tasks. Press CTRL+C to exit')
    channel.start_consuming()

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Exiting consumer.')
        sys.exit(0)