9.1 Practical Producer Patterns & Real-World Scenarios

Module Overview

Goal: Bridge Kafka theory to production systems. Apply everything learned to real-world patterns.

What You'll Learn:

High-throughput event producers
Transactional outbox pattern
Producing from HTTP services
Backpressure handling
Graceful shutdown strategies
Safe producer configuration templates
Common anti-patterns and solutions
Production readiness checklist

Prerequisites:

All previous modules (this is the capstone)

1. Pattern 1: High-Throughput Event Producer

Use Case

Scenario: Process 500,000 events/sec from multiple sources, publish to Kafka.

Requirements:

Maximize throughput
Latency acceptable (< 100ms)
No data loss

Implementation

public class HighThroughputProducer {
    private final KafkaProducer<String, byte[]> producer;
    
    public HighThroughputProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
        props.put("client.id", "event-producer-" + InetAddress.getLocalHost().getHostName());
        
        // Serialization
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
        
        // Throughput optimizations
        props.put("batch.size", 65536);              // 64 KB batches
        props.put("linger.ms", 20);                  // Wait to batch
        props.put("compression.type", "lz4");        // Fast compression
        props.put("buffer.memory", 134217728);       // 128 MB buffer
        
        // Reliability
        props.put("acks", "all");
        props.put("enable.idempotence", "true");
        props.put("max.in.flight.requests.per.connection", 5);
        props.put("delivery.timeout.ms", 120000);
        
        this.producer = new KafkaProducer<>(props);
        
        // Register shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(this::close));
    }
    
    public void sendEvent(String key, byte[] value) {
        ProducerRecord<String, byte[]> record = 
            new ProducerRecord<>("events", key, value);
        
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                logger.error("Send failed: key={}", key, exception);
                metrics.increment("producer.failure");
                // Send to DLQ
                sendToDLQ(key, value, exception);
            } else {
                metrics.increment("producer.success");
            }
        });
    }
    
    public void close() {
        logger.info("Closing producer...");
        producer.close(Duration.ofSeconds(30));
        logger.info("Producer closed");
    }
}

Expected Performance

Configuration: High throughput
Throughput: 500,000 msgs/sec
Average message: 1 KB
Batch size: ~64 KB (64 messages)
Requests/sec: ~8,000
Network usage: ~150 MB/sec (compressed from 500 MB/sec)
Latency p99: ~35ms

2. Pattern 2: Transactional Outbox

Use Case

Scenario: Ensure database updates and Kafka writes happen atomically.

Problem: How to update database AND send Kafka message reliably?

❌ PROBLEM:
1. Update database → Success
2. Send to Kafka → Failure
Result: Database updated, but event not published (inconsistent)

Solution: Outbox Pattern

DATABASE TRANSACTION:
1. Update business table
2. Insert into outbox table

SEPARATE PROCESS (Outbox Publisher):
1. Read from outbox table
2. Send to Kafka
3. Delete from outbox (or mark sent)

Implementation

Step 1: Outbox Table

