2.1 Producer APIs & Core Configuration

Module Overview

Goal: Teach what configuration knobs exist and what they control — without optimization yet.

What You'll Learn:

• Essential producer configurations grouped by purpose
• KafkaProducer API and ProducerRecord usage
• Synchronous vs asynchronous send patterns
• Callback mechanisms for handling responses
• When to use different send methods

What's NOT Covered Here:

• Performance optimization (covered in Module 8)
• Exactly-once semantics (covered in Module 5.3)
• Internal threads and buffers (covered in Module 7.1)
• Advanced tuning strategies (covered in Module 8.2)

1. The KafkaProducer API

Creating a Producer

The KafkaProducer class is the main entry point for producing records.

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

// Create configuration
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer");

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

Thread Safety

⚠️ Important: The KafkaProducer is thread-safe.

// ✅ GOOD: Share one producer across multiple threads
private static final KafkaProducer<String, String> producer = 
    new KafkaProducer<>(props);

// ❌ BAD: Creating a new producer per thread (unnecessary overhead)
// Don't do this unless you have specific isolation requirements

Best Practice: Use a single producer instance per application unless you need logical isolation.

2. Sending Records: Three Patterns

Pattern 1: Fire-and-Forget

Send and don't wait for confirmation.

ProducerRecord<String, String> record = 
    new ProducerRecord<>("topic-name", "key", "value");

producer.send(record); // Returns immediately

Characteristics:

• ✅ Fastest (non-blocking)
• ✅ Simplest code
• ❌ No delivery guarantee confirmation
• ❌ Silent failures (unless you monitor metrics)

Use When:

• Data loss is acceptable (e.g., non-critical metrics)
• Performance is paramount
• You have external monitoring

Pattern 2: Asynchronous Send with Callback

Send and handle result in a callback.

producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            // Success
            System.out.printf(
                "Sent to topic=%s partition=%d offset=%d timestamp=%d%n",
                metadata.topic(),
                metadata.partition(),
                metadata.offset(),
                metadata.timestamp()
            );
        } else {
            // Failure
            System.err.println("Send failed: " + exception.getMessage());
            // Handle error (retry, log, alert)
        }
    }
});

Using Lambda (Java 8+):

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        logger.error("Failed to send record", exception);
        // Handle error
    } else {
        logger.info("Record sent successfully to partition {}", 
            metadata.partition());
    }
});

Characteristics:

• ✅ Non-blocking (efficient)
• ✅ Delivery confirmation
• ✅ Error handling capability
• ✅ Recommended for most use cases

Use When:

• You need delivery confirmation without blocking
• You want to handle errors asynchronously
• You're building production applications

Pattern 3: Synchronous Send

Send and block until broker responds.

