Comma-separated list of URIs to publish to ZooKeeper for clients to use, if different than the listeners config property.

Default: null

In IaaS environments (e.g. Docker, Kubernetes, a cloud), advertised.listeners may need to be different from the interface to which a Kafka broker binds. That ensures that the Kafka broker advertises an address that is accessible from both local and external hosts. If not set, the value for listeners is used.

Unlike listeners it is invalid to advertise the 0.0.0.0 non-routable meta-address.

Dynamic reconfigurable configuration

Fully-qualified class name of the Authorizer for request authorization

Default: (empty)

Supports authorizers that implement the deprecated kafka.security.auth.Authorizer trait which was previously used for authorization before Kafka 2.4.0 (KIP-504 - Add new Java Authorizer Interface).

How often (in milliseconds) consumer offsets should be auto-committed when enable.auto.commit is enabled

Enables auto creation of a topic

Default: true

Enables auto leader balancing

Default: true

A background thread checks and triggers leader balance if required.

Reset policy — what to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g. because that data has been deleted):

Default: latest

The number of threads to use for various background processing tasks

Default: 10

Must be at least 1

Dynamic reconfigurable configuration

A comma-separated list of host:port pairs that are the addresses of one or more brokers in a Kafka cluster, e.g. localhost:9092 or localhost:9092,another.host:9092.

Default: (empty)

Enables automatic broker id generation of a Kafka broker

Default: true

When enabled (true) the value configured for reserved.broker.max.id should be checked.

Default: (randomly-generated)

Connection close delay on failed authentication: this is the time (in milliseconds) by which connection close will be delayed on authentication failure. This must be configured to be less than connections.max.idle.ms to prevent connection timeout.

Default: 100

Has to be at least 0

Idle connections timeout: the server socket processor threads close the connections that idle more than this

Default: 10 * 60 * 1000L

Name of the listener for communication between controller and brokers

Default: null (undefined), i.e. no dedicated endpoint

Must be defined in listener.security.protocol.map

Must be different than interBrokerListenerName

Broker will use the name to locate the endpoint in listeners, to listen for connections from the controller.

For example, if a broker’s config is:

On startup, the broker will start listening on "192.1.1.8:9094" with security protocol "SSL". On controller side, when it discovers a broker’s published endpoints through zookeeper, it will use the name to find the endpoint, which it will use to establish connection to the broker.

For example, if the broker’s published endpoints on zookeeper are:

and the controller’s config is:

then a controller will use "broker1:9094" with security protocol "SSL" to connect to the broker.

The default replication factor that is used for auto-created topics

Default: 1

Increase the default value to at least 2

Enables topic deletion

Default: true

When enabled (i.e. true) consumer offsets are committed automatically in the background (aka consumer auto commit) every auto.commit.interval.ms

Default: true

When disabled, offsets have to be committed manually (synchronously using KafkaConsumer.commitSync or asynchronously KafkaConsumer.commitAsync). On restart restore the position of a consumer using KafkaConsumer.seek.

Used when KafkaConsumer is created and creates a ConsumerCoordinator.

The hostname a Kafka broker listens on

Default: (empty)

Name of the listener for inter-broker communication (resolved per listener.security.protocol.map)

Default: security.inter.broker.protocol

Must be among advertised.listeners

It is an error to use together with security.inter.broker.protocol

Version of the inter-broker protocol

Default: the latest ApiVersion (e.g. 2.1-IV2)

Typically bumped up after all brokers were upgraded to a new version

Comma-separated list of ConsumerInterceptor class names.

Default: (empty)

How often the active KafkaController schedules the auto-leader-rebalance-task (aka AutoLeaderRebalance or AutoPreferredReplicaLeaderElection or auto leader balancing)

Default: 300

Allowed ratio of leader imbalance per broker. The controller would trigger a leader balance if it goes above this value per broker. The value is specified in percentage.

Default: 10

Comma-separated list of URIs and listener names that a Kafka broker will listen on

Default: PLAINTEXT://host.name:port

Use 0.0.0.0 to bind to all the network interfaces on a machine or leave it empty to bind to the default interface.

Dynamic reconfigurable configuration

Map of listener names and security protocols (key and value are separated by a colon and map entries are separated by commas). Each listener name should only appear once in the map.

Default: Map with PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL keys

This map must be defined for the same security protocol to be usable in more than one port or IP. For example, internal and external traffic can be separated even if SSL is required for both. Concretely, the user could define listeners with names INTERNAL and EXTERNAL and this property as: INTERNAL:SSL,EXTERNAL:SSL.

Different security (SSL and SASL) settings can be configured for each listener by adding a normalised prefix (the listener name is lowercased) to the config name. For example, to set a different keystore for the INTERNAL listener, a config with name listener.name.internal.ssl.keystore.location would be set. If the config for the listener name is not set, the config will fallback to the generic config (ssl.keystore.location).