CREATE TABLE outbox (
    id BIGSERIAL PRIMARY KEY,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_outbox_created ON outbox(created_at);

Step 2: Business Logic (Transactional)

@Transactional
public void placeOrder(Order order) {
    // 1. Update business table
    orderRepository.save(order);
    
    // 2. Insert into outbox (same transaction)
    OutboxEvent event = new OutboxEvent(
        "Order",
        order.getId(),
        "OrderPlaced",
        serializeOrder(order)
    );
    outboxRepository.save(event);
    
    // Both succeed or both fail (atomic)
}

Step 3: Outbox Publisher

@Component
public class OutboxPublisher {
    private final KafkaProducer<String, String> producer;
    private final OutboxRepository outboxRepository;
    
    @Scheduled(fixedDelay = 100) // Every 100ms
    public void publishEvents() {
        List<OutboxEvent> events = outboxRepository.findUnpublished(100);
        
        if (events.isEmpty()) {
            return;
        }
        
        try {
            producer.beginTransaction();
            
            for (OutboxEvent event : events) {
                ProducerRecord<String, String> record = new ProducerRecord<>(
                    event.getAggregateType().toLowerCase() + "-events",
                    event.getAggregateId(),
                    event.getPayload()
                );
                
                producer.send(record);
            }
            
            // Mark as published
            markAsPublished(events);
            
            producer.commitTransaction();
            
        } catch (Exception e) {
            producer.abortTransaction();
            logger.error("Failed to publish outbox events", e);
        }
    }
    
    @Transactional
    private void markAsPublished(List<OutboxEvent> events) {
        for (OutboxEvent event : events) {
            outboxRepository.delete(event);
            // Or: outboxRepository.markPublished(event);
        }
    }
}

Benefits

✅ Atomic: Database + Kafka updates are consistent
✅ Reliable: No message loss
✅ Retry-safe: Can retry outbox publishing
✅ Auditable: Outbox table provides audit trail

3. Pattern 3: Producing from HTTP Service

Use Case

Scenario: HTTP API receives requests, publishes events to Kafka.

Challenge: Handle backpressure, ensure reliability, maintain low latency.

Implementation

@RestController
@RequestMapping("/api/events")
public class EventController {
    private final KafkaProducer<String, String> producer;
    private final Semaphore semaphore;
    
    public EventController(KafkaProducer<String, String> producer) {
        this.producer = producer;
        // Limit concurrent sends (backpressure)
        this.semaphore = new Semaphore(1000);
    }
    
    @PostMapping
    public ResponseEntity<EventResponse> publishEvent(@RequestBody EventRequest request) {
        // Check if producer can accept more
        if (!semaphore.tryAcquire()) {
            return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
                .body(new EventResponse("Producer overloaded, retry later"));
        }
        
        try {
            // Validate
            if (request.getPayload() == null) {
                return ResponseEntity.badRequest()
                    .body(new EventResponse("Payload required"));
            }
            
            ProducerRecord<String, String> record = new ProducerRecord<>(
                "api-events",
                request.getKey(),
                request.getPayload()
            );
            
            // Send async
            CompletableFuture<RecordMetadata> future = new CompletableFuture<>();
            
            producer.send(record, (metadata, exception) -> {
                semaphore.release();
                
                if (exception != null) {
                    future.completeExceptionally(exception);
                } else {
                    future.complete(metadata);
                }
            });
            
            // Wait for result (with timeout)
            try {
                RecordMetadata metadata = future.get(5, TimeUnit.SECONDS);
                
                return ResponseEntity.ok(new EventResponse(
                    "Event published",
                    metadata.topic(),
                    metadata.partition(),
                    metadata.offset()
                ));
                
            } catch (TimeoutException e) {
                return ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT)
                    .body(new EventResponse("Timeout waiting for Kafka"));
            }
            
        } catch (Exception e) {
            semaphore.release();
            logger.error("Failed to publish event", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(new EventResponse("Failed to publish event"));
        }
    }
}

Backpressure Handling

// Option 1: Reject requests when overloaded
if (!semaphore.tryAcquire()) {
    return ResponseEntity.status(429).body("Rate limited");
}

// Option 2: Queue requests (bounded queue)
BlockingQueue<ProducerRecord> queue = new ArrayBlockingQueue<>(10000);
if (!queue.offer(record)) {
    return ResponseEntity.status(429).body("Queue full");
}

// Option 3: Apply backpressure to client (return 503)
if (producer.metrics().get("buffer-available-bytes").value() < threshold) {
    return ResponseEntity.status(503).body("Service unavailable");
}

4. Pattern 4: Graceful Shutdown

Use Case

Scenario: Shutdown application without losing in-flight messages.

Challenge: Ensure all buffered messages are sent before exiting.

Implementation

public class GracefulProducer {
    private final KafkaProducer<String, String> producer;
    private final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
    private final AtomicInteger pendingMessages = new AtomicInteger(0);
    
    public GracefulProducer() {
        Properties props = createProps();
        this.producer = new KafkaProducer<>(props);
        
        // Register shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
    }
    
    public void send(ProducerRecord<String, String> record) {
        if (shutdownRequested.get()) {
            throw new IllegalStateException("Producer is shutting down");
        }
        
        pendingMessages.incrementAndGet();
        
        producer.send(record, (metadata, exception) -> {
            pendingMessages.decrementAndGet();
            
            if (exception != null) {
                logger.error("Send failed", exception);
            }
        });
    }
    
    public void shutdown() {
        logger.info("Shutdown requested...");
        shutdownRequested.set(true);
        
        // Stop accepting new messages
        logger.info("Stopped accepting new messages");
        
        // Wait for pending messages (with timeout)
        long startWait = System.currentTimeMillis();
        long timeout = 30000; // 30 seconds
        
        while (pendingMessages.get() > 0) {
            if (System.currentTimeMillis() - startWait > timeout) {
                logger.warn("Timeout waiting for pending messages: {}", 
                    pendingMessages.get());
                break;
            }
            
            logger.info("Waiting for {} pending messages...", pendingMessages.get());
            
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
        
        // Flush remaining buffered messages
        logger.info("Flushing producer...");
        producer.flush();
        
        // Close producer
        logger.info("Closing producer...");
        producer.close(Duration.ofSeconds(10));
        
        logger.info("Producer shutdown complete");
    }
}

Kubernetes/Docker Integration

// Handle SIGTERM from Kubernetes
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    logger.info("SIGTERM received, starting graceful shutdown...");
    gracefulProducer.shutdown();
}));