try {
    RecordMetadata metadata = producer.send(record).get(); // Blocks!
    System.out.println("Offset: " + metadata.offset());
} catch (ExecutionException e) {
    // Handle send failure
    System.err.println("Send failed: " + e.getCause());
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

With Timeout:

try {
    RecordMetadata metadata = producer.send(record).get(5, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    System.err.println("Send timed out after 5 seconds");
}

Characteristics:

• ✅ Immediate success/failure knowledge
• ✅ Simplest error handling
• ❌ Blocks thread (low throughput)
• ❌ Serializes sends (only one at a time per thread)

Use When:

• Sending single, critical records
• Application requires immediate confirmation before proceeding
• Throughput is not a concern
• Testing/debugging

3. Essential Producer Configurations

Configuration Categories

┌────────────────────────────────────────────────┐
│         PRODUCER CONFIGURATIONS                │
├────────────────────────────────────────────────┤
│ 1. Connection & Discovery                     │
│ 2. Serialization                              │
│ 3. Delivery Semantics                         │
│ 4. Batching & Buffering                       │
│ 5. Timeout & Retry                            │
│ 6. Compression                                │
│ 7. Idempotence & Transactions                 │
└────────────────────────────────────────────────┘

Category 1: Connection & Discovery

bootstrap.servers (Required)

props.put("bootstrap.servers", "localhost:9092");
// or multiple brokers
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");

Purpose: Initial broker(s) to connect to for cluster discovery.

Notes:

• You don't need ALL brokers — just a few for redundancy
• Producer discovers the full cluster automatically
• Format: host1:port1,host2:port2

client.id (Optional)

props.put("client.id", "payment-service-producer");

Purpose: Logical identifier for this producer in logs and metrics.

Benefits:

• Easier troubleshooting (shows in broker logs)
• Better monitoring (group metrics by client)
• Helps identify which application is producing

⚠️ Important: Make client.id Unique Per Producer Instance

// ❌ BAD: Same client.id across multiple instances
props.put("client.id", "payment-service"); // All instances use same ID

// ✅ GOOD: Unique per instance
String hostname = InetAddress.getLocalHost().getHostName();
props.put("client.id", "payment-service-producer-" + hostname);

// Or use instance ID:
props.put("client.id", "payment-service-" + instanceId);

Why it matters:

• Kafka does NOT enforce uniqueness of client.id
• Using the same ID across multiple instances causes:
  • Confused metrics (can't tell instances apart)
  • Misattributed logs
  • Difficult troubleshooting
• While Kafka won't reject connections, monitoring becomes impossible

Best Practice: Always include hostname, instance ID, or pod name in client.id.

Category 2: Serialization

key.serializer (Required)

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

Purpose: Converts key object to bytes.

Common Serializers:

• StringSerializer — for String keys
• IntegerSerializer — for Integer keys
• LongSerializer — for Long keys
• ByteArraySerializer — for byte[] keys
• Custom serializers for domain objects

value.serializer (Required)

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

Purpose: Converts value object to bytes.

Common Serializers:

• Same options as key serializer
• Often use: JsonSerializer, AvroSerializer, ProtobufSerializer

Note: Serialization details are covered in Module 3.1.

Category 3: Delivery Semantics

acks (Critical for Reliability)

props.put("acks", "all"); // or "0", "1", "all"

Purpose: Number of broker acknowledgments required before considering send successful.

Options:

Examples:

// Fire-and-forget (fastest, least durable)
props.put("acks", "0");

// Wait for leader only (default)
props.put("acks", "1");

// Wait for all replicas (most durable)
props.put("acks", "all");

Deep Dive: See Module 5.1 (Delivery Semantics Explained) for complete understanding.

retries & retry.backoff.ms

props.put("retries", Integer.MAX_VALUE); // Retry indefinitely
props.put("retry.backoff.ms", 100);      // Wait 100ms between retries

Purpose:

• retries: How many times to retry a failed send
• retry.backoff.ms: Wait time between retry attempts

Modern Best Practice:

// Let producer retry indefinitely (Kafka handles it intelligently)
props.put("retries", Integer.MAX_VALUE);

// Combined with delivery timeout
props.put("delivery.timeout.ms", 120000); // 2 minutes total

Note: Delivery semantics are explained in Module 5.

enable.idempotence

props.put("enable.idempotence", "true");

Purpose: Prevents duplicate messages even if retries happen.

When Enabled:

• Automatically sets: acks=all, retries=Integer.MAX_VALUE, max.in.flight.requests.per.connection=5
• Guarantees: No duplicates within a producer session

Default: true (Kafka 3.0+)

Deep Dive: See Module 5.2 (Idempotent Producer) for internals.

Category 4: Batching & Buffering

batch.size

props.put("batch.size", 16384); // 16 KB (default)

Purpose: Maximum size (in bytes) of a batch of records going to the same partition.

How It Works:

• Producer accumulates records into batches
• When batch reaches this size, it's sent immediately
• Smaller than this? Wait for linger.ms

Explained (not tuned yet):

• Larger batch = higher throughput, more latency
• Smaller batch = lower latency, less throughput

Tuning: Covered in Module 8.1 (Batching & Buffering).

linger.ms

props.put("linger.ms", 0); // Send immediately (default)
// OR
props.put("linger.ms", 10); // Wait up to 10ms to batch more records

Purpose: How long to wait before sending a batch (even if not full).

How It Works:

• 0: Send as soon as a record arrives (low latency)
• >0: Wait up to N milliseconds to batch more records (higher throughput)

Conceptual Trade-off:

• Low linger.ms → Lower latency, fewer records per batch
• High linger.ms → Higher throughput, more latency

Tuning: Covered in Module 8.2 (Latency vs Throughput).

buffer.memory

props.put("buffer.memory", 33554432); // 32 MB (default)

Purpose: Total memory (bytes) available to buffer records before sending.

What Happens When Full:

• send() blocks for up to max.block.ms
• If timeout expires → throws TimeoutException

Intuition:

• Larger buffer = can handle more in-flight records
• Too small = producer blocks waiting for space

Category 5: Timeouts

delivery.timeout.ms

props.put("delivery.timeout.ms", 120000); // 2 minutes (default)

Purpose: Maximum time from send() call until success or failure.

Includes:

• Time waiting in buffer
• Time waiting for acknowledgment
• Time for retries

Relationship with Other Timeouts:

delivery.timeout.ms ≥ linger.ms + request.timeout.ms

⚠️ Critical Configuration Rule:

delivery.timeout.ms must be greater than or equal to request.timeout.ms + buffer wait time.

Why? If delivery.timeout.ms < request.timeout.ms, the producer may time out the entire send operation before a single request even completes. This creates confusing behavior where:

• The delivery times out
• But the request is still waiting for broker response
• You get a timeout error even though the message might eventually succeed

Correct Configuration:

// ✅ GOOD: delivery timeout > request timeout
props.put("request.timeout.ms", 30000);      // 30 seconds per request
props.put("delivery.timeout.ms", 120000);    // 2 minutes total (4x request timeout)

// ❌ BAD: delivery timeout < request timeout
props.put("request.timeout.ms", 60000);      // 60 seconds per request
props.put("delivery.timeout.ms", 30000);     // 30 seconds total (!!)
// This will fail — delivery times out before request completes

Rule of Thumb:

delivery.timeout.ms ≥ (request.timeout.ms × expected_retries) + linger.ms + buffer_wait

For safety:

delivery.timeout.ms = request.timeout.ms × 3  // At minimum

request.timeout.ms

props.put("request.timeout.ms", 30000); // 30 seconds (default)

Purpose: How long to wait for broker response for a single request.

Use Case:

• If broker doesn't respond within this time, retry or fail

max.block.ms

props.put("max.block.ms", 60000); // 60 seconds (default)

Purpose: How long send() and partitionsFor() can block.

Blocks When:

• Buffer is full (buffer.memory exhausted)
• Metadata not available

What Happens:

• Blocks up to this time
• Then throws TimeoutException

Category 6: Compression

compression.type

props.put("compression.type", "none"); // Default: no compression
// Options: "none", "gzip", "snappy", "lz4", "zstd"

Purpose: Compress batches before sending.

Options:

Quick Guidance:

• Large payloads (JSON, text) → Use compression
• Small payloads (<1KB) → May not be worth it
• Network-bound → Compress
• CPU-bound → Don't compress

Deep Dive: Module 8.3 (Compression Strategies)

Category 7: Ordering & In-Flight Requests

max.in.flight.requests.per.connection

props.put("max.in.flight.requests.per.connection", 5); // Default

Purpose: How many requests can be sent to a broker before receiving responses.

Impact on Ordering:

• >1 + retries → Can reorder messages on failure
• =1 → Guarantees ordering (but slow)
• With idempotence enabled → Can use 5 safely

Ordering details: Module 4.1 (Partitioning Strategy & Message Ordering)

4. Putting It Together: Configuration Templates

Template 1: Default Producer (Balanced)

Properties props = new Properties();

// Connection
props.put("bootstrap.servers", "localhost:9092");
props.put("client.id", "my-app-producer");

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

// Delivery (defaults shown explicitly)
props.put("acks", "1");                        // Leader ack
props.put("retries", 2147483647);              // Retry indefinitely
props.put("delivery.timeout.ms", 120000);       // 2 minutes
props.put("enable.idempotence", "true");       // No duplicates

// Batching (defaults)
props.put("batch.size", 16384);                // 16 KB
props.put("linger.ms", 0);                     // Send immediately
props.put("buffer.memory", 33554432);          // 32 MB

// No compression
props.put("compression.type", "none");

Use For: General-purpose applications with balanced requirements.

Template 2: High-Reliability Producer

Properties props = new Properties();

props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("client.id", "critical-transactions-producer");

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

// Maximum reliability
props.put("acks", "all");                      // All replicas must ack
props.put("retries", Integer.MAX_VALUE);       // Retry indefinitely
props.put("enable.idempotence", "true");       // Prevent duplicates
props.put("max.in.flight.requests.per.connection", 5);

// Generous timeouts
props.put("delivery.timeout.ms", 300000);       // 5 minutes
props.put("request.timeout.ms", 60000);         // 1 minute

// Moderate batching
props.put("linger.ms", 10);                    // Wait 10ms for batching
props.put("batch.size", 32768);                // 32 KB

// Compression for efficiency
props.put("compression.type", "lz4");

Use For: Financial transactions, critical business data, regulatory compliance.

Template 3: High-Throughput Producer

Properties props = new Properties();

props.put("bootstrap.servers", "localhost:9092");
props.put("client.id", "analytics-producer");

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

// Relaxed reliability for speed
props.put("acks", "1");                        // Leader only
props.put("retries", 3);                       // Limited retries

// Aggressive batching
props.put("linger.ms", 20);                    // Wait 20ms
props.put("batch.size", 65536);                // 64 KB batches
props.put("buffer.memory", 67108864);          // 64 MB buffer

// Compression
props.put("compression.type", "snappy");

// More in-flight requests
props.put("max.in.flight.requests.per.connection", 10);

Use For: Analytics, logs, metrics where some loss is acceptable.

5. Working with Callbacks

Callback Execution

Important: Callbacks execute in the producer's I/O thread.

producer.send(record, (metadata, exception) -> {
    // ⚠️ This runs in producer I/O thread!
    if (exception == null) {
        // ✅ Fast operations OK
        logger.info("Sent: {}", metadata.offset());
    } else {
        // ❌ Don't do slow operations here (blocks other callbacks)
        // Don't call external APIs, databases, etc.
        logger.error("Failed", exception);
    }
});

Callback Best Practices

✅ Good: Fast Operations

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        // Update in-memory counter
        successCounter.increment();
        logger.debug("Sent to partition {}", metadata.partition());
    } else {
        errorCounter.increment();
        logger.error("Send failed", exception);
    }
});