Dynamic reconfigurable configuration

Dynamic reconfigurable configuration

Dynamic reconfigurable configuration

Enables the log cleaner process to run on a Kafka broker (true). Should be enabled if using any topics with a cleanup.policy=compact including the internal offsets topic. If disabled those topics will not be compacted and continually grow in size.

Default: true

Dynamic reconfigurable configuration

Dynamic reconfigurable configuration

Dynamic reconfigurable configuration

Dynamic reconfigurable configuration

Log Cleanup Policies (Strategies) — Log Compaction And Retention

Default: delete

Included in copyKafkaConfigToLog (to set cleanup.policy of topics)

The directory in which the log data is kept

Default: /tmp/kafka-logs

The directories in which the log data is kept

Default: log.dir

Use KafkaConfig.logDirs to access the current value.

Number of messages written to a log partition is kept in memory before flushing to disk (by forcing an fsync)

Default: Long.MaxValue (maximum possible long value)

E.g. if this was set to 1 we would fsync after every message; if it were 5 we would fsync after every five messages.

It is recommended not setting this and using replication for durability and allowing the operating system’s background flush capabilities as it is more efficient.

Must be at least 0

Topic-level configuration: flush.messages

How long (in millis) a message written to a log partition is kept in memory before flushing to disk (by forcing an fsync). If not set, the value in log.flush.scheduler.interval.ms is used.

Default: null (undefined)

E.g. if this was set to 1000 we would fsync after 1000 ms had passed.

Used exclusively when LogManager is requested to flushDirtyLogs.

It is recommended not setting this and using replication for durability and allowing the operating system’s background flush capabilities as it is more efficient.

Must be undefined or at least 0

Topic-level configuration: flush.messages

Maximum size (in bytes) of the offset index file (that maps offsets to file positions). It is preallocated and shrinked only after log rolls.

Default: 10 * 1024 * 1024

You generally should not need to change this setting.

Must be at least 0

Maximum size of a partition (which consists of log segments) to grow before discarding old segments and free up space.

log.retention.bytes is enforced at the partition level, multiply it by the number of partitions to compute the topic retention in bytes.

Default: -1L

Must be at least -1

How often (in millis) the LogManager (as kafka-log-retention task) checks whether any log is eligible for deletion

Default: 5 * 60 * 1000L (millis)

Must be at least 1

How long (in millis) to keep a log file before deleting it. -1 denotes no time limit

Default: 24 * 7 * 60 * 60 * 1000L (7 days)

Must be at least -1

Unless set, the value of log.retention.minutes is used.

How long (in mins) to keep a log file before deleting it. -1 denotes no time limit

Unless set, the value of log.retention.hours is used. Secondary to the log.retention.ms.

How long (in hours) to keep a log file before deleting it. -1 denotes no time limit

Considered the last unless log.retention.ms and log.retention.minutes were set.

Time (in millis) after which Kafka forces the log to roll even if the segment file isn’t full to ensure that retention can delete or compact old data.

Default: 604800000 (7 days)

Must be at least 1

The maximum size of a segment file of logs. Retention and cleaning are always done one file at a time so a larger segment size means fewer files but less granular control over retention.

Default: 1 * 1024 * 1024 * 1024

Must be at least 14 bytes (LegacyRecord.RECORD_OVERHEAD_V0)

Use KafkaConfig.logSegmentBytes to access the current value.

The maximum amount of data per-partition the server will return. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch will still be returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (topic config).

Use ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG

(KafkaConsumer) The maximum number of records returned from a Kafka Consumer when polling topics for records.

The default setting (-1) sets no upper bound on the number of records, i.e. Consumer.poll() will return as soon as either any data is available or the passed timeout expires.

max.poll.records was added to Kafka in 0.10.0.0 by KIP-41: KafkaConsumer Max Records.

From kafka-clients mailing list:

Use ConsumerConfig.MAX_POLL_RECORDS_CONFIG.

Internally, max.poll.records is used exclusively when KafkaConsumer is created (to create a Fetcher).

Dynamic reconfigurable configuration

The minimum number of replicas in ISR that is needed to commit a produce request with required.acks=-1 (or all)

Default: 1

Must be at least 1

When a Kafka producer sets acks to all (or -1), this configuration specifies the minimum number of replicas that must acknowledge a write for the write to be considered successful.

If this minimum cannot be met, then the producer will raise an exception (either NotEnoughReplicas or NotEnoughReplicasAfterAppend).

Used together with acks allows you to enforce greater durability guarantees.

A typical scenario would be to create a topic with a replication factor of 3, set min.insync.replicas to 2, and produce with acks of "all". This will ensure that the producer raises an exception if a majority of replicas do not receive a write.