// Kubernetes terminationGracePeriodSeconds
// Set to > producer shutdown timeout (e.g., 60 seconds)

5. Pattern 5: Multi-Topic Producer

Use Case

Scenario: Produce to multiple topics with different configurations.

Challenge: Balance shared vs dedicated producers.

Option 1: Single Producer (Simple)

public class MultiTopicProducer {
    private final KafkaProducer<String, String> producer;
    
    public void sendUserEvent(UserEvent event) {
        producer.send(new ProducerRecord<>("user-events", event.getUserId(), serialize(event)));
    }
    
    public void sendOrderEvent(OrderEvent event) {
        producer.send(new ProducerRecord<>("order-events", event.getOrderId(), serialize(event)));
    }
    
    public void sendPaymentEvent(PaymentEvent event) {
        producer.send(new ProducerRecord<>("payment-events", event.getPaymentId(), serialize(event)));
    }
}

Pros: Simple, single connection, shared buffer
Cons: All topics use same config (can't optimize per-topic)

Option 2: Multiple Producers (Optimized)

public class OptimizedMultiTopicProducer {
    private final KafkaProducer<String, String> criticalProducer;
    private final KafkaProducer<String, String> bulkProducer;
    
    public OptimizedMultiTopicProducer() {
        // Critical producer: Low latency
        Properties criticalProps = new Properties();
        criticalProps.put("linger.ms", 0);
        criticalProps.put("batch.size", 8192);
        criticalProps.put("compression.type", "none");
        this.criticalProducer = new KafkaProducer<>(criticalProps);
        
        // Bulk producer: High throughput
        Properties bulkProps = new Properties();
        bulkProps.put("linger.ms", 20);
        bulkProps.put("batch.size", 65536);
        bulkProps.put("compression.type", "lz4");
        this.bulkProducer = new KafkaProducer<>(bulkProps);
    }
    
    public void sendPaymentEvent(PaymentEvent event) {
        // Critical: Use low-latency producer
        criticalProducer.send(new ProducerRecord<>("payment-events", serialize(event)));
    }
    
    public void sendAnalyticsEvent(AnalyticsEvent event) {
        // Non-critical: Use high-throughput producer
        bulkProducer.send(new ProducerRecord<>("analytics-events", serialize(event)));
    }
}

Pros: Optimized per use case
Cons: More complexity, more connections

6. Safe Producer Configuration Templates

Template 1: Production Default (Recommended)

Properties props = new Properties();

// Connection
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("client.id", "my-app-" + hostname);

// Serialization
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Reliability (no data loss, no duplicates)
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("retries", Integer.MAX_VALUE);
props.put("delivery.timeout.ms", 120000);
props.put("max.in.flight.requests.per.connection", 5);

// Performance (balanced)
props.put("batch.size", 16384);
props.put("linger.ms", 10);
props.put("compression.type", "lz4");
props.put("buffer.memory", 33554432);

// Timeouts
props.put("request.timeout.ms", 30000);
props.put("max.block.ms", 60000);

✅ Use for: Most production applications

Template 2: Critical/Financial Systems

Properties props = new Properties();

// ... base config ...

// Maximum reliability
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("min.insync.replicas", 2); // Broker/topic config
props.put("retries", Integer.MAX_VALUE);
props.put("delivery.timeout.ms", 300000); // 5 minutes

// Conservative timeouts
props.put("request.timeout.ms", 60000);
props.put("max.block.ms", 120000);

// Moderate performance
props.put("linger.ms", 0); // Send immediately
props.put("compression.type", "lz4");

✅ Use for: Financial transactions, critical business data

Template 3: High-Volume Analytics

Properties props = new Properties();

// ... base config ...

// Balanced reliability (ok with rare loss)
props.put("acks", "1");
props.put("retries", 3);

// Maximum throughput
props.put("batch.size", 131072); // 128 KB
props.put("linger.ms", 50);
props.put("compression.type", "lz4");
props.put("buffer.memory", 134217728); // 128 MB

✅ Use for: Analytics, logs, metrics

7. Common Anti-Patterns

❌ Anti-Pattern 1: Creating Producer Per Message

// ❌ BAD: New producer for every message
public void sendMessage(String message) {
    KafkaProducer producer = new KafkaProducer<>(props);
    producer.send(new ProducerRecord<>("topic", message));
    producer.close();
}

Problem: Extremely slow, wastes resources.

Fix:

// ✅ GOOD: Reuse producer
private static final KafkaProducer producer = new KafkaProducer<>(props);

public void sendMessage(String message) {
    producer.send(new ProducerRecord<>("topic", message));
}

❌ Anti-Pattern 2: Ignoring Send Failures

// ❌ BAD: Fire and forget without monitoring
producer.send(record);

Problem: Silent failures, data loss.

Fix:

// ✅ GOOD: Handle failures
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        logger.error("Send failed", exception);
        metrics.increment("producer.failure");
        sendToDLQ(record);
    }
});

