Producers

Message Production

Producers create and send messages to message queues or topics.

Production Flow

Producer Flow:

1. Create message (payload + metadata)
2. Serialize message
3. Select destination (topic/queue)
4. Send message
5. Handle acknowledgment/error

Kafka Producer Example

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("retries", 3);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Send message
ProducerRecord<String, String> record = new ProducerRecord<>(
    "orders",          // topic
    "order-123",       // key
    orderJson          // value
);

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("Failed to send message", exception);
    } else {
        log.info("Message sent to partition {} offset {}",
            metadata.partition(), metadata.offset());
    }
});

RabbitMQ Producer Example

import pika
import json

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# Publish message
channel.basic_publish(
    exchange='orders',
    routing_key='order.created',
    body=json.dumps({
        'order_id': '123',
        'user_id': '456',
        'amount': 99.99
    }),
    properties=pika.BasicProperties(
        delivery_mode=2,  # Persistent
        content_type='application/json'
    )
)

Production Patterns

Pattern	Description	Use Case
Fire-and-forget	No acknowledgment	Logs, metrics
Sync publish	Wait for ack	Critical data
Async publish	Callback on ack	High throughput
Batch publish	Group messages	Bulk operations

Message Serialization

Serialization Formats

Comparison:

Format      | Size    | Speed   | Schema  | Human Readable
------------|---------|---------|---------|----------------
JSON        | Large   | Slow    | No      | Yes
Avro        | Small   | Fast    | Yes     | No
Protobuf    | Small   | Fast    | Yes     | No
MessagePack | Medium  | Fast    | No      | No
Thrift      | Small   | Fast    | Yes     | No

JSON Serialization

import json

# Serialize
data = {'order_id': 123, 'items': ['a', 'b']}
message = json.dumps(data)

# Deserialize
parsed = json.loads(message)

Avro Serialization

import avro.schema
from io import BytesIO
import avro.io

# Schema
schema = avro.schema.parse('''
{
    "type": "record",
    "name": "Order",
    "fields": [
        {"name": "order_id", "type": "int"},
        {"name": "amount", "type": "float"}
    ]
}
''')

# Serialize
writer = avro.io.DatumWriter(schema)
buffer = BytesIO()
encoder = avro.io.Encoder(buffer)
writer.write(data, encoder)
message = buffer.getvalue()

# Deserialize
reader = avro.io.DatumReader(schema)
buffer = BytesIO(message)
decoder = avro.io.Decoder(buffer)
parsed = reader.read(decoder)

Protobuf Serialization

// order.proto
syntax = "proto3";

message Order {
    int32 order_id = 1;
    float amount = 2;
    repeated string items = 3;
}

# Python usage
import order_pb2

# Serialize
order = order_pb2.Order()
order.order_id = 123
order.amount = 99.99
message = order.SerializeToString()

# Deserialize
parsed = order_pb2.Order()
parsed.ParseFromString(message)

When to Use Each

Format	When to Use
JSON	Prototyping, human debugging, simple schemas
Avro	Kafka, schema evolution, compact size
Protobuf	gRPC, strict schemas, performance
MessagePack	Alternative to JSON, better performance

Producer Error Handling

Error Types

1. Network Errors:
   - Connection timeout
   - DNS resolution failure
   - Network partition

2. Broker Errors:
   - Leader not available
   - Not enough replicas
   - Request timeout

3. Serialization Errors:
   - Schema mismatch
   - Invalid data format

4. Configuration Errors:
   - Invalid topic
   - Invalid partition

Retry Strategy

class ResilientProducer:
    def __init__(self, producer, max_retries=3, retry_delay=1.0):
        self.producer = producer
        self.max_retries = max_retries
        self.retry_delay = retry_delay
    
    def send(self, topic, key, value):
        for attempt in range(self.max_retries):
            try:
                future = self.producer.send(topic, key, value)
                record_metadata = future.get(timeout=10)
                return record_metadata
            except KafkaError as e:
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay * (2 ** attempt))  # Exponential backoff
                else:
                    raise e

Idempotent Producer

// Kafka idempotent producer
Properties props = new Properties();
props.put("enable.idempotence", "true");
props.put("acks", "all");
props.put("retries", Integer.MAX_VALUE);
props.put("max.in.flight.requests.per.connection", 5);

// Guarantees:
// - Exactly-once delivery per partition
// - Messages in order within partition

Dead Letter Queue

def send_with_dlq(producer, topic, dlq_topic, message):
    try:
        producer.send(topic, message)
    except Exception as e:
        # Send to DLQ for later processing
        producer.send(dlq_topic, {
            'original_topic': topic,
            'message': message,
            'error': str(e),
            'timestamp': time.time()
        })

Best Practices

1. Use idempotent producers to prevent duplicates
2. Implement retries with exponential backoff
3. Set appropriate timeouts for send operations
4. Use DLQ for failed messages
5. Monitor producer metrics (send rate, error rate, latency)