❌ Bad: Slow Operations

producer.send(record, (metadata, exception) -> {
    // ❌ DON'T: Make HTTP calls
    httpClient.post("/notify", metadata);
    
    // ❌ DON'T: Write to database
    database.logSuccess(metadata);
    
    // ❌ DON'T: Do heavy processing
    processLargeFile();
});

✅ Better: Offload Work

private final ExecutorService executor = Executors.newFixedThreadPool(4);

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        // Offload slow work to separate thread pool
        executor.submit(() -> {
            httpClient.post("/notify", metadata);
        });
    }
});

6. Producer Lifecycle Management

Creating Producers

// Option 1: Properties
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
// ... more properties
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Option 2: Map (more modern)
Map<String, Object> config = new HashMap<>();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
KafkaProducer<String, String> producer = new KafkaProducer<>(config);

Closing Producers

Standard Close

// Flushes pending records and closes
producer.close();

What happens:

1. Sends all buffered records
2. Waits for acknowledgments
3. Closes network connections
4. Releases resources

Close with Timeout

// Wait up to 30 seconds, then force close
producer.close(Duration.ofSeconds(30));

Use when: You want bounded shutdown time.

Flush Before Close

try {
    // Ensure all records are sent
    producer.flush();
} finally {
    producer.close();
}