❌ Anti-Pattern 3: Blocking Main Thread with Sync Send

// ❌ BAD: Blocks thread for every message
for (Record record : records) {
    producer.send(record).get(); // Blocks!
}

Problem: Serializes sends, extremely slow.

Fix:

// ✅ GOOD: Async sends
for (Record record : records) {
    producer.send(record, callback);
}
producer.flush(); // Wait once at end

❌ Anti-Pattern 4: Not Closing Producer

// ❌ BAD: No cleanup
KafkaProducer producer = new KafkaProducer<>(props);
producer.send(record);
// Process exits without closing

Problem: Buffered messages lost.

Fix:

// ✅ GOOD: Always close
try (KafkaProducer producer = new KafkaProducer<>(props)) {
    producer.send(record);
} // Auto-closes

8. Production Readiness Checklist

Configuration Checklist

bootstrap.servers — Multiple brokers for redundancy
client.id — Unique identifier per instance
acks=all — No data loss
enable.idempotence=true — No duplicates
retries=Integer.MAX_VALUE — Automatic retries
delivery.timeout.ms — Appropriate timeout
compression.type — Enable for text data
batch.size, linger.ms — Tuned for workload

Code Checklist

Single producer instance (shared across threads)
Callbacks for error handling
Metrics and monitoring
Dead letter queue for failures
Graceful shutdown logic
Shutdown hook registered
Serialization error handling
Backpressure mechanism (if accepting input from external source)

Monitoring Checklist

Send success/failure rate
Latency (p50, p99)
Throughput (msgs/sec, MB/sec)
Batch size average
Buffer available bytes
Request rate
Error counts by type
DLQ size

