Kafka Producer Configuration

This document provides a comprehensive reference for all Kafka producer configuration options, organized by functional area.

---

Required Configuration

These settings must be specified for every producer.

Configuration	Type	Description
bootstrap.servers	list	Comma-separated list of host:port pairs for initial connection
key.serializer	class	Serializer class for record keys
value.serializer	class	Serializer class for record values

bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

---

Reliability Configuration

Settings that control delivery guarantees and durability.

Acknowledgments

Configuration	Default	Range	Description
acks	all	0, 1, all	Number of acknowledgments required

Value	Behavior
0	Producer does not wait for acknowledgment
1	Leader writes to local log before acknowledging
all / -1	Leader waits for full ISR acknowledgment

Retries

Configuration	Default	Range	Description
retries	2147483647	≥ 0	Number of retry attempts for failed sends
retry.backoff.ms	100	≥ 0	Time between retry attempts
retry.backoff.max.ms	1000	≥ 0	Maximum retry backoff
delivery.timeout.ms	120000	≥ 0	Upper bound on time to report success/failure

Idempotence and Transactions

Configuration	Default	Description
enable.idempotence	true (3.0+)	Enable idempotent producer
transactional.id	null	Unique identifier for transactional producer
transaction.timeout.ms	60000	Transaction timeout

Idempotence Requirements

When enable.idempotence=true:

• acks must be all
• retries must be > 0
• max.in.flight.requests.per.connection must be ≤ 5

---

Performance Configuration

Settings that affect throughput and latency.

Batching

Configuration	Default	Range	Description
batch.size	16384	≥ 0	Maximum batch size in bytes
linger.ms	0	≥ 0	Time to wait for batch to fill
buffer.memory	33554432	≥ 0	Total memory for buffering
max.block.ms	60000	≥ 0	Time to block when buffer full

Request Handling

Configuration	Default	Range	Description
max.in.flight.requests.per.connection	5	≥ 1	Max unacknowledged requests per connection
max.request.size	1048576	≥ 0	Maximum size of a request
request.timeout.ms	30000	≥ 0	Time to wait for request response

Network

Configuration	Default	Range	Description
send.buffer.bytes	131072	≥ -1	TCP send buffer size (-1 = OS default)
receive.buffer.bytes	32768	≥ -1	TCP receive buffer size (-1 = OS default)
connections.max.idle.ms	540000	≥ 0	Close idle connections after this time
reconnect.backoff.ms	50	≥ 0	Initial reconnection backoff
reconnect.backoff.max.ms	1000	≥ 0	Maximum reconnection backoff

---

Compression Configuration

Configuration	Default	Options	Description
compression.type	none	none, gzip, snappy, lz4, zstd	Compression codec

Compression Trade-offs

Type	Compression Ratio	CPU Usage	Latency Impact
none	1.0x	None	None
gzip	~0.3-0.5x	High	Moderate
snappy	~0.5-0.7x	Low	Low
lz4	~0.5-0.7x	Low	Lowest
zstd	~0.4-0.6x	Medium	Low

Compression ratios are workload-dependent and should be validated with representative data.

---

Partitioning Configuration

Configuration	Default	Description
partitioner.class	DefaultPartitioner	Partitioner implementation
partitioner.ignore.keys	false	Ignore keys for partitioning (sticky only)
partitioner.adaptive.partitioning.enable	true	Enable adaptive partitioning
partitioner.availability.timeout.ms	0	Timeout for partition availability check

Default Partitioner Behavior

---

Serialization Configuration

Configuration	Default	Description
key.serializer	-	Key serializer class (required)
value.serializer	-	Value serializer class (required)

Common Serializers

Serializer	Class	Use Case
String	org.apache.kafka.common.serialization.StringSerializer	Text data
Bytes	org.apache.kafka.common.serialization.ByteArraySerializer	Binary data
Integer	org.apache.kafka.common.serialization.IntegerSerializer	Integer keys
Long	org.apache.kafka.common.serialization.LongSerializer	Long keys
JSON	Custom or library	Structured data
Avro	io.confluent.kafka.serializers.KafkaAvroSerializer	Schema Registry
Protobuf	io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer	Schema Registry

---

Security Configuration

SSL/TLS

Configuration	Default	Description
security.protocol	PLAINTEXT	Protocol: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
ssl.truststore.location	-	Truststore file path
ssl.truststore.password	-	Truststore password
ssl.keystore.location	-	Keystore file path
ssl.keystore.password	-	Keystore password
ssl.key.password	-	Private key password
ssl.endpoint.identification.algorithm	https	Hostname verification algorithm

SASL

Configuration	Default	Description
sasl.mechanism	GSSAPI	SASL mechanism: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI, OAUTHBEARER
sasl.jaas.config	-	JAAS configuration
sasl.client.callback.handler.class	-	Callback handler class
sasl.kerberos.service.name	-	Kerberos service name

# SASL/SCRAM example
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="producer" \
    password="secret";

---

Interceptor Configuration

Configuration	Default	Description
interceptor.classes	-	Comma-separated list of interceptor classes

public class MetricsInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // Add timestamp header, track metrics
        record.headers().add("send_time",
            Long.toString(System.currentTimeMillis()).getBytes());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // Track delivery metrics
    }
}

---

Metadata Configuration

Configuration	Default	Range	Description
metadata.max.age.ms	300000	≥ 0	Max age of cached metadata
metadata.max.idle.ms	300000	≥ 5000	Remove metadata for unused topics

---

Monitoring Configuration

Configuration	Default	Description
client.id	-	Client identifier for logging and metrics
metric.reporters	-	Comma-separated list of metric reporter classes
metrics.num.samples	2	Number of samples for metrics
metrics.sample.window.ms	30000	Metrics sample window
metrics.recording.level	INFO	Metrics recording level: INFO, DEBUG

---

Configuration Profiles

Low Latency Profile

# Minimize latency
acks=1
linger.ms=0
batch.size=16384
compression.type=lz4
max.in.flight.requests.per.connection=5

High Throughput Profile

# Maximize throughput
acks=all
linger.ms=50
batch.size=131072
buffer.memory=67108864
compression.type=zstd
max.in.flight.requests.per.connection=5

Maximum Reliability Profile

# Maximum durability
acks=all
enable.idempotence=true
retries=2147483647
delivery.timeout.ms=300000
max.in.flight.requests.per.connection=5

Transactional Profile

# Exactly-once semantics
acks=all
enable.idempotence=true
transactional.id=my-app-instance-1
transaction.timeout.ms=60000
max.in.flight.requests.per.connection=5

---

Configuration Validation

Check	Requirement
Idempotence	acks=all, retries > 0, max.in.flight <= 5
Transactions	enable.idempotence=true, transactional.id set
Ordering	max.in.flight.requests.per.connection=1 OR enable.idempotence=true

// Configuration validation example
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

// Validate before creating producer
try {
    new KafkaProducer<>(props);
} catch (ConfigException e) {
    log.error("Invalid configuration: {}", e.getMessage());
}

---

Related Documentation

• Producer Guide - Producer overview
• Batching - Batching details
• Compression - Compression options
• Idempotence - Idempotent producer
• Transactions - Transactional producer