flush():

• Blocks until all buffered records are sent and acknowledged
• Useful before shutdown or checkpoints

Try-with-Resources (Java 7+)

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    producer.send(new ProducerRecord<>("topic", "key", "value"));
    // Producer automatically closed
}

Recommended for simple use cases.

7. Common Patterns

Pattern: Producer Singleton

public class ProducerManager {
    private static final KafkaProducer<String, String> producer;
    
    static {
        Properties props = new Properties();
        // ... configuration
        producer = new KafkaProducer<>(props);
        
        // Register shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            producer.close();
        }));
    }
    
    public static KafkaProducer<String, String> getProducer() {
        return producer;
    }
}

Pattern: Guaranteed Delivery with Sync Send

public void sendCriticalEvent(String key, String value) {
    ProducerRecord<String, String> record = 
        new ProducerRecord<>("critical-events", key, value);
    
    try {
        RecordMetadata metadata = producer.send(record).get();
        logger.info("Sent successfully: offset={}", metadata.offset());
    } catch (Exception e) {
        logger.error("Critical: failed to send event", e);
        // Alert operations team
        throw new RuntimeException("Failed to send critical event", e);
    }
}

Pattern: Async Send with Retry Logic

public void sendWithRetry(ProducerRecord<String, String> record, int maxAttempts) {
    AtomicInteger attempts = new AtomicInteger(0);
    
    Callback callback = new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null) {
                int attempt = attempts.incrementAndGet();
                if (attempt < maxAttempts) {
                    logger.warn("Retry attempt {}/{}", attempt, maxAttempts);
                    producer.send(record, this); // Retry with same callback
                } else {
                    logger.error("Failed after {} attempts", maxAttempts, exception);
                }
            }
        }
    };
    
    producer.send(record, callback);
}

Summary

Essential Configurations

Send Methods

Key Takeaways

✅ One producer instance can be shared across threads
✅ Async with callback is the recommended pattern
✅ Idempotence should be enabled for production
✅ acks=all for critical data
✅ Callbacks run in I/O thread — keep them fast

What's Next

• Module 3.1: Serialization & Data Modeling
  Deep dive into serializers, Avro, JSON, schema evolution

• Module 4.1: Partitioning Strategy & Message Ordering
  How partition selection works and ordering guarantees

• Module 5: Reliability, Delivery Guarantees & Semantics
  Complete understanding of acks, idempotence, exactly-once

Previous Module: ← 1.1 - Introduction & Producer Mental Model
Next Module: 3.1 - Serialization & Data Modeling →