8.5 Key Producer Metrics to Monitor

Why Monitoring Matters

Without metrics, you're flying blind:

Can't detect performance degradation
Can't diagnose issues
Can't optimize configurations
Can't prove SLAs

Kafka Producer exposes JMX metrics — use them!

Critical Metrics (Must Monitor)

1. record-send-rate

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=record-send-rate
Unit: records/second

What it tells you: How many records/second you're sending.

Why it matters:

Detect throughput drops
Validate capacity planning
Identify bottlenecks

Alert on:

Drop below expected rate
Sudden spikes (may indicate retry storms)

2. record-error-rate

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=record-error-rate
Unit: errors/second

What it tells you: How many records/second are failing.

Why it matters:

Immediate indicator of problems
May indicate broker issues, network problems, or serialization errors

Alert on:

Any non-zero value (if you expect 100% success)
Sustained errors

3. request-latency-avg / request-latency-max

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=request-latency-avg
Unit: milliseconds

What it tells you: How long requests to brokers take.

Why it matters:

High latency = slow producers
May indicate network issues or broker overload

Alert on:

p99 latency > 100ms (or your SLA)
Sudden latency spikes

4. buffer-available-bytes

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=buffer-available-bytes
Unit: bytes

What it tells you: How much buffer memory is available.

Why it matters:

Low available bytes → Buffer filling up
May lead to blocking or BufferExhaustedException

Alert on:

Available bytes < 10% of buffer.memory
Sustained low values

5. batch-size-avg

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=batch-size-avg
Unit: bytes

What it tells you: Average batch size being sent.

Why it matters:

Small batches = poor throughput
Large batches = good batching efficiency

Optimize for:

Batch size close to batch.size config (if high throughput)
Smaller batches OK if low latency is goal

6. compression-rate-avg

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=compression-rate-avg
Unit: ratio (e.g., 0.25 = 4:1 compression)

What it tells you: How effective compression is.

Why it matters:

Low compression rate = wasting network/storage
High compression rate = good savings

Optimize for:

Text data: 0.2-0.4 (2.5:1 to 5:1)
Binary data: 0.6-0.8 (1.25:1 to 1.67:1)

7. record-retry-rate

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=record-retry-rate
Unit: retries/second

What it tells you: How often records are being retried.

Why it matters:

High retry rate = transient errors
May indicate broker issues, network problems, or under-replicated partitions

Alert on:

Sustained high retry rate
Retry storms (sudden spike)

Important Metrics (Should Monitor)

8. records-per-request-avg

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=records-per-request-avg
Unit: records/request

What it tells you: How many records per batch.

Why it matters:

Low value = poor batching (throughput suffers)
High value = good batching

9. request-rate

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=request-rate
Unit: requests/second

What it tells you: How many requests/second to brokers.

Why it matters:

High request rate with low records/request = inefficient
Low request rate with high records/request = efficient

10. produce-throttle-time-avg

Metric: kafka.producer:type=producer-metrics,client-id=<client-id>,name=produce-throttle-time-avg
Unit: milliseconds

What it tells you: How long broker is throttling your producer.

Why it matters:

Non-zero = broker applying quotas
May indicate you're exceeding quota limits

Alert on:

Sustained throttling
Unexpected throttling (if no quotas configured)

How to Access Metrics

Option 1: JMX (Java Management Extensions)

// Enable JMX in your application
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
ObjectName name = new ObjectName("kafka.producer:type=producer-metrics,client-id=my-producer");
Double recordSendRate = (Double) mbs.getAttribute(name, "record-send-rate");

Option 2: Prometheus + JMX Exporter

# jmx_exporter config
rules:
  - pattern: kafka.producer<type=producer-metrics, client-id=(.+)><>(.+)
    name: kafka_producer_$2
    labels:
      client_id: $1

Option 3: Application Metrics Library

// Using Micrometer (Spring Boot)
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
new KafkaProducerMetrics(producer).bindTo(meterRegistry);

Recommended Alerting Thresholds