The number of threads that KafkaServer uses for processing requests, which may include disk I/O

Default: 8

Must be at least 1

Dynamic reconfigurable configuration

The maximum number of connections allowed from each ip address.

Default: Int.MaxValue

Must be at least 0 (with 0 if there are overrides configured using max.connections.per.ip.overrides property)

Dynamic reconfigurable configuration

A comma-separated list of per-ip or hostname overrides to the default maximum number of connections, e.g. hostName:100,127.0.0.1:200

Default: (empty)

Dynamic reconfigurable configuration

The number of threads that SocketServer uses for the number of processors per endpoint (for receiving requests from the network and sending responses to the network)

Default: 3

Must be at least 1

Dynamic reconfigurable configuration

The number of threads per log data directory for log recovery at startup and flushing at shutdown

Default: 1

Must be at least 1

Dynamic reconfigurable configuration

The number of threads that can move replicas between log directories, which may include disk I/O

Default: null

The number of fetcher threads that ReplicaFetcherManager uses for replicating messages from a source broker

Default: 1

The higher the value the higher degree of I/O parallelism in a follower broker.

Dynamic reconfigurable configuration

Fully-qualified name of KafkaPrincipalBuilder implementation to build the KafkaPrincipal object for authorization

Default: null (i.e. DefaultKafkaPrincipalBuilder)

Supports the deprecated PrincipalBuilder interface which was previously used for client authentication over SSL.

If no principal builder is defined, the default behavior depends on the security protocol in use:

Used when ChannelBuilders is requested to create a KafkaPrincipalBuilder

Dynamic reconfigurable configuration

How long (in millis) a fetcher thread is going to sleep when there are no active partitions (while sending a fetch request) or after a fetch partition error and handlePartitionsWithErrors

Default: 1000 (millis)

Must be at least 0

The number of bytes of messages to attempt to fetch for each partition

Default: 1024 * 1024 (1 MB)

Must be at least 0

This is not an absolute maximum, if the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch will still be returned to ensure that progress can be made. The maximum record batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (topic config).

Maximum bytes expected for the entire fetch response

Default: 10 * 1024 * 1024

Must be at least 0

Records are fetched in batches, and if the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch will still be returned to ensure that progress can be made. As such, this is not an absolute maximum. The maximum record batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (topic config).

The number of queued requests allowed before blocking the network threads

Default: 500

Must be at least 1

How long to wait for a follower to consume up to the leader’s log end offset (LEO) before the leader removes the follower from the ISR of a partition

Default: 10000L (millis)

Socket timeout of ReplicaFetcherBlockingSend when sending network requests to partition leader brokers

Default: 30 * 1000 (30 seconds)

Should always be at least replica.fetch.wait.max.ms to prevent unnecessary socket timeouts

Security protocol for inter-broker communication

Default: PLAINTEXT

Possible values:

It is an error when defined with inter.broker.listener.name (as it then should only be in listener.security.protocol.map).

It is validated when the inter-broker communication uses a SASL protocol (SASL_PLAINTEXT or SASL_SSL) for…​FIXME

The maximum number of bytes in a socket request

Default: 100 * 1024 * 1024

Must be at least 1

Rules for mapping from the distinguished name from a client certificate to short name.

Default: DEFAULT (i.e. the distinguished name of a X.500 certificate is the principal)

The rules are evaluated in order and the first rule that matches a principal name is used to map it to a short name. Any later rules in the list are ignored.

This configuration is ignored for a custom KafkaPrincipalBuilder as defined by the principal.builder.class configuration.

Used when SslChannelBuilder is configured (to create a SslPrincipalMapper)

Client authentication

Default: none

Supported values (case-insensitive): required, requested, none

Non-reconfigurable

ReconfigurableConfigs

The maximum allowed timeout for transactions (in millis).

If a client’s requested transaction time exceed this, then the broker will return an error in InitProducerIdRequest. This prevents a client from a too large timeout that can stall consumers reading from topics included in the transaction.

Default: 15 minutes

Must be at least 1

Enables Unclean Partition Leader Election

Cluster-wide property: unclean.leader.election.enable

Topic-level property: unclean.leader.election.enable

Comma-separated list of Zookeeper hosts (as host:port pairs) that brokers register to, e.g. localhost:2181, 127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002

Default: (empty)

Zookeeper URIs can have an optional chroot path suffix at the end, e.g. 127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a

If the optional chroot path suffix is used, all paths are relative to this path.

It is recommended to include all the hosts in a Zookeeper ensemble (cluster)

The max time that the client waits to establish a connection to zookeeper

Default: zookeeper.session.timeout.ms

The maximum number of unacknowledged requests the client will send to Zookeeper before blocking. Has to be at least 1

Default: 10

Zookeeper session timeout

Default: 6000

Enables secure ACLs

Default: false