Metric	Threshold	Action
`record-error-rate`	> 0	Investigate immediately
`request-latency-avg`	> 100ms	Check network/broker
`buffer-available-bytes`	< 10% of total	Increase buffer or reduce send rate
`record-retry-rate`	> 10% of send rate	Check broker health
`produce-throttle-time-avg`	> 0	Check quotas

Example Monitoring Dashboard

Key Panels:

Throughput: record-send-rate (line chart)
Errors: record-error-rate (line chart)
Latency: request-latency-avg, request-latency-max (line chart)
Buffer Health: buffer-available-bytes (gauge)
Batching Efficiency: batch-size-avg, records-per-request-avg (line chart)
Compression: compression-rate-avg (gauge)
Retries: record-retry-rate (line chart)

Operational Checklist

Tested failover scenarios (broker down)
Tested network partition
Load tested at peak volume
Documented configuration choices
Runbooks for common issues
Alerting on key metrics
Log aggregation configured
Tested graceful shutdown

9. Debugging Production Issues

Issue 1: High Latency

Symptoms:

p99 latency > 100ms
Users complaining of slow responses

Diagnosis:

// Check metrics
kafka.producer:name=request-latency-avg → High?
kafka.producer:name=batch-size-avg → Is batching effective?
kafka.producer:name=records-per-request-avg → Low? (poor batching)

Fix:

Reduce linger.ms for lower latency
Check network to brokers
Check broker performance

Issue 2: Low Throughput

Symptoms:

Can't achieve target msgs/sec
CPU underutilized

Diagnosis:

kafka.producer:name=record-send-rate → Too low?
kafka.producer:name=batch-size-avg → Small batches?

Fix:

Increase linger.ms to batch more
Increase batch.size
Enable compression
Check network bandwidth

Issue 3: Producer Blocking

Symptoms:

send() calls taking seconds
Application appears hung

Diagnosis:

kafka.producer:name=buffer-available-bytes → Zero?
kafka.producer:name=buffer-exhausted-rate → High?

Fix:

Increase buffer.memory
Check broker availability
Check network connectivity
Reduce send rate

Summary

Key Takeaways

✅ Reuse producers — One per application (thread-safe)
✅ Handle failures — Always use callbacks
✅ Monitor metrics — Track success/failure rates
✅ Graceful shutdown — Flush and close properly
✅ Backpressure — Handle overload gracefully
✅ Use templates — Start with proven configurations
✅ Test failures — Network partition, broker down
✅ Production readiness — Use checklist before going live

Production-Ready Producer Pattern

@Component
public class ProductionProducer {
    private final KafkaProducer<String, String> producer;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    
    public ProductionProducer() {
        this.producer = new KafkaProducer<>(createProductionConfig());
        Runtime.getRuntime().addShutdownHook(new Thread(this::close));
    }
    
    public void send(String key, String value) {
        if (closed.get()) {
            throw new IllegalStateException("Producer closed");
        }
        
        ProducerRecord<String, String> record = 
            new ProducerRecord<>("my-topic", key, value);
        
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                handleFailure(record, exception);
            } else {
                handleSuccess(metadata);
            }
        });
    }
    
    private void handleFailure(ProducerRecord record, Exception exception) {
        logger.error("Send failed", exception);
        metrics.increment("producer.failure");
        sendToDLQ(record, exception);
    }
    
    private void handleSuccess(RecordMetadata metadata) {
        metrics.increment("producer.success");
        logger.debug("Sent: offset={}", metadata.offset());
    }
    
    public void close() {
        if (closed.compareAndSet(false, true)) {
            logger.info("Closing producer...");
            producer.close(Duration.ofSeconds(30));
            logger.info("Producer closed");
        }
    }
}

Congratulations!

You've completed the Kafka Producers Course. You now have the knowledge to:

✅ Build reliable producers
✅ Optimize for your use case
✅ Handle failures gracefully
✅ Deploy to production with confidence

Next Steps:

Apply to your projects
Experiment with different configurations
Monitor and iterate
Share knowledge with your team

Previous Module: ← 8.3 - Compression Strategies
Course Complete! 🎉