Kafka Command-Line Interface (CLI) Tools¶
Once you download Apache Kafka® and extract the files,
you can access several CLI tools under the /bin directory.
These tools enable you to start and stop Kafka, create and update topics, manage partitions and many more common operations.
You can get details on how to run each tool by by running it with no argument, or by using the --help argument.
Confluent Tip
When you Install Confluent Platform, all of the Kafka tools listed in this topic are also installed
in the
$CONFLUENT_HOME/bin folder, along with additional Confluent tools. Confluent has dropped the .sh extensions,
so you do not need to use the extensions when calling the Confluent versions of these tools. For more information,
see CLI Tools for Confluent Platform.To use the Kafka tools with Confluent Cloud, see
Using Apache Kafka CLI Tools with Confluent Cloud (blog post).
The following sections group the tools by function and provide basic usage information. In some cases, a tool is listed in more than one section.
Tools to start and stop Kafka and configure metadata¶
This section contains tools to start Kafka running either ZooKeeper or in KRaft-mode, and manage brokers.
kafka-server-start.sh¶
Use this tool start a Kafka Server. You can optionally pass the path to the properties file you want to use. If you are using ZooKeeper for metadata management, you must start ZooKeeper first. For KRaft-mode, generate a cluster ID and store it in the properties file. See the Kafka quickstart for an example of how to start Kafka.
Usage details
USAGE: ./kafka-server-start.sh [-daemon] server.properties [--override property=value]*
Option Description
------ -----------
--override <String> Optional property that should override values set in
server.properties file
--version Print version information and exit.
kafka-server-stop.sh¶
Stops the running Kafka server. This command takes no arguments.
Usage details
USAGE: ./kafka-server-stop.sh
zookeeper-server-start.sh¶
Starts the ZooKeeper server. See the Kafka quickstart for an example of how to use this command.
Usage details
USAGE: ./zookeeper-server-start.sh [-daemon] zookeeper.properties
zookeeper-server-stop.sh¶
Stops the ZooKeeper server. This command accepts no arguments.
Usage details
USAGE: ./zookeeper-server-stop.sh
kafka-storage.sh¶
A tool to generate a Cluster UUID and format storage with the ID when running in KRaft mode. See the Kafka quickstart for an example of how to use this command.
Usage details
USAGE: kafka-storage.sh [-h] {info,format,random-uuid} ...
The Kafka storage tool.
positional arguments:
{info,format,random-uuid}
info Get information about the Kafka log directories on this node.
format Format the Kafka log directories on this node.
random-uuid Print a random UUID.
optional arguments:
-h, --help show this help message and exit
kafka-cluster.sh¶
Use to get the ID of a cluster or unregister a cluster.
Usage details
USAGE: kafka-cluster.sh [-h] {cluster-id,unregister}
The Kafka cluster tool.
positional arguments:
{cluster-id,unregister}
cluster-id Get information about the ID of a cluster.
unregister Unregister a broker.
optional arguments:
-h, --help show this help message and exit
zookeeper-shell.sh¶
This tool connects to the ZooKeeper shell.
Usage details
USAGE: ./zookeeper-shell zookeeper_host:port[/path] [-zk-tls-config-file file] [args...]
kafka-features.sh¶
Manages feature flags to disable or enable functionality at runtime in Kafka.
Pass the describe argument to describe the current active feature flags, upgrade to upgrade one
or more feature flags, downgrade to downgrade one or more, and disable to disable
one or more feature flags, which is the same as downgrading the version to zero.
Usage details
USAGE: kafka-features.sh [-h] --bootstrap-server BOOTSTRAP_SERVER [--command-config COMMAND_CONFIG]
{describe,upgrade,downgrade,disable} ...
This tool manages feature flags in Kafka.
positional arguments:
{describe,upgrade,downgrade,disable}
describe Describe one or more feature flags.
upgrade Upgrade one or more feature flags.
downgrade Upgrade one or more feature flags.
disable Disable one or more feature flags. This is the same as downgrading the version to zero.
optional arguments:
-h, --help show this help message and exit
--bootstrap-server BOOTSTRAP_SERVER
A comma-separated list of host:port pairs to use for establishing the connection to the
Kafka cluster.
--command-config COMMAND_CONFIG
Property file containing configs to be passed to Admin Client.
kafka-broker-api-versions.sh¶
This tool retrieves and displays Broker information.
Example: ./bin/kafka-broker-api-versions.sh --bootstrap-server <hostname:port>.
Usage details
This tool helps to retrieve broker version information.
Option Description
------ -----------
--bootstrap-server <String: server(s) REQUIRED: The server to connect to.
to use for bootstrapping>
--command-config <String: command A property file containing configs to
config property file> be passed to Admin Client.
--help Print usage information.
--version Display Kafka version.
kafka-metadata-quorum.sh¶
Describes the metadata quorum status.
Pass the describe command to describe the current state of the metadata quorum.
Usage details
USAGE: kafka-metadata-quorum.sh [-h] --bootstrap-server BOOTSTRAP_SERVER [--command-config COMMAND_CONFIG] {describe}
...
This tool describes kraft metadata quorum status.
positional arguments:
{describe}
describe Describe the metadata quorum info
optional arguments:
-h, --help show this help message and exit
--bootstrap-server BOOTSTRAP_SERVER
A comma-separated list of host:port pairs to use for establishing the connection to the
Kafka cluster.
--command-config COMMAND_CONFIG
Property file containing configs to be passed to Admin Client.
kafka-metadata-shell.sh¶
A tool that enables you to interactively examine the metadata stored in a KRaft cluster. For more information, see
the Kafka Wiki.
Usage details
usage: metadata-shell-tool.sh [-h] [--cluster-id CLUSTER_ID] [--offset OFFSET] (--directory DIRECTORY |
--controllers CONTROLLERS) [command [command ...]]
The Apache Kafka metadata tool
positional arguments:
command The command to run.
optional arguments:
-h, --help show this help message and exit
--cluster-id CLUSTER_ID, -t CLUSTER_ID
The cluster id. Required when using --controllers
--directory DIRECTORY, -d DIRECTORY
The __cluster_metadata-0 directory to read.
--controllers CONTROLLERS, -q CONTROLLERS
The controller.quorum.voters.
--offset OFFSET, -o OFFSET
The offset to read up to
kafka-configs.sh¶
Use this tool to change and describe topic, client, user, broker, ip, or cluster link configuration settings.
To change a property, specify the entity-type to the desired entity (topic, broker, user, etc), and use the
alter option. See Topic Operations for example of how to modify topics.
Usage details
This tool helps to manipulate and describe entity config for a topic, client, user, broker, ip, or cluster link.
Option Description
------ -----------
--add-config <String> Key Value pairs of configs to add.
Square brackets can be used to group
values which contain commas: 'k1=v1,
k2=[v1,v2,v2],k3=v3'. The following
is a list of valid configurations:
For entity-type 'topics':
cleanup.policy
compression.type
confluent.append.record.interceptor.
classes
confluent.compacted.topic.prefer.tier.
fetch.ms
confluent.key.schema.validation
confluent.key.subject.name.strategy
confluent.placement.constraints
confluent.prefer.tier.fetch.ms
confluent.segment.speculative.
prefetch.enable
confluent.tier.cleaner.compact.min.
efficiency
confluent.tier.cleaner.compact.
segment.min.bytes
confluent.tier.cleaner.dual.compaction
confluent.tier.cleaner.enable
confluent.tier.cleaner.min.cleanable.
ratio
confluent.tier.enable
confluent.tier.local.hotset.bytes
confluent.tier.local.hotset.ms
confluent.tier.segment.hotset.roll.
min.bytes
confluent.value.schema.validation
confluent.value.subject.name.strategy
delete.retention.ms
file.delete.delay.ms
flush.messages
flush.ms
follower.replication.throttled.
replicas
index.interval.bytes
leader.replication.throttled.replicas
local.retention.bytes
local.retention.ms
max.compaction.lag.ms
max.message.bytes
message.downconversion.enable
message.format.version
message.timestamp.difference.max.ms
message.timestamp.type
min.cleanable.dirty.ratio
min.compaction.lag.ms
min.insync.replicas
preallocate
remote.storage.enable
retention.bytes
retention.ms
segment.bytes
segment.index.bytes
segment.jitter.ms
segment.ms
unclean.leader.election.enable
For entity-type 'brokers':
advertised.listeners
auto.create.topics.enable
background.threads
compression.type
confluent.append.record.interceptor.
classes
confluent.automatic.alter.leadership.
priority.retry.backoff.ms
confluent.backpressure.disk.enable
confluent.backpressure.disk.free.
threshold.bytes
confluent.backpressure.disk.produce.
bytes.per.second
confluent.backpressure.disk.threshold.
recovery.factor
confluent.backpressure.request.min.
broker.limit
confluent.backpressure.request.queue.
size.percentile
confluent.backpressure.types
confluent.balancer.enable
confluent.balancer.exclude.topic.names
confluent.balancer.exclude.topic.
prefixes
confluent.balancer.heal.uneven.load.
trigger
confluent.balancer.rebalancing.goals
confluent.balancer.throttle.bytes.per.
second
confluent.balancer.triggering.goals
confluent.broker.health.manager.
engine.request.handler.threads.stuck.
criteria
confluent.broker.health.manager.hard.
kill.duration.ms
confluent.broker.health.manager.
mitigation.enabled
confluent.broker.health.manager.num.
samples.before.broker.suspect
confluent.broker.health.manager.num.
samples.before.broker.unhealthy
confluent.broker.health.manager.
percentage.unhealthy.samples.before.
broker.suspect
confluent.broker.health.manager.
percentage.unhealthy.samples.before.
broker.unhealthy
confluent.broker.health.manager.
sample.duration.ms
confluent.broker.health.manager.
storage.background.threads.stuck.
criteria
confluent.broker.health.manager.
storage.network.threads.stuck.
criteria
confluent.broker.health.manager.
storage.request.handler.threads.
stuck.criteria
confluent.broker.load.average.service.
request.time.ms
confluent.broker.load.workload.
coefficient
confluent.clm.enabled
confluent.clm.frequency.in.hours
confluent.clm.max.backup.days
confluent.clm.min.delay.in.minutes
confluent.clm.thread.pool.size
confluent.clm.topic.retention.days.to.
backup.days
confluent.cluster.link.fetch.response.
min.bytes
confluent.cluster.link.fetch.response.
total.bytes
confluent.cluster.link.io.max.bytes.
per.second
confluent.cluster.link.replication.
quota.mode
confluent.compacted.topic.prefer.tier.
fetch.ms
confluent.durability.audit.checks
confluent.durability.events.allowed
confluent.eligible.controllers
confluent.fetch.partition.pruning.
enable
confluent.metadata.server.cluster.
registry.clusters
confluent.min.acks
confluent.network.health.manager.min.
healthy.network.samples
confluent.network.health.manager.
mitigation.enabled
confluent.network.health.manager.
network.sample.window.size
confluent.network.health.manager.
sample.duration.ms
confluent.plugins.cluster.link.policy.
max.destination.links.per.tenant
confluent.plugins.cluster.link.policy.
max.source.links.per.tenant
confluent.plugins.topic.policy.max.
partitions.per.cluster
confluent.plugins.topic.policy.max.
partitions.per.tenant
confluent.plugins.topic.policy.max.
topics.per.cluster
confluent.prefer.tier.fetch.ms
confluent.quota.dynamic.enable
confluent.quota.tenant.default.
controller.mutation.rate
confluent.quota.tenant.fetch.
multiplier
confluent.quota.tenant.produce.
multiplier
confluent.quota.tenant.user.quotas.
enable
confluent.reporters.telemetry.auto.
enable
confluent.security.event.router.config
confluent.segment.speculative.
prefetch.enable
confluent.telemetry.enabled
confluent.tier.cleaner.compact.min.
efficiency
confluent.tier.cleaner.compact.
segment.min.bytes
confluent.tier.cleaner.dedupe.buffer.
size
confluent.tier.cleaner.dual.compaction
confluent.tier.cleaner.enable
confluent.tier.cleaner.io.buffer.load.
factor
confluent.tier.cleaner.io.buffer.size
confluent.tier.cleaner.io.max.bytes.
per.second
confluent.tier.cleaner.min.cleanable.
ratio
confluent.tier.cleaner.num.threads
confluent.tier.enable
confluent.tier.fetcher.memorypool.
bytes
confluent.tier.local.hotset.bytes
confluent.tier.local.hotset.ms
confluent.tier.max.partition.fetch.
bytes.override
confluent.tier.metadata.snapshots.
enable
confluent.tier.metadata.snapshots.
interval.ms
confluent.tier.segment.hotset.roll.
min.bytes
confluent.tier.topic.delete.backoff.ms
confluent.tier.topic.delete.check.
interval.ms
confluent.tier.topic.delete.max.
inprogress.partitions
confluent.traffic.network.region.id
confluent.transaction.logging.
verbosity
connections.max.age.ms
fetch.max.bytes
floor.max.connection.creation.rate
follower.replication.throttled.rate
follower.replication.throttled.
replicas
leader.replication.throttled.rate
leader.replication.throttled.replicas
listener.security.protocol.map
listeners
log.cleaner.backoff.ms
log.cleaner.dedupe.buffer.size
log.cleaner.delete.retention.ms
log.cleaner.io.buffer.load.factor
log.cleaner.io.buffer.size
log.cleaner.io.max.bytes.per.second
log.cleaner.max.compaction.lag.ms
log.cleaner.min.cleanable.ratio
log.cleaner.min.compaction.lag.ms
log.cleaner.threads
log.cleanup.policy
log.deletion.max.segments.per.run
log.deletion.throttler.disk.free.
headroom.bytes
log.flush.interval.messages
log.flush.interval.ms
log.index.interval.bytes
log.index.size.max.bytes
log.message.downconversion.enable
log.message.timestamp.difference.max.
ms
log.message.timestamp.type
log.preallocate
log.retention.bytes
log.retention.ms
log.roll.jitter.ms
log.roll.ms
log.segment.bytes
log.segment.delete.delay.ms
max.connection.creation.rate
max.connection.creation.rate.per.ip.
enable.threshold
max.connections
max.connections.per.ip
max.connections.per.ip.overrides
message.max.bytes
metric.reporters
min.insync.replicas
num.io.threads
num.network.threads
num.partitions
num.recovery.threads.per.data.dir
num.replica.fetchers
principal.builder.class
quotas.expiration.time.ms
replica.alter.log.dirs.io.max.bytes.
per.second
sasl.enabled.mechanisms
sasl.jaas.config
sasl.kerberos.kinit.cmd
sasl.kerberos.min.time.before.relogin
sasl.kerberos.principal.to.local.rules
sasl.kerberos.service.name
sasl.kerberos.ticket.renew.jitter
sasl.kerberos.ticket.renew.window.
factor
sasl.login.refresh.buffer.seconds
sasl.login.refresh.min.period.seconds
sasl.login.refresh.window.factor
sasl.login.refresh.window.jitter
sasl.mechanism.inter.broker.protocol
socket.send.buffer.bytes
ssl.cipher.suites
ssl.client.auth
ssl.enabled.protocols
ssl.endpoint.identification.algorithm
ssl.engine.factory.class
ssl.key.password
ssl.keymanager.algorithm
ssl.keystore.certificate.chain
ssl.keystore.key
ssl.keystore.location
ssl.keystore.password
ssl.keystore.type
ssl.protocol
ssl.provider
ssl.secure.random.implementation
ssl.trustmanager.algorithm
ssl.truststore.certificates
ssl.truststore.location
ssl.truststore.password
ssl.truststore.type
unclean.leader.election.enable
For entity-type 'users':
SCRAM-SHA-256
SCRAM-SHA-512
consumer_byte_rate
controller_mutation_rate
producer_byte_rate
request_percentage
For entity-type 'clients':
consumer_byte_rate
controller_mutation_rate
producer_byte_rate
request_percentage
For entity-type 'ips':
connection_creation_rate
For entity-type 'cluster-links':
acl.filters
acl.sync.enable
acl.sync.ms
auto.create.mirror.topics.enable
auto.create.mirror.topics.filters
availability.check.consecutive.
failure.threshold
availability.check.ms
bootstrap.servers
client.dns.lookup
cluster.link.paused
cluster.link.prefix
cluster.link.retry.timeout.ms
confluent.replica.fetch.backoff.max.ms
connection.mode
connections.max.idle.ms
consumer.group.prefix.enable
consumer.offset.group.filters
consumer.offset.sync.enable
consumer.offset.sync.ms
link.fetcher.enforce.max.lagging.
partitions.ms
link.fetcher.flow.control
link.fetcher.lagging.partition.ms
link.fetcher.max.lagging.partitions
link.mode
local.listener.name
local.sasl.client.callback.handler.
class
local.sasl.jaas.config
local.sasl.kerberos.kinit.cmd
local.sasl.kerberos.min.time.before.
relogin
local.sasl.kerberos.service.name
local.sasl.kerberos.ticket.renew.
jitter
local.sasl.kerberos.ticket.renew.
window.factor
local.sasl.login.callback.handler.
class
local.sasl.login.class
local.sasl.login.connect.timeout.ms
local.sasl.login.read.timeout.ms
local.sasl.login.refresh.buffer.
seconds
local.sasl.login.refresh.min.period.
seconds
local.sasl.login.refresh.window.factor
local.sasl.login.refresh.window.jitter
local.sasl.login.retry.backoff.max.ms
local.sasl.login.retry.backoff.ms
local.sasl.mechanism
local.sasl.oauthbearer.clock.skew.
seconds
local.sasl.oauthbearer.expected.
audience
local.sasl.oauthbearer.expected.issuer
local.sasl.oauthbearer.jwks.endpoint.
refresh.ms
local.sasl.oauthbearer.jwks.endpoint.
retry.backoff.max.ms
local.sasl.oauthbearer.jwks.endpoint.
retry.backoff.ms
local.sasl.oauthbearer.jwks.endpoint.
url
local.sasl.oauthbearer.scope.claim.
name
local.sasl.oauthbearer.sub.claim.name
local.sasl.oauthbearer.token.endpoint.
url
local.security.protocol
local.ssl.cipher.suites
local.ssl.enabled.protocols
local.ssl.endpoint.identification.
algorithm
local.ssl.engine.factory.class
local.ssl.key.password
local.ssl.keymanager.algorithm
local.ssl.keystore.certificate.chain
local.ssl.keystore.key
local.ssl.keystore.location
local.ssl.keystore.password
local.ssl.keystore.type
local.ssl.protocol
local.ssl.provider
local.ssl.secure.random.implementation
local.ssl.trustmanager.algorithm
local.ssl.truststore.certificates
local.ssl.truststore.location
local.ssl.truststore.password
local.ssl.truststore.type
metadata.max.age.ms
num.cluster.link.fetchers
reconnect.backoff.max.ms
reconnect.backoff.ms
replica.fetch.backoff.ms
replica.fetch.max.bytes
replica.fetch.min.bytes
replica.fetch.response.max.bytes
replica.fetch.wait.max.ms
replica.socket.receive.buffer.bytes
replica.socket.timeout.ms
request.timeout.ms
retry.backoff.ms
reverse.connection.setup.timeout.ms
sasl.client.callback.handler.class
sasl.jaas.config
sasl.kerberos.kinit.cmd
sasl.kerberos.min.time.before.relogin
sasl.kerberos.service.name
sasl.kerberos.ticket.renew.jitter
sasl.kerberos.ticket.renew.window.
factor
sasl.login.callback.handler.class
sasl.login.class
sasl.login.connect.timeout.ms
sasl.login.read.timeout.ms
sasl.login.refresh.buffer.seconds
sasl.login.refresh.min.period.seconds
sasl.login.refresh.window.factor
sasl.login.refresh.window.jitter
sasl.login.retry.backoff.max.ms
sasl.login.retry.backoff.ms
sasl.mechanism
sasl.oauthbearer.clock.skew.seconds
sasl.oauthbearer.expected.audience
sasl.oauthbearer.expected.issuer
sasl.oauthbearer.jwks.endpoint.
refresh.ms
sasl.oauthbearer.jwks.endpoint.retry.
backoff.max.ms
sasl.oauthbearer.jwks.endpoint.retry.
backoff.ms
sasl.oauthbearer.jwks.endpoint.url
sasl.oauthbearer.scope.claim.name
sasl.oauthbearer.sub.claim.name
sasl.oauthbearer.token.endpoint.url
security.protocol
socket.connection.setup.timeout.max.ms
socket.connection.setup.timeout.ms
ssl.cipher.suites
ssl.enabled.protocols
ssl.endpoint.identification.algorithm
ssl.engine.factory.class
ssl.key.password
ssl.keymanager.algorithm
ssl.keystore.certificate.chain
ssl.keystore.key
ssl.keystore.location
ssl.keystore.password
ssl.keystore.type
ssl.protocol
ssl.provider
ssl.secure.random.implementation
ssl.trustmanager.algorithm
ssl.truststore.certificates
ssl.truststore.location
ssl.truststore.password
ssl.truststore.type
topic.config.sync.include
topic.config.sync.ms
Entity types 'users' and 'clients' may
be specified together to update
config for clients of a specific
user.
--add-config-file <String> Path to a properties file with configs
to add. See add-config for a list of
valid configurations.
--all List all configs for the given topic,
broker, or broker-logger entity
(includes static configuration when
the entity type is brokers)
--alter Alter the configuration for the entity.
--bootstrap-server <String: server to The Kafka server to connect to. This
connect to> is required for describing and
altering broker configs.
--broker <String> The broker's ID.
--broker-defaults The config defaults for all brokers.
--broker-logger <String> The broker's ID for its logger config.
--client <String> The client's ID.
--client-defaults The config defaults for all clients.
--cluster-link <String> The cluster link's name.
--command-config <String: command Property file containing configs to be
config property file> passed to Admin Client. This is used
only with --bootstrap-server option
for describing and altering broker
configs.
--delete-config <String> config keys to remove 'k1,k2'
--describe List configs for the given entity.
--entity-default Default entity name for
clients/users/brokers/ips (applies
to corresponding entity type in
command line)
--entity-name <String> Name of entity (topic name/client
id/user principal name/broker
id/ip/cluster link)
--entity-type <String> Type of entity
(topics/clients/users/brokers/broker-
loggers/ips/cluster-links)
--force Suppress console prompts
--help Print usage information.
--ip <String> The IP address.
--ip-defaults The config defaults for all IPs.
--replica-placement <String: Replica This configuration is a JSON object
placement JSON file path.> that controls the set of brokers
(replicas) which will always be
allowed to join the ISR. And the set
of brokers (observers) which are not
allowed to join the ISR. The format
of JSON is:
{
"version": 1,
"replicas": [
{
"count": 2,
"constraints": {"rack": "east-1"}
},
{
"count": 1,
"constraints": {"rack": "east-2"}
}
],
"observers":[
{
"count": 1,
"constraints": {"rack": "west-1"}
}
]
}
--topic <String> The topic's name.
--user <String> The user's principal name.
--user-defaults The config defaults for all users.
--version Display Kafka version.
--zk-tls-config-file <String: Identifies the file where ZooKeeper
ZooKeeper TLS configuration> client TLS connectivity properties
are defined. Any properties other
than zookeeper.clientCnxnSocket,
zookeeper.ssl.cipher.suites,
zookeeper.ssl.client.enable,
zookeeper.ssl.crl.enable, zookeeper.
ssl.enabled.protocols, zookeeper.ssl.
endpoint.identification.algorithm,
zookeeper.ssl.keystore.location,
zookeeper.ssl.keystore.password,
zookeeper.ssl.keystore.type,
zookeeper.ssl.ocsp.enable, zookeeper.
ssl.protocol, zookeeper.ssl.
truststore.location, zookeeper.ssl.
truststore.password, zookeeper.ssl.
truststore.type are ignored.
--zookeeper <String: urls> DEPRECATED. The connection string for
the zookeeper connection in the form
host:port. Multiple URLS can be
given to allow fail-over. Required
when configuring SCRAM credentials
for users or dynamic broker configs
when the relevant broker(s) are
down. Not allowed otherwise.
zookeeper-security-migration.sh¶
This tool is used to restrict or provide access to ZooKeeper metadata. The tool updates the ACLs of znodes.
Usage details
ZooKeeper Migration Tool Help. This tool updates the ACLs of znodes as part of the process of setting up ZooKeeper authentication.
Option Description
------ -----------
--enable.path.check Checks if all the root paths exist in
ZooKeeper before migration. If not,
exit the command.
--help Print usage information.
--version Display Kafka version.
--zk-tls-config-file <String: Identifies the file where ZooKeeper
ZooKeeper TLS configuration> client TLS connectivity properties
are defined. Any properties other
than zookeeper.ssl.keystore.
password, zookeeper.ssl.truststore.
location, zookeeper.ssl.truststore.
type, zookeeper.ssl.ocsp.enable,
zookeeper.ssl.crl.enable, zookeeper.
ssl.truststore.password, zookeeper.
ssl.protocol, zookeeper.ssl.keystore.
type, zookeeper.ssl.client.enable,
zookeeper.ssl.cipher.suites,
zookeeper.clientCnxnSocket,
zookeeper.ssl.endpoint.
identification.algorithm, zookeeper.
ssl.enabled.protocols, zookeeper.ssl.
keystore.location are ignored.
--zookeeper.acl <String> Indicates whether to make the Kafka
znodes in ZooKeeper secure or
unsecure. The options are 'secure'
and 'unsecure'
--zookeeper.connect <String> Sets the ZooKeeper connect string
(ensemble). This parameter takes a
comma-separated list of host:port
pairs. (default: localhost:2181)
--zookeeper.connection.timeout Sets the ZooKeeper connection timeout.
<Integer> (default: 30000)
--zookeeper.session.timeout <Integer> Sets the ZooKeeper session timeout.
(default: 30000)
Topic, partition, and replica tools¶
kafka-topics.sh¶
Use to create, delete, or change a topic. You can also use the tool to retrieve a list of topics associated with a Kafka cluster. See Topic Operations.
Example: kafka-topics.sh --bootstrap-server <hostname:port> --topic test-topic --partitions 3.
Usage details
Create, delete, describe, or change a topic.
Option Description
------ -----------
--alter Alter the number of partitions,
replica assignment, and/or
configuration for the topic.
--at-min-isr-partitions if set when describing topics, only
show partitions whose isr count is
equal to the configured minimum.
--bootstrap-server <String: server to REQUIRED: The Kafka server to connect
connect to> to.
--command-config <String: command Property file containing configs to be
config property file> passed to Admin Client. This is used
only with --bootstrap-server option
for describing and altering broker
configs.
--config <String: name=value> A topic configuration override for the
topic being created or altered. The
following is a list of valid
configurations:
cleanup.policy
compression.type
confluent.append.record.interceptor.
classes
confluent.compacted.topic.prefer.tier.
fetch.ms
confluent.key.schema.validation
confluent.key.subject.name.strategy
confluent.placement.constraints
confluent.prefer.tier.fetch.ms
confluent.segment.speculative.
prefetch.enable
confluent.tier.cleaner.compact.min.
efficiency
confluent.tier.cleaner.compact.
segment.min.bytes
confluent.tier.cleaner.dual.compaction
confluent.tier.cleaner.enable
confluent.tier.cleaner.min.cleanable.
ratio
confluent.tier.enable
confluent.tier.local.hotset.bytes
confluent.tier.local.hotset.ms
confluent.tier.segment.hotset.roll.
min.bytes
confluent.value.schema.validation
confluent.value.subject.name.strategy
delete.retention.ms
file.delete.delay.ms
flush.messages
flush.ms
follower.replication.throttled.
replicas
index.interval.bytes
leader.replication.throttled.replicas
local.retention.bytes
local.retention.ms
max.compaction.lag.ms
max.message.bytes
message.downconversion.enable
message.format.version
message.timestamp.difference.max.ms
message.timestamp.type
min.cleanable.dirty.ratio
min.compaction.lag.ms
min.insync.replicas
preallocate
remote.storage.enable
retention.bytes
retention.ms
segment.bytes
segment.index.bytes
segment.jitter.ms
segment.ms
unclean.leader.election.enable
See the Kafka documentation for full
details on the topic configs. It is
supported only in combination with --
create if --bootstrap-server option
is used (the kafka-configs CLI
supports altering topic configs with
a --bootstrap-server option).
--create Create a new topic.
--delete Delete a topic
--delete-config <String: name> A topic configuration override to be
removed for an existing topic (see
the list of configurations under the
--config option). Not supported with
the --bootstrap-server option.
--describe List details for the given topics.
--disable-rack-aware Disable rack aware replica assignment
--exclude-internal exclude internal topics when running
list or describe command. The
internal topics will be listed by
default
--help Print usage information.
--if-exists if set when altering or deleting or
describing topics, the action will
only execute if the topic exists.
--if-not-exists if set when creating topics, the
action will only execute if the
topic does not already exist.
--invalid-replica-placement-partitions if set when describing topics, only
show partitions whose placement
does not adhere to the replica
placement constraints.
--list List all available topics.
--partitions <Integer: num partitions> The number of partitions for the topic
being created or altered (WARNING:
If partitions are increased for a
topic that has a key, the partition
logic or ordering of the messages
will be affected). If not supplied
for create, defaults to the cluster
default.
--replica-assignment <String: A list of manual partition-to-broker
broker_id_for_part1_replica1 : assignments for the topic being
broker_id_for_part1_replica2 , created or altered.
broker_id_for_part2_replica1 :
broker_id_for_part2_replica2 , ...>
--replica-placement <String: Replica This configuration is a JSON object
placement JSON file path.> that controls the set of brokers
(replicas) which will always be
allowed to join the ISR. And the set
of brokers (observers) which are not
allowed to join the ISR. The format
of JSON is:
{
"version": 1,
"replicas": [
{
"count": 2,
"constraints": {"rack": "east-1"}
},
{
"count": 1,
"constraints": {"rack": "east-2"}
}
],
"observers":[
{
"count": 1,
"constraints": {"rack": "west-1"}
}
]
}
--replication-factor <Integer: The replication factor for each
replication factor> partition in the topic being
created. If not supplied, defaults
to the cluster default.
--topic <String: topic> The topic to create, alter, describe
or delete. It also accepts a regular
expression, except for --create
option. Put topic name in double
quotes and use the '\' prefix to
escape regular expression symbols; e.
g. "test\.topic".
--topic-id <String: topic-id> The topic-id to describe.This is used
only with --bootstrap-server option
for describing topics.
--topics-with-overrides if set when describing topics, only
show topics that have overridden
configs
--unavailable-partitions if set when describing topics, only
show partitions whose leader is not
available
--under-min-isr-partitions if set when describing topics, only
show partitions whose isr count is
less than the configured minimum.
--under-replicated-partitions if set when describing topics, only
show under replicated partitions
--version Display Kafka version.
kafka-configs.sh¶
Use this tool to change and describe topic, client, user, broker, ip, or cluster link configuration settings.
To change a property, specify the entity-type to the desired entity (topic, broker, user, etc), and use the
alter option. See Topic Operations for example of how to modify topics.
Usage details
This tool helps to manipulate and describe entity config for a topic, client, user, broker, ip, or cluster link.
Option Description
------ -----------
--add-config <String> Key Value pairs of configs to add.
Square brackets can be used to group
values which contain commas: 'k1=v1,
k2=[v1,v2,v2],k3=v3'. The following
is a list of valid configurations:
For entity-type 'topics':
cleanup.policy
compression.type
confluent.append.record.interceptor.
classes
confluent.compacted.topic.prefer.tier.
fetch.ms
confluent.key.schema.validation
confluent.key.subject.name.strategy
confluent.placement.constraints
confluent.prefer.tier.fetch.ms
confluent.segment.speculative.
prefetch.enable
confluent.tier.cleaner.compact.min.
efficiency
confluent.tier.cleaner.compact.
segment.min.bytes
confluent.tier.cleaner.dual.compaction
confluent.tier.cleaner.enable
confluent.tier.cleaner.min.cleanable.
ratio
confluent.tier.enable
confluent.tier.local.hotset.bytes
confluent.tier.local.hotset.ms
confluent.tier.segment.hotset.roll.
min.bytes
confluent.value.schema.validation
confluent.value.subject.name.strategy
delete.retention.ms
file.delete.delay.ms
flush.messages
flush.ms
follower.replication.throttled.
replicas
index.interval.bytes
leader.replication.throttled.replicas
local.retention.bytes
local.retention.ms
max.compaction.lag.ms
max.message.bytes
message.downconversion.enable
message.format.version
message.timestamp.difference.max.ms
message.timestamp.type
min.cleanable.dirty.ratio
min.compaction.lag.ms
min.insync.replicas
preallocate
remote.storage.enable
retention.bytes
retention.ms
segment.bytes
segment.index.bytes
segment.jitter.ms
segment.ms
unclean.leader.election.enable
For entity-type 'brokers':
advertised.listeners
auto.create.topics.enable
background.threads
compression.type
confluent.append.record.interceptor.
classes
confluent.automatic.alter.leadership.
priority.retry.backoff.ms
confluent.backpressure.disk.enable
confluent.backpressure.disk.free.
threshold.bytes
confluent.backpressure.disk.produce.
bytes.per.second
confluent.backpressure.disk.threshold.
recovery.factor
confluent.backpressure.request.min.
broker.limit
confluent.backpressure.request.queue.
size.percentile
confluent.backpressure.types
confluent.balancer.enable
confluent.balancer.exclude.topic.names
confluent.balancer.exclude.topic.
prefixes
confluent.balancer.heal.uneven.load.
trigger
confluent.balancer.rebalancing.goals
confluent.balancer.throttle.bytes.per.
second
confluent.balancer.triggering.goals
confluent.broker.health.manager.
engine.request.handler.threads.stuck.
criteria
confluent.broker.health.manager.hard.
kill.duration.ms
confluent.broker.health.manager.
mitigation.enabled
confluent.broker.health.manager.num.
samples.before.broker.suspect
confluent.broker.health.manager.num.
samples.before.broker.unhealthy
confluent.broker.health.manager.
percentage.unhealthy.samples.before.
broker.suspect
confluent.broker.health.manager.
percentage.unhealthy.samples.before.
broker.unhealthy
confluent.broker.health.manager.
sample.duration.ms
confluent.broker.health.manager.
storage.background.threads.stuck.
criteria
confluent.broker.health.manager.
storage.network.threads.stuck.
criteria
confluent.broker.health.manager.
storage.request.handler.threads.
stuck.criteria
confluent.broker.load.average.service.
request.time.ms
confluent.broker.load.workload.
coefficient
confluent.clm.enabled
confluent.clm.frequency.in.hours
confluent.clm.max.backup.days
confluent.clm.min.delay.in.minutes
confluent.clm.thread.pool.size
confluent.clm.topic.retention.days.to.
backup.days
confluent.cluster.link.fetch.response.
min.bytes
confluent.cluster.link.fetch.response.
total.bytes
confluent.cluster.link.io.max.bytes.
per.second
confluent.cluster.link.replication.
quota.mode
confluent.compacted.topic.prefer.tier.
fetch.ms
confluent.durability.audit.checks
confluent.durability.events.allowed
confluent.eligible.controllers
confluent.fetch.partition.pruning.
enable
confluent.metadata.server.cluster.
registry.clusters
confluent.min.acks
confluent.network.health.manager.min.
healthy.network.samples
confluent.network.health.manager.
mitigation.enabled
confluent.network.health.manager.
network.sample.window.size
confluent.network.health.manager.
sample.duration.ms
confluent.plugins.cluster.link.policy.
max.destination.links.per.tenant
confluent.plugins.cluster.link.policy.
max.source.links.per.tenant
confluent.plugins.topic.policy.max.
partitions.per.cluster
confluent.plugins.topic.policy.max.
partitions.per.tenant
confluent.plugins.topic.policy.max.
topics.per.cluster
confluent.prefer.tier.fetch.ms
confluent.quota.dynamic.enable
confluent.quota.tenant.default.
controller.mutation.rate
confluent.quota.tenant.fetch.
multiplier
confluent.quota.tenant.produce.
multiplier
confluent.quota.tenant.user.quotas.
enable
confluent.reporters.telemetry.auto.
enable
confluent.security.event.router.config
confluent.segment.speculative.
prefetch.enable
confluent.telemetry.enabled
confluent.tier.cleaner.compact.min.
efficiency
confluent.tier.cleaner.compact.
segment.min.bytes
confluent.tier.cleaner.dedupe.buffer.
size
confluent.tier.cleaner.dual.compaction
confluent.tier.cleaner.enable
confluent.tier.cleaner.io.buffer.load.
factor
confluent.tier.cleaner.io.buffer.size
confluent.tier.cleaner.io.max.bytes.
per.second
confluent.tier.cleaner.min.cleanable.
ratio
confluent.tier.cleaner.num.threads
confluent.tier.enable
confluent.tier.fetcher.memorypool.
bytes
confluent.tier.local.hotset.bytes
confluent.tier.local.hotset.ms
confluent.tier.max.partition.fetch.
bytes.override
confluent.tier.metadata.snapshots.
enable
confluent.tier.metadata.snapshots.
interval.ms
confluent.tier.segment.hotset.roll.
min.bytes
confluent.tier.topic.delete.backoff.ms
confluent.tier.topic.delete.check.
interval.ms
confluent.tier.topic.delete.max.
inprogress.partitions
confluent.traffic.network.region.id
confluent.transaction.logging.
verbosity
connections.max.age.ms
fetch.max.bytes
floor.max.connection.creation.rate
follower.replication.throttled.rate
follower.replication.throttled.
replicas
leader.replication.throttled.rate
leader.replication.throttled.replicas
listener.security.protocol.map
listeners
log.cleaner.backoff.ms
log.cleaner.dedupe.buffer.size
log.cleaner.delete.retention.ms
log.cleaner.io.buffer.load.factor
log.cleaner.io.buffer.size
log.cleaner.io.max.bytes.per.second
log.cleaner.max.compaction.lag.ms
log.cleaner.min.cleanable.ratio
log.cleaner.min.compaction.lag.ms
log.cleaner.threads
log.cleanup.policy
log.deletion.max.segments.per.run
log.deletion.throttler.disk.free.
headroom.bytes
log.flush.interval.messages
log.flush.interval.ms
log.index.interval.bytes
log.index.size.max.bytes
log.message.downconversion.enable
log.message.timestamp.difference.max.
ms
log.message.timestamp.type
log.preallocate
log.retention.bytes
log.retention.ms
log.roll.jitter.ms
log.roll.ms
log.segment.bytes
log.segment.delete.delay.ms
max.connection.creation.rate
max.connection.creation.rate.per.ip.
enable.threshold
max.connections
max.connections.per.ip
max.connections.per.ip.overrides
message.max.bytes
metric.reporters
min.insync.replicas
num.io.threads
num.network.threads
num.partitions
num.recovery.threads.per.data.dir
num.replica.fetchers
principal.builder.class
quotas.expiration.time.ms
replica.alter.log.dirs.io.max.bytes.
per.second
sasl.enabled.mechanisms
sasl.jaas.config
sasl.kerberos.kinit.cmd
sasl.kerberos.min.time.before.relogin
sasl.kerberos.principal.to.local.rules
sasl.kerberos.service.name
sasl.kerberos.ticket.renew.jitter
sasl.kerberos.ticket.renew.window.
factor
sasl.login.refresh.buffer.seconds
sasl.login.refresh.min.period.seconds
sasl.login.refresh.window.factor
sasl.login.refresh.window.jitter
sasl.mechanism.inter.broker.protocol
socket.send.buffer.bytes
ssl.cipher.suites
ssl.client.auth
ssl.enabled.protocols
ssl.endpoint.identification.algorithm
ssl.engine.factory.class
ssl.key.password
ssl.keymanager.algorithm
ssl.keystore.certificate.chain
ssl.keystore.key
ssl.keystore.location
ssl.keystore.password
ssl.keystore.type
ssl.protocol
ssl.provider
ssl.secure.random.implementation
ssl.trustmanager.algorithm
ssl.truststore.certificates
ssl.truststore.location
ssl.truststore.password
ssl.truststore.type
unclean.leader.election.enable
For entity-type 'users':
SCRAM-SHA-256
SCRAM-SHA-512
consumer_byte_rate
controller_mutation_rate
producer_byte_rate
request_percentage
For entity-type 'clients':
consumer_byte_rate
controller_mutation_rate
producer_byte_rate
request_percentage
For entity-type 'ips':
connection_creation_rate
For entity-type 'cluster-links':
acl.filters
acl.sync.enable
acl.sync.ms
auto.create.mirror.topics.enable
auto.create.mirror.topics.filters
availability.check.consecutive.
failure.threshold
availability.check.ms
bootstrap.servers
client.dns.lookup
cluster.link.paused
cluster.link.prefix
cluster.link.retry.timeout.ms
confluent.replica.fetch.backoff.max.ms
connection.mode
connections.max.idle.ms
consumer.group.prefix.enable
consumer.offset.group.filters
consumer.offset.sync.enable
consumer.offset.sync.ms
link.fetcher.enforce.max.lagging.
partitions.ms
link.fetcher.flow.control
link.fetcher.lagging.partition.ms
link.fetcher.max.lagging.partitions
link.mode
local.listener.name
local.sasl.client.callback.handler.
class
local.sasl.jaas.config
local.sasl.kerberos.kinit.cmd
local.sasl.kerberos.min.time.before.
relogin
local.sasl.kerberos.service.name
local.sasl.kerberos.ticket.renew.
jitter
local.sasl.kerberos.ticket.renew.
window.factor
local.sasl.login.callback.handler.
class
local.sasl.login.class
local.sasl.login.connect.timeout.ms
local.sasl.login.read.timeout.ms
local.sasl.login.refresh.buffer.
seconds
local.sasl.login.refresh.min.period.
seconds
local.sasl.login.refresh.window.factor
local.sasl.login.refresh.window.jitter
local.sasl.login.retry.backoff.max.ms
local.sasl.login.retry.backoff.ms
local.sasl.mechanism
local.sasl.oauthbearer.clock.skew.
seconds
local.sasl.oauthbearer.expected.
audience
local.sasl.oauthbearer.expected.issuer
local.sasl.oauthbearer.jwks.endpoint.
refresh.ms
local.sasl.oauthbearer.jwks.endpoint.
retry.backoff.max.ms
local.sasl.oauthbearer.jwks.endpoint.
retry.backoff.ms
local.sasl.oauthbearer.jwks.endpoint.
url
local.sasl.oauthbearer.scope.claim.
name
local.sasl.oauthbearer.sub.claim.name
local.sasl.oauthbearer.token.endpoint.
url
local.security.protocol
local.ssl.cipher.suites
local.ssl.enabled.protocols
local.ssl.endpoint.identification.
algorithm
local.ssl.engine.factory.class
local.ssl.key.password
local.ssl.keymanager.algorithm
local.ssl.keystore.certificate.chain
local.ssl.keystore.key
local.ssl.keystore.location
local.ssl.keystore.password
local.ssl.keystore.type
local.ssl.protocol
local.ssl.provider
local.ssl.secure.random.implementation
local.ssl.trustmanager.algorithm
local.ssl.truststore.certificates
local.ssl.truststore.location
local.ssl.truststore.password
local.ssl.truststore.type
metadata.max.age.ms
num.cluster.link.fetchers
reconnect.backoff.max.ms
reconnect.backoff.ms
replica.fetch.backoff.ms
replica.fetch.max.bytes
replica.fetch.min.bytes
replica.fetch.response.max.bytes
replica.fetch.wait.max.ms
replica.socket.receive.buffer.bytes
replica.socket.timeout.ms
request.timeout.ms
retry.backoff.ms
reverse.connection.setup.timeout.ms
sasl.client.callback.handler.class
sasl.jaas.config
sasl.kerberos.kinit.cmd
sasl.kerberos.min.time.before.relogin
sasl.kerberos.service.name
sasl.kerberos.ticket.renew.jitter
sasl.kerberos.ticket.renew.window.
factor
sasl.login.callback.handler.class
sasl.login.class
sasl.login.connect.timeout.ms
sasl.login.read.timeout.ms
sasl.login.refresh.buffer.seconds
sasl.login.refresh.min.period.seconds
sasl.login.refresh.window.factor
sasl.login.refresh.window.jitter
sasl.login.retry.backoff.max.ms
sasl.login.retry.backoff.ms
sasl.mechanism
sasl.oauthbearer.clock.skew.seconds
sasl.oauthbearer.expected.audience
sasl.oauthbearer.expected.issuer
sasl.oauthbearer.jwks.endpoint.
refresh.ms
sasl.oauthbearer.jwks.endpoint.retry.
backoff.max.ms
sasl.oauthbearer.jwks.endpoint.retry.
backoff.ms
sasl.oauthbearer.jwks.endpoint.url
sasl.oauthbearer.scope.claim.name
sasl.oauthbearer.sub.claim.name
sasl.oauthbearer.token.endpoint.url
security.protocol
socket.connection.setup.timeout.max.ms
socket.connection.setup.timeout.ms
ssl.cipher.suites
ssl.enabled.protocols
ssl.endpoint.identification.algorithm
ssl.engine.factory.class
ssl.key.password
ssl.keymanager.algorithm
ssl.keystore.certificate.chain
ssl.keystore.key
ssl.keystore.location
ssl.keystore.password
ssl.keystore.type
ssl.protocol
ssl.provider
ssl.secure.random.implementation
ssl.trustmanager.algorithm
ssl.truststore.certificates
ssl.truststore.location
ssl.truststore.password
ssl.truststore.type
topic.config.sync.include
topic.config.sync.ms
Entity types 'users' and 'clients' may
be specified together to update
config for clients of a specific
user.
--add-config-file <String> Path to a properties file with configs
to add. See add-config for a list of
valid configurations.
--all List all configs for the given topic,
broker, or broker-logger entity
(includes static configuration when
the entity type is brokers)
--alter Alter the configuration for the entity.
--bootstrap-server <String: server to The Kafka server to connect to. This
connect to> is required for describing and
altering broker configs.
--broker <String> The broker's ID.
--broker-defaults The config defaults for all brokers.
--broker-logger <String> The broker's ID for its logger config.
--client <String> The client's ID.
--client-defaults The config defaults for all clients.
--cluster-link <String> The cluster link's name.
--command-config <String: command Property file containing configs to be
config property file> passed to Admin Client. This is used
only with --bootstrap-server option
for describing and altering broker
configs.
--delete-config <String> config keys to remove 'k1,k2'
--describe List configs for the given entity.
--entity-default Default entity name for
clients/users/brokers/ips (applies
to corresponding entity type in
command line)
--entity-name <String> Name of entity (topic name/client
id/user principal name/broker
id/ip/cluster link)
--entity-type <String> Type of entity
(topics/clients/users/brokers/broker-
loggers/ips/cluster-links)
--force Suppress console prompts
--help Print usage information.
--ip <String> The IP address.
--ip-defaults The config defaults for all IPs.
--replica-placement <String: Replica This configuration is a JSON object
placement JSON file path.> that controls the set of brokers
(replicas) which will always be
allowed to join the ISR. And the set
of brokers (observers) which are not
allowed to join the ISR. The format
of JSON is:
{
"version": 1,
"replicas": [
{
"count": 2,
"constraints": {"rack": "east-1"}
},
{
"count": 1,
"constraints": {"rack": "east-2"}
}
],
"observers":[
{
"count": 1,
"constraints": {"rack": "west-1"}
}
]
}
--topic <String> The topic's name.
--user <String> The user's principal name.
--user-defaults The config defaults for all users.
--version Display Kafka version.
--zk-tls-config-file <String: Identifies the file where ZooKeeper
ZooKeeper TLS configuration> client TLS connectivity properties
are defined. Any properties other
than zookeeper.clientCnxnSocket,
zookeeper.ssl.cipher.suites,
zookeeper.ssl.client.enable,
zookeeper.ssl.crl.enable, zookeeper.
ssl.enabled.protocols, zookeeper.ssl.
endpoint.identification.algorithm,
zookeeper.ssl.keystore.location,
zookeeper.ssl.keystore.password,
zookeeper.ssl.keystore.type,
zookeeper.ssl.ocsp.enable, zookeeper.
ssl.protocol, zookeeper.ssl.
truststore.location, zookeeper.ssl.
truststore.password, zookeeper.ssl.
truststore.type are ignored.
--zookeeper <String: urls> DEPRECATED. The connection string for
the zookeeper connection in the form
host:port. Multiple URLS can be
given to allow fail-over. Required
when configuring SCRAM credentials
for users or dynamic broker configs
when the relevant broker(s) are
down. Not allowed otherwise.
kafka-get-offsets.sh¶
A tool to retrieve topic-partition offsets.
Usage details
An interactive shell for getting topic-partition offsets.
Option Description
------ -----------
--bootstrap-server <String: HOST1: REQUIRED. The server(s) to connect to
PORT1,...,HOST3:PORT3> in the form HOST1:PORT1,HOST2:PORT2.
--broker-list <String: HOST1:PORT1,..., DEPRECATED, use --bootstrap-server
HOST3:PORT3> instead; ignored if --bootstrap-
server is specified. The server(s)
to connect to in the form HOST1:
PORT1,HOST2:PORT2.
--command-config <String: config file> Property file containing configs to be
passed to Admin Client.
--exclude-internal-topics By default, internal topics are
included. If specified, internal
topics are excluded.
--partitions <String: partition ids> Comma separated list of partition ids
to get the offsets for. If not
present, all partitions of the
authorized topics are queried.
Cannot be used if --topic-partitions
is present.
--time <String: <timestamp> / -1 or timestamp of the offsets before that.
latest / -2 or earliest / -3 or max- [Note: No offset is returned, if the
timestamp> timestamp greater than recently
committed record timestamp is
given.] (default: latest)
--topic <String: topic> The topic to get the offsets for. It
also accepts a regular expression.
If not present, all authorized
topics are queried. Cannot be used
if --topic-partitions is present.
--topic-partitions <String: topic1:1, Comma separated list of topic-
topic2:0-3,topic3,topic4:5-,topic5:-3 partition patterns to get the
> offsets for, with the format of:
([^:,]*)(?::(?:([0-9]*)|(?:([0-9]*)
-([0-9]*))))?. The first group is
an optional regex for the topic
name, if omitted, it matches any
topic name. The section after ':'
describes a partition pattern,
which can be: a number, a range in
the format of NUMBER-NUMBER (lower
inclusive, upper exclusive), an
inclusive lower bound in the format
of NUMBER-, an exclusive upper
bound in the format of -NUMBER or
may be omitted to accept all
partitions.
kafka-leader-election.sh¶
This tool attempts to elect a new leader for a set of topic partitions.
Run this tool manually to restore leadership if the auto.leader.rebalance.enable property is set to false.
Usage details
This tool attempts to elect a new leader for a set of topic partitions. The type of elections supported are preferred replicas and unclean replicas.
Option Description
------ -----------
--admin.config <String: config file> Configuration properties files to pass
to the admin client
--all-topic-partitions Perform election on all of the
eligible topic partitions based on
the type of election (see the --
election-type flag). Not allowed if
--topic or --path-to-json-file is
specified.
--bootstrap-server <String: host:port> A hostname and port for the broker to
connect to, in the form host:port.
Multiple comma separated URLs can be
given. REQUIRED.
--election-type <[PREFERRED,UNCLEAN]: Type of election to attempt. Possible
election type> values are "preferred" for preferred
leader election or "unclean" for
unclean leader election. If
preferred election is selection, the
election is only performed if the
current leader is not the preferred
leader for the topic partition. If
unclean election is selected, the
election is only performed if there
are no leader for the topic
partition. REQUIRED.
--help Print usage information.
--partition <Integer: partition id> Partition id for which to perform an
election. REQUIRED if --topic is
specified.
--path-to-json-file <String: Path to The JSON file with the list of
JSON file> partition for which leader elections
should be performed. This is an
example format.
{"partitions":
[{"topic": "foo", "partition": 1},
{"topic": "foobar", "partition": 2}]
}
Not allowed if --all-topic-partitions
or --topic flags are specified.
--topic <String: topic name> Name of topic for which to perform an
election. Not allowed if --path-to-
json-file or --all-topic-partitions
is specified.
--version Display Kafka version.
kafka-transactions.sh¶
A tool to list and describe transactions. Use to detect and abort hanging transactions. See Detect and Abort Hanging Transcations
Usage details
usage: kafka-transactions.sh [-h] [-v] [--command-config FILE] --bootstrap-server host:port COMMAND ...
This tool is used to analyze the transactional state of producers in the cluster. It can be used to detect and
recover from hanging transactions.
optional arguments:
-h, --help show this help message and exit
-v, --version show the version of this Kafka distribution and exit
--command-config FILE property file containing configs to be passed to admin client
--bootstrap-server host:port
hostname and port for the broker to connect to, in the form `host:port`
(multiple comma-separated entries can be given)
commands:
list list transactions
describe describe the state of an active transactional-id
describe-producers describe the states of active producers for a topic partition
abort abort a hanging transaction (requires administrative privileges)
find-hanging find hanging transactions
kafka-reassign-partitions.sh¶
A tool to move topic partitions between replicas You pass a JSON-formatted file to specify the new replicas. To learn more, see Changing the replication factor in the Confluent docs.
Usage details
This tool helps to move topic partitions between replicas.
Option Description
------ -----------
--additional Execute this reassignment in addition
to any other ongoing ones. This
option can also be used to change
the throttle of an ongoing
reassignment.
--bootstrap-server <String: Server(s) REQUIRED: the server(s) to use for
to use for bootstrapping> bootstrapping.
--broker-list <String: brokerlist> The list of brokers to which the
partitions need to be reassigned in
the form "0,1,2". This is required
if --topics-to-move-json-file is
used to generate reassignment
configuration
--cancel Cancel an active reassignment.
--command-config <String: Admin client Property file containing configs to be
property file> passed to Admin Client.
--disable-rack-aware Disable rack aware replica assignment
--execute Kick off the reassignment as specified
by the --reassignment-json-file
option.
--generate Generate a candidate partition
reassignment configuration. Note
that this only generates a candidate
assignment, it does not execute it.
--help Print usage information.
--list List all active partition
reassignments.
--preserve-throttles Do not modify broker or topic
throttles.
--reassignment-json-file <String: The JSON file with the partition
manual assignment json file path> reassignment configurationThe format
to use is -
{"partitions":
[{"topic": "foo",
"partition": 1,
"replicas": [1,2,3,4],
"observers":[3,4],
"log_dirs": ["dir1","dir2","dir3","
dir4"] }],
"version":1
}
Note that "log_dirs" is optional. When
it is specified, its length must
equal the length of the replicas
list. The value in this list can be
either "any" or the absolution path
of the log directory on the broker.
If absolute log directory path is
specified, the replica will be moved
to the specified log directory on
the broker.
Note that "observers" is optional.
When it is specified it must be a
suffix of the replicas list.
--replica-alter-log-dirs-throttle The movement of replicas between log
<Long: replicaAlterLogDirsThrottle> directories on the same broker will
be throttled to this value
(bytes/sec). This option can be
included with --execute when a
reassignment is started, and it can
be altered by resubmitting the
current reassignment along with the
--additional flag. The throttle rate
should be at least 1 KB/s. (default:
-1)
--throttle <Long: throttle> The movement of partitions between
brokers will be throttled to this
value (bytes/sec). This option can
be included with --execute when a
reassignment is started, and it can
be altered by resubmitting the
current reassignment along with the
--additional flag. The throttle rate
should be at least 1 KB/s. (default:
-1)
--timeout <Long: timeout> The maximum time in ms to wait for log
directory replica assignment to
begin. (default: 10000)
--topics-to-move-json-file <String: Generate a reassignment configuration
topics to reassign json file path> to move the partitions of the
specified topics to the list of
brokers specified by the --broker-
list option. The format to use is -
{"topics":
[{"topic": "foo"},{"topic": "foo1"}],
"version":1
}
--verify Verify if the reassignment completed
as specified by the --reassignment-
json-file option. If there is a
throttle engaged for the replicas
specified, and the rebalance has
completed, the throttle will be
removed
--version Display Kafka version.
kafka-delete-records.sh¶
This tool deletes partition records. Use this if a topic receives bad data.
Pass a JSON-formatted file that specifies the topic, partition, and offset for data deletion. Data will be deleted up to the offset specified. Example:
./kafka-delete-records.sh --bootstrap-server <hostname:port> --offset-json-file deleteme.json
Usage details
This tool helps to delete records of the given partitions down to the specified offset.
Option Description
------ -----------
--bootstrap-server <String: server(s) REQUIRED: The server to connect to.
to use for bootstrapping>
--command-config <String: command A property file containing configs to
config property file path> be passed to Admin Client.
--help Print usage information.
--offset-json-file <String: Offset REQUIRED: The JSON file with offset
json file path> per partition. The format to use is:
{"partitions":
[{"topic": "foo", "partition": 1,
"offset": 1}],
"version":1
}
--version Display Kafka version.
kafka-log-dirs.sh¶
This tool gets a list of replicas per log directory on a broker.
Usage details
This tool helps to query log directory usage on the specified brokers.
------ -----------
--bootstrap-server <String: The server REQUIRED: the server(s) to use for
(s) to use for bootstrapping> bootstrapping
--broker-list <String: Broker list> The list of brokers to be queried in
the form "0,1,2". All brokers in the
cluster will be queried if no broker
list is specified
--command-config <String: Admin client Property file containing configs to be
property file> passed to Admin Client.
--describe Describe the specified log directories
on the specified brokers.
--help Print usage information.
--topic-list <String: Topic list> The list of topics to be queried in
the form "topic1,topic2,topic3". All
topics will be queried if no topic
list is specified (default: )
--version Display Kafka version.
kafka-replica-verification.sh¶
Used to verify that all replicas of a topic contain the same data.
Requires a broker-list parameter that contains a list of <hostname:port> entries specifying the server/port to connect to.
Usage details
Validate that all replicas for a set of topics have the same data.
Option Description
------ -----------
--broker-list <String: hostname: REQUIRED: The list of hostname and
port,...,hostname:port> port of the server to connect to.
--fetch-size <Integer: bytes> The fetch size of each request.
(default: 1048576)
--help Print usage information.
--max-wait-ms <Integer: ms> The max amount of time each fetch
request waits. (default: 1000)
--report-interval-ms <Long: ms> The reporting interval. (default:
30000)
--time <Long: timestamp/-1(latest)/-2 Timestamp for getting the initial
(earliest)> offsets. (default: -1)
--topic-white-list <String: Java regex DEPRECATED use --topics-include
(String)> instead; ignored if --topics-include
specified. List of topics to verify
replica consistency. Defaults to '.
*' (all topics) (default: .*)
--topics-include <String: Java regex List of topics to verify replica
(String)> consistency. Defaults to '.*' (all
topics) (default: .*)
--version Print version information and exit.
kafka-mirror-maker.sh¶
DEPRECATED: See connect-mirror-maker.sh. Enables the creation of a replica of an existing Kafka cluster.
Example: ./bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters secondary.
Learn more Kafka mirroring
Usage details
This tool helps to continuously copy data between two Kafka clusters.
Option Description
------ -----------
--abort.on.send.failure <String: Stop Configure the mirror maker to exit on
the entire mirror maker when a send a failed send. (default: true)
failure occurs>
--consumer.config <String: config file> Embedded consumer config for consuming
from the source cluster.
--consumer.rebalance.listener <String: The consumer rebalance listener to use
A custom rebalance listener of type for mirror maker consumer.
ConsumerRebalanceListener>
--help Print usage information.
--include <String: Java regex (String)> List of included topics to mirror.
--message.handler <String: A custom Message handler which will process
message handler of type every record in-between consumer and
MirrorMakerMessageHandler> producer.
--message.handler.args <String: Arguments used by custom message
Arguments passed to message handler handler for mirror maker.
constructor.>
--new.consumer DEPRECATED Use new consumer in mirror
maker (this is the default so this
option will be removed in a future
version).
--num.streams <Integer: Number of Number of consumption streams.
threads> (default: 1)
--offset.commit.interval.ms <Integer: Offset commit interval in ms.
offset commit interval in (default: 60000)
millisecond>
--producer.config <String: config file> Embedded producer config.
--rebalance.listener.args <String: Arguments used by custom rebalance
Arguments passed to custom rebalance listener for mirror maker consumer.
listener constructor as a string.>
--version Display Kafka version.
--whitelist <String: Java regex DEPRECATED, use --include instead;
(String)> ignored if --include specified. List
of included topics to mirror.
connect-mirror-maker.sh¶
Replicates topics from one cluster to another using the Connect framework.
You must pass an an mm2.properties MM2 configuration file. For more information, see
KIP-382: MirrorMaker 2.0
Usage details
usage: connect-mirror-maker [-h] [--clusters CLUSTER [CLUSTER ...]] mm2.properties
MirrorMaker 2.0 driver
positional arguments:
mm2.properties MM2 configuration file.
optional arguments:
-h, --help show this help message and exit
--clusters CLUSTER [CLUSTER ...]
Target cluster to use for this node.
Client, producer, and consumer tools¶
kafka-verifiable-consumer.sh¶
This tool consumes messages from a topic and emits consumer events as JSON objects to STDOUT. For example, group rebalances, received messages, and offsets committed. Intended for internal testing.
Usage details
usage: verifiable-consumer [-h] --topic TOPIC --group-id GROUP_ID
[--group-instance-id GROUP_INSTANCE_ID]
[--max-messages MAX-MESSAGES] [--session-timeout TIMEOUT_MS]
[--verbose] [--enable-autocommit] [--send-offset-for-times-data]
[--reset-policy RESETPOLICY]
[--assignment-strategy ASSIGNMENTSTRATEGY]
[--consumer.config CONFIG_FILE]
(--bootstrap-server HOST1:PORT1[,HOST2:PORT2[...]] |
--broker-list HOST1:PORT1[,HOST2:PORT2[...]])
This tool consumes messages from a specific topic and emits consumer events (e.g. group
rebalances, received messages, and offsets committed) as JSON objects to STDOUT.
optional arguments:
-h, --help show this help message and exit
--topic TOPIC Consumes messages from this topic.
--group-id GROUP_ID The groupId shared among members of the consumer group
--group-instance-id GROUP_INSTANCE_ID
A unique identifier of the consumer instance
--max-messages MAX-MESSAGES
Consume this many messages. If -1 (the default), the consumer will
consume until the process is killed externally (default: -1)
--session-timeout TIMEOUT_MS
Set the consumer's session timeout (default: 30000)
--verbose Enable to log individual consumed records (default: false)
--enable-autocommit Enable offset auto-commit on consumer (default: false)
--send-offset-for-times-data
Consumer sends offsetForTimes() information for all the partitions it
has subscribed to. Use when version = DEV_BRANCH (default: false)
--reset-policy RESETPOLICY
Set reset policy (must be either 'earliest', 'latest', or 'none'
(default: earliest)
--assignment-strategy ASSIGNMENTSTRATEGY
Set assignment strategy (e.g. org.apache.kafka.clients.consumer.
RoundRobinAssignor) (default: org.apache.kafka.clients.consumer.
RangeAssignor)
--consumer.config CONFIG_FILE
Consumer config properties file (config options shared with command
line parameters will be overridden).
Connection Group:
Group of arguments for connection to brokers
--bootstrap-server HOST1:PORT1[,HOST2:PORT2[...]]
REQUIRED unless --broker-list(deprecated) is specified. The server(s)
to connect to. Comma-separated list of Kafka brokers in the form
HOST1:PORT1,HOST2:PORT2,...
--broker-list HOST1:PORT1[,HOST2:PORT2[...]]
DEPRECATED, use --bootstrap-server instead; ignored if --bootstrap-
server is specified. Comma-separated list of Kafka brokers in the
form HOST1:PORT1,HOST2:PORT2,...
kafka-configs.sh¶
Use this tool to change and describe topic, client, user, broker, ip, or cluster link configuration settings.
To change a property, specify the entity-type to the desired entity (topic, broker, user, etc), and use the
alter option. See Topic Operations for example of how to modify topics.
Usage details
This tool helps to manipulate and describe entity config for a topic, client, user, broker, ip, or cluster link.
Option Description
------ -----------
--add-config <String> Key Value pairs of configs to add.
Square brackets can be used to group
values which contain commas: 'k1=v1,
k2=[v1,v2,v2],k3=v3'. The following
is a list of valid configurations:
For entity-type 'topics':
cleanup.policy
compression.type
confluent.append.record.interceptor.
classes
confluent.compacted.topic.prefer.tier.
fetch.ms
confluent.key.schema.validation
confluent.key.subject.name.strategy
confluent.placement.constraints
confluent.prefer.tier.fetch.ms
confluent.segment.speculative.
prefetch.enable
confluent.tier.cleaner.compact.min.
efficiency
confluent.tier.cleaner.compact.
segment.min.bytes
confluent.tier.cleaner.dual.compaction
confluent.tier.cleaner.enable
confluent.tier.cleaner.min.cleanable.
ratio
confluent.tier.enable
confluent.tier.local.hotset.bytes
confluent.tier.local.hotset.ms
confluent.tier.segment.hotset.roll.
min.bytes
confluent.value.schema.validation
confluent.value.subject.name.strategy
delete.retention.ms
file.delete.delay.ms
flush.messages
flush.ms
follower.replication.throttled.
replicas
index.interval.bytes
leader.replication.throttled.replicas
local.retention.bytes
local.retention.ms
max.compaction.lag.ms
max.message.bytes
message.downconversion.enable
message.format.version
message.timestamp.difference.max.ms
message.timestamp.type
min.cleanable.dirty.ratio
min.compaction.lag.ms
min.insync.replicas
preallocate
remote.storage.enable
retention.bytes
retention.ms
segment.bytes
segment.index.bytes
segment.jitter.ms
segment.ms
unclean.leader.election.enable
For entity-type 'brokers':
advertised.listeners
auto.create.topics.enable
background.threads
compression.type
confluent.append.record.interceptor.
classes
confluent.automatic.alter.leadership.
priority.retry.backoff.ms
confluent.backpressure.disk.enable
confluent.backpressure.disk.free.
threshold.bytes
confluent.backpressure.disk.produce.
bytes.per.second
confluent.backpressure.disk.threshold.
recovery.factor
confluent.backpressure.request.min.
broker.limit
confluent.backpressure.request.queue.
size.percentile
confluent.backpressure.types
confluent.balancer.enable
confluent.balancer.exclude.topic.names
confluent.balancer.exclude.topic.
prefixes
confluent.balancer.heal.uneven.load.
trigger
confluent.balancer.rebalancing.goals
confluent.balancer.throttle.bytes.per.
second
confluent.balancer.triggering.goals
confluent.broker.health.manager.
engine.request.handler.threads.stuck.
criteria
confluent.broker.health.manager.hard.
kill.duration.ms
confluent.broker.health.manager.
mitigation.enabled
confluent.broker.health.manager.num.
samples.before.broker.suspect
confluent.broker.health.manager.num.
samples.before.broker.unhealthy
confluent.broker.health.manager.
percentage.unhealthy.samples.before.
broker.suspect
confluent.broker.health.manager.
percentage.unhealthy.samples.before.
broker.unhealthy
confluent.broker.health.manager.
sample.duration.ms
confluent.broker.health.manager.
storage.background.threads.stuck.
criteria
confluent.broker.health.manager.
storage.network.threads.stuck.
criteria
confluent.broker.health.manager.
storage.request.handler.threads.
stuck.criteria
confluent.broker.load.average.service.
request.time.ms
confluent.broker.load.workload.
coefficient
confluent.clm.enabled
confluent.clm.frequency.in.hours
confluent.clm.max.backup.days
confluent.clm.min.delay.in.minutes
confluent.clm.thread.pool.size
confluent.clm.topic.retention.days.to.
backup.days
confluent.cluster.link.fetch.response.
min.bytes
confluent.cluster.link.fetch.response.
total.bytes
confluent.cluster.link.io.max.bytes.
per.second
confluent.cluster.link.replication.
quota.mode
confluent.compacted.topic.prefer.tier.
fetch.ms
confluent.durability.audit.checks
confluent.durability.events.allowed
confluent.eligible.controllers
confluent.fetch.partition.pruning.
enable
confluent.metadata.server.cluster.
registry.clusters
confluent.min.acks
confluent.network.health.manager.min.
healthy.network.samples
confluent.network.health.manager.
mitigation.enabled
confluent.network.health.manager.
network.sample.window.size
confluent.network.health.manager.
sample.duration.ms
confluent.plugins.cluster.link.policy.
max.destination.links.per.tenant
confluent.plugins.cluster.link.policy.
max.source.links.per.tenant
confluent.plugins.topic.policy.max.
partitions.per.cluster
confluent.plugins.topic.policy.max.
partitions.per.tenant
confluent.plugins.topic.policy.max.
topics.per.cluster
confluent.prefer.tier.fetch.ms
confluent.quota.dynamic.enable
confluent.quota.tenant.default.
controller.mutation.rate
confluent.quota.tenant.fetch.
multiplier
confluent.quota.tenant.produce.
multiplier
confluent.quota.tenant.user.quotas.
enable
confluent.reporters.telemetry.auto.
enable
confluent.security.event.router.config
confluent.segment.speculative.
prefetch.enable
confluent.telemetry.enabled
confluent.tier.cleaner.compact.min.
efficiency
confluent.tier.cleaner.compact.
segment.min.bytes
confluent.tier.cleaner.dedupe.buffer.
size
confluent.tier.cleaner.dual.compaction
confluent.tier.cleaner.enable
confluent.tier.cleaner.io.buffer.load.
factor
confluent.tier.cleaner.io.buffer.size
confluent.tier.cleaner.io.max.bytes.
per.second
confluent.tier.cleaner.min.cleanable.
ratio
confluent.tier.cleaner.num.threads
confluent.tier.enable
confluent.tier.fetcher.memorypool.
bytes
confluent.tier.local.hotset.bytes
confluent.tier.local.hotset.ms
confluent.tier.max.partition.fetch.
bytes.override
confluent.tier.metadata.snapshots.
enable
confluent.tier.metadata.snapshots.
interval.ms
confluent.tier.segment.hotset.roll.
min.bytes
confluent.tier.topic.delete.backoff.ms
confluent.tier.topic.delete.check.
interval.ms
confluent.tier.topic.delete.max.
inprogress.partitions
confluent.traffic.network.region.id
confluent.transaction.logging.
verbosity
connections.max.age.ms
fetch.max.bytes
floor.max.connection.creation.rate
follower.replication.throttled.rate
follower.replication.throttled.
replicas
leader.replication.throttled.rate
leader.replication.throttled.replicas
listener.security.protocol.map
listeners
log.cleaner.backoff.ms
log.cleaner.dedupe.buffer.size
log.cleaner.delete.retention.ms
log.cleaner.io.buffer.load.factor
log.cleaner.io.buffer.size
log.cleaner.io.max.bytes.per.second
log.cleaner.max.compaction.lag.ms
log.cleaner.min.cleanable.ratio
log.cleaner.min.compaction.lag.ms
log.cleaner.threads
log.cleanup.policy
log.deletion.max.segments.per.run
log.deletion.throttler.disk.free.
headroom.bytes
log.flush.interval.messages
log.flush.interval.ms
log.index.interval.bytes
log.index.size.max.bytes
log.message.downconversion.enable
log.message.timestamp.difference.max.
ms
log.message.timestamp.type
log.preallocate
log.retention.bytes
log.retention.ms
log.roll.jitter.ms
log.roll.ms
log.segment.bytes
log.segment.delete.delay.ms
max.connection.creation.rate
max.connection.creation.rate.per.ip.
enable.threshold
max.connections
max.connections.per.ip
max.connections.per.ip.overrides
message.max.bytes
metric.reporters
min.insync.replicas
num.io.threads
num.network.threads
num.partitions
num.recovery.threads.per.data.dir
num.replica.fetchers
principal.builder.class
quotas.expiration.time.ms
replica.alter.log.dirs.io.max.bytes.
per.second
sasl.enabled.mechanisms
sasl.jaas.config
sasl.kerberos.kinit.cmd
sasl.kerberos.min.time.before.relogin
sasl.kerberos.principal.to.local.rules
sasl.kerberos.service.name
sasl.kerberos.ticket.renew.jitter
sasl.kerberos.ticket.renew.window.
factor
sasl.login.refresh.buffer.seconds
sasl.login.refresh.min.period.seconds
sasl.login.refresh.window.factor
sasl.login.refresh.window.jitter
sasl.mechanism.inter.broker.protocol
socket.send.buffer.bytes
ssl.cipher.suites
ssl.client.auth
ssl.enabled.protocols
ssl.endpoint.identification.algorithm
ssl.engine.factory.class
ssl.key.password
ssl.keymanager.algorithm
ssl.keystore.certificate.chain
ssl.keystore.key
ssl.keystore.location
ssl.keystore.password
ssl.keystore.type
ssl.protocol
ssl.provider
ssl.secure.random.implementation
ssl.trustmanager.algorithm
ssl.truststore.certificates
ssl.truststore.location
ssl.truststore.password
ssl.truststore.type
unclean.leader.election.enable
For entity-type 'users':
SCRAM-SHA-256
SCRAM-SHA-512
consumer_byte_rate
controller_mutation_rate
producer_byte_rate
request_percentage
For entity-type 'clients':
consumer_byte_rate
controller_mutation_rate
producer_byte_rate
request_percentage
For entity-type 'ips':
connection_creation_rate
For entity-type 'cluster-links':
acl.filters
acl.sync.enable
acl.sync.ms
auto.create.mirror.topics.enable
auto.create.mirror.topics.filters
availability.check.consecutive.
failure.threshold
availability.check.ms
bootstrap.servers
client.dns.lookup
cluster.link.paused
cluster.link.prefix
cluster.link.retry.timeout.ms
confluent.replica.fetch.backoff.max.ms
connection.mode
connections.max.idle.ms
consumer.group.prefix.enable
consumer.offset.group.filters
consumer.offset.sync.enable
consumer.offset.sync.ms
link.fetcher.enforce.max.lagging.
partitions.ms
link.fetcher.flow.control
link.fetcher.lagging.partition.ms
link.fetcher.max.lagging.partitions
link.mode
local.listener.name
local.sasl.client.callback.handler.
class
local.sasl.jaas.config
local.sasl.kerberos.kinit.cmd
local.sasl.kerberos.min.time.before.
relogin
local.sasl.kerberos.service.name
local.sasl.kerberos.ticket.renew.
jitter
local.sasl.kerberos.ticket.renew.
window.factor
local.sasl.login.callback.handler.
class
local.sasl.login.class
local.sasl.login.connect.timeout.ms
local.sasl.login.read.timeout.ms
local.sasl.login.refresh.buffer.
seconds
local.sasl.login.refresh.min.period.
seconds
local.sasl.login.refresh.window.factor
local.sasl.login.refresh.window.jitter
local.sasl.login.retry.backoff.max.ms
local.sasl.login.retry.backoff.ms
local.sasl.mechanism
local.sasl.oauthbearer.clock.skew.
seconds
local.sasl.oauthbearer.expected.
audience
local.sasl.oauthbearer.expected.issuer
local.sasl.oauthbearer.jwks.endpoint.
refresh.ms
local.sasl.oauthbearer.jwks.endpoint.
retry.backoff.max.ms
local.sasl.oauthbearer.jwks.endpoint.
retry.backoff.ms
local.sasl.oauthbearer.jwks.endpoint.
url
local.sasl.oauthbearer.scope.claim.
name
local.sasl.oauthbearer.sub.claim.name
local.sasl.oauthbearer.token.endpoint.
url
local.security.protocol
local.ssl.cipher.suites
local.ssl.enabled.protocols
local.ssl.endpoint.identification.
algorithm
local.ssl.engine.factory.class
local.ssl.key.password
local.ssl.keymanager.algorithm
local.ssl.keystore.certificate.chain
local.ssl.keystore.key
local.ssl.keystore.location
local.ssl.keystore.password
local.ssl.keystore.type
local.ssl.protocol
local.ssl.provider
local.ssl.secure.random.implementation
local.ssl.trustmanager.algorithm
local.ssl.truststore.certificates
local.ssl.truststore.location
local.ssl.truststore.password
local.ssl.truststore.type
metadata.max.age.ms
num.cluster.link.fetchers
reconnect.backoff.max.ms
reconnect.backoff.ms
replica.fetch.backoff.ms
replica.fetch.max.bytes
replica.fetch.min.bytes
replica.fetch.response.max.bytes
replica.fetch.wait.max.ms
replica.socket.receive.buffer.bytes
replica.socket.timeout.ms
request.timeout.ms
retry.backoff.ms
reverse.connection.setup.timeout.ms
sasl.client.callback.handler.class
sasl.jaas.config
sasl.kerberos.kinit.cmd
sasl.kerberos.min.time.before.relogin
sasl.kerberos.service.name
sasl.kerberos.ticket.renew.jitter
sasl.kerberos.ticket.renew.window.
factor
sasl.login.callback.handler.class
sasl.login.class
sasl.login.connect.timeout.ms
sasl.login.read.timeout.ms
sasl.login.refresh.buffer.seconds
sasl.login.refresh.min.period.seconds
sasl.login.refresh.window.factor
sasl.login.refresh.window.jitter
sasl.login.retry.backoff.max.ms
sasl.login.retry.backoff.ms
sasl.mechanism
sasl.oauthbearer.clock.skew.seconds
sasl.oauthbearer.expected.audience
sasl.oauthbearer.expected.issuer
sasl.oauthbearer.jwks.endpoint.
refresh.ms
sasl.oauthbearer.jwks.endpoint.retry.
backoff.max.ms
sasl.oauthbearer.jwks.endpoint.retry.
backoff.ms
sasl.oauthbearer.jwks.endpoint.url
sasl.oauthbearer.scope.claim.name
sasl.oauthbearer.sub.claim.name
sasl.oauthbearer.token.endpoint.url
security.protocol
socket.connection.setup.timeout.max.ms
socket.connection.setup.timeout.ms
ssl.cipher.suites
ssl.enabled.protocols
ssl.endpoint.identification.algorithm
ssl.engine.factory.class
ssl.key.password
ssl.keymanager.algorithm
ssl.keystore.certificate.chain
ssl.keystore.key
ssl.keystore.location
ssl.keystore.password
ssl.keystore.type
ssl.protocol
ssl.provider
ssl.secure.random.implementation
ssl.trustmanager.algorithm
ssl.truststore.certificates
ssl.truststore.location
ssl.truststore.password
ssl.truststore.type
topic.config.sync.include
topic.config.sync.ms
Entity types 'users' and 'clients' may
be specified together to update
config for clients of a specific
user.
--add-config-file <String> Path to a properties file with configs
to add. See add-config for a list of
valid configurations.
--all List all configs for the given topic,
broker, or broker-logger entity
(includes static configuration when
the entity type is brokers)
--alter Alter the configuration for the entity.
--bootstrap-server <String: server to The Kafka server to connect to. This
connect to> is required for describing and
altering broker configs.
--broker <String> The broker's ID.
--broker-defaults The config defaults for all brokers.
--broker-logger <String> The broker's ID for its logger config.
--client <String> The client's ID.
--client-defaults The config defaults for all clients.
--cluster-link <String> The cluster link's name.
--command-config <String: command Property file containing configs to be
config property file> passed to Admin Client. This is used
only with --bootstrap-server option
for describing and altering broker
configs.
--delete-config <String> config keys to remove 'k1,k2'
--describe List configs for the given entity.
--entity-default Default entity name for
clients/users/brokers/ips (applies
to corresponding entity type in
command line)
--entity-name <String> Name of entity (topic name/client
id/user principal name/broker
id/ip/cluster link)
--entity-type <String> Type of entity
(topics/clients/users/brokers/broker-
loggers/ips/cluster-links)
--force Suppress console prompts
--help Print usage information.
--ip <String> The IP address.
--ip-defaults The config defaults for all IPs.
--replica-placement <String: Replica This configuration is a JSON object
placement JSON file path.> that controls the set of brokers
(replicas) which will always be
allowed to join the ISR. And the set
of brokers (observers) which are not
allowed to join the ISR. The format
of JSON is:
{
"version": 1,
"replicas": [
{
"count": 2,
"constraints": {"rack": "east-1"}
},
{
"count": 1,
"constraints": {"rack": "east-2"}
}
],
"observers":[
{
"count": 1,
"constraints": {"rack": "west-1"}
}
]
}
--topic <String> The topic's name.
--user <String> The user's principal name.
--user-defaults The config defaults for all users.
--version Display Kafka version.
--zk-tls-config-file <String: Identifies the file where ZooKeeper
ZooKeeper TLS configuration> client TLS connectivity properties
are defined. Any properties other
than zookeeper.clientCnxnSocket,
zookeeper.ssl.cipher.suites,
zookeeper.ssl.client.enable,
zookeeper.ssl.crl.enable, zookeeper.
ssl.enabled.protocols, zookeeper.ssl.
endpoint.identification.algorithm,
zookeeper.ssl.keystore.location,
zookeeper.ssl.keystore.password,
zookeeper.ssl.keystore.type,
zookeeper.ssl.ocsp.enable, zookeeper.
ssl.protocol, zookeeper.ssl.
truststore.location, zookeeper.ssl.
truststore.password, zookeeper.ssl.
truststore.type are ignored.
--zookeeper <String: urls> DEPRECATED. The connection string for
the zookeeper connection in the form
host:port. Multiple URLS can be
given to allow fail-over. Required
when configuring SCRAM credentials
for users or dynamic broker configs
when the relevant broker(s) are
down. Not allowed otherwise.
kafka-verifiable-producer.sh¶
This tool produces increasing integers to the specified topic and prints JSON metadata to STDOUT on each sendrequest. This tool shows which messages have been acked and which have not. Used for internal testing.
Usage details
usage: verifiable-producer [-h] --topic TOPIC [--max-messages MAX-MESSAGES]
[--throughput THROUGHPUT] [--acks ACKS]
[--producer.config CONFIG_FILE] [--message-create-time CREATETIME]
[--value-prefix VALUE-PREFIX] [--repeating-keys REPEATING-KEYS]
(--bootstrap-server HOST1:PORT1[,HOST2:PORT2[...]] |
--broker-list HOST1:PORT1[,HOST2:PORT2[...]])
This tool produces increasing integers to the specified topic and prints JSON metadata to
stdout on each "send" request, making externally visible which messages have been acked and
which have not.
optional arguments:
-h, --help show this help message and exit
--topic TOPIC Produce messages to this topic.
--max-messages MAX-MESSAGES
Produce this many messages. If -1, produce messages until the process
is killed externally. (default: -1)
--throughput THROUGHPUT
If set >= 0, throttle maximum message throughput to *approximately*
THROUGHPUT messages/sec. (default: -1)
--acks ACKS Acks required on each produced message. See Kafka docs on acks for
details. (default: -1)
--producer.config CONFIG_FILE
Producer config properties file.
--message-create-time CREATETIME
Send messages with creation time starting at the arguments value, in
milliseconds since epoch (default: -1)
--value-prefix VALUE-PREFIX
If specified, each produced value will have this prefix with a dot
separator
--repeating-keys REPEATING-KEYS
If specified, each produced record will have a key starting at 0
increment by 1 up to the number specified (exclusive), then the key
is set to 0 again
Connection Group:
Group of arguments for connection to brokers
--bootstrap-server HOST1:PORT1[,HOST2:PORT2[...]]
REQUIRED: The server(s) to connect to. Comma-separated list of Kafka
brokers in the form HOST1:PORT1,HOST2:PORT2,...
--broker-list HOST1:PORT1[,HOST2:PORT2[...]]
DEPRECATED, use --bootstrap-server instead; ignored if --bootstrap-
server is specified. Comma-separated list of Kafka brokers in the
form HOST1:PORT1,HOST2:PORT2,...
kafka-console-consumer.sh¶
A tool that enables you to consume records from a topic.
Requires a broker-list parameter that contains a list of <hostname:port> entries specifying the server/port to connect to.
Example: kafka-console-consumer.sh --bootstrap-server <BOOTSTRAP_BROKER_LIST> --consumer.config config.properties --topic <TOPIC_NAME> --property "print.key=true"
Usage details
This tool helps to read data from Kafka topics and outputs it to standard output.
Option Description
------ -----------
--bootstrap-server <String: server to REQUIRED: The server(s) to connect to.
connect to>
--consumer-property <String: A mechanism to pass user-defined
consumer_prop> properties in the form key=value to
the consumer.
--consumer.config <String: config file> Consumer config properties file. Note
that [consumer-property] takes
precedence over this config.
--enable-systest-events Log lifecycle events of the consumer
in addition to logging consumed
messages. (This is specific for
system tests.)
--formatter <String: class> The name of a class to use for
formatting kafka messages for
display. (default: kafka.tools.
DefaultMessageFormatter)
--from-beginning If the consumer does not already have
an established offset to consume
from, start with the earliest
message present in the log rather
than the latest message.
--group <String: consumer group id> The consumer group id of the consumer.
--help Print usage information.
--include <String: Java regex (String)> Regular expression specifying list of
topics to include for consumption.
--isolation-level <String> Set to read_committed in order to
filter out transactional messages
which are not committed. Set to
read_uncommitted to read all
messages. (default: read_uncommitted)
--key-deserializer <String:
deserializer for key>
--max-messages <Integer: num_messages> The maximum number of messages to
consume before exiting. If not set,
consumption is continual.
--offset <String: consume offset> The offset to consume from (a non-
negative number), or 'earliest'
which means from beginning, or
'latest' which means from end
(default: latest)
--partition <Integer: partition> The partition to consume from.
Consumption starts from the end of
the partition unless '--offset' is
specified.
--property <String: prop> The properties to initialize the
message formatter. Default
properties include:
print.timestamp=true|false
print.key=true|false
print.offset=true|false
print.partition=true|false
print.headers=true|false
print.value=true|false
key.separator=<key.separator>
line.separator=<line.separator>
headers.separator=<line.separator>
null.literal=<null.literal>
key.deserializer=<key.deserializer>
value.deserializer=<value.
deserializer>
header.deserializer=<header.
deserializer>
Users can also pass in customized
properties for their formatter; more
specifically, users can pass in
properties keyed with 'key.
deserializer.', 'value.
deserializer.' and 'headers.
deserializer.' prefixes to configure
their deserializers.
--skip-message-on-error If there is an error when processing a
message, skip it instead of halt.
--timeout-ms <Integer: timeout_ms> If specified, exit if no message is
available for consumption for the
specified interval.
--topic <String: topic> The topic to consume on.
--value-deserializer <String:
deserializer for values>
--version Display Kafka version.
--whitelist <String: Java regex DEPRECATED, use --include instead;
(String)> ignored if --include specified.
Regular expression specifying list
of topics to include for consumption.
kafka-console-producer.sh¶
Use to produce records to a topic
Requires a broker-list parameter that contains a list of <hostname:port> entries specifying the server/port to connect to.
Example kafka-console-producer.sh --broker-list <BOOTSTRAP_BROKER_LIST> --producer.config config.properties --topic <TOPIC_NAME> --property "parse.key=true" --property "key.separator=:"
Usage details
This tool helps to read data from standard input and publish it to Kafka.
Option Description
------ -----------
--batch-size <Integer: size> Number of messages to send in a single
batch if they are not being sent
synchronously. please note that this
option will be replaced if max-
partition-memory-bytes is also set
(default: 16384)
--bootstrap-server <String: server to REQUIRED unless --broker-list
connect to> (deprecated) is specified. The server
(s) to connect to. The broker list
string in the form HOST1:PORT1,HOST2:
PORT2.
--broker-list <String: broker-list> DEPRECATED, use --bootstrap-server
instead; ignored if --bootstrap-
server is specified. The broker
list string in the form HOST1:PORT1,
HOST2:PORT2.
--compression-codec [String: The compression codec: either 'none',
compression-codec] 'gzip', 'snappy', 'lz4', or 'zstd'.
If specified without value, then it
defaults to 'gzip'
--help Print usage information.
--line-reader <String: reader_class> The class name of the class to use for
reading lines from standard in. By
default each line is read as a
separate message. (default: kafka.
tools.
ConsoleProducer$LineMessageReader)
--max-block-ms <Long: max block on The max time that the producer will
send> block for during a send request.
(default: 60000)
--max-memory-bytes <Long: total memory The total memory used by the producer
in bytes> to buffer records waiting to be sent
to the server. This is the option to
control `buffer.memory` in producer
configs. (default: 33554432)
--max-partition-memory-bytes <Integer: The buffer size allocated for a
memory in bytes per partition> partition. When records are received
which are smaller than this size the
producer will attempt to
optimistically group them together
until this size is reached. This is
the option to control `batch.size`
in producer configs. (default: 16384)
--message-send-max-retries <Integer> Brokers can fail receiving the message
for multiple reasons, and being
unavailable transiently is just one
of them. This property specifies the
number of retries before the
producer give up and drop this
message. This is the option to
control `retries` in producer
configs. (default: 3)
--metadata-expiry-ms <Long: metadata The period of time in milliseconds
expiration interval> after which we force a refresh of
metadata even if we haven't seen any
leadership changes. This is the
option to control `metadata.max.age.
ms` in producer configs. (default:
300000)
--producer-property <String: A mechanism to pass user-defined
producer_prop> properties in the form key=value to
the producer.
--producer.config <String: config file> Producer config properties file. Note
that [producer-property] takes
precedence over this config.
--property <String: prop> A mechanism to pass user-defined
properties in the form key=value to
the message reader. This allows
custom configuration for a user-
defined message reader.
Default properties include:
parse.key=false
parse.headers=false
ignore.error=false
key.separator=\t
headers.delimiter=\t
headers.separator=,
headers.key.separator=:
null.marker= When set, any fields
(key, value and headers) equal to
this will be replaced by null
Default parsing pattern when:
parse.headers=true and parse.key=true:
"h1:v1,h2:v2...\tkey\tvalue"
parse.key=true:
"key\tvalue"
parse.headers=true:
"h1:v1,h2:v2...\tvalue"
--request-required-acks <String: The required `acks` of the producer
request required acks> requests (default: -1)
--request-timeout-ms <Integer: request The ack timeout of the producer
timeout ms> requests. Value must be non-negative
and non-zero. (default: 1500)
--retry-backoff-ms <Long> Before each retry, the producer
refreshes the metadata of relevant
topics. Since leader election takes
a bit of time, this property
specifies the amount of time that
the producer waits before refreshing
the metadata. This is the option to
control `retry.backoff.ms` in
producer configs. (default: 100)
--socket-buffer-size <Integer: size> The size of the tcp RECV size. This is
the option to control `send.buffer.
bytes` in producer configs.
(default: 102400)
--sync If set message send requests to the
brokers are synchronously, one at a
time as they arrive.
--timeout <Long: timeout_ms> If set and the producer is running in
asynchronous mode, this gives the
maximum amount of time a message
will queue awaiting sufficient batch
size. The value is given in ms. This
is the option to control `linger.ms`
in producer configs. (default: 1000)
--topic <String: topic> REQUIRED: The topic id to produce
messages to.
--version Display Kafka version.
kafka-producer-perf-test.sh¶
Enables you to produce a large quantity of data to test producer performance for the Kafka cluster.
Example: ./bin/kafka-producer-perf-test.sh --topic topic-a --num-records 200000 --record-size 1000 --throughput 10000000 --producer-props bootstrap.servers=<hostname:port>
Usage details
usage: producer-performance [-h] --topic TOPIC --num-records NUM-RECORDS
[--payload-delimiter PAYLOAD-DELIMITER] --throughput THROUGHPUT
[--producer-props PROP-NAME=PROP-VALUE [PROP-NAME=PROP-VALUE ...]]
[--producer.config CONFIG-FILE] [--print-metrics]
[--transactional-id TRANSACTIONAL-ID]
[--transaction-duration-ms TRANSACTION-DURATION]
(--record-size RECORD-SIZE | --payload-file PAYLOAD-FILE)
This tool is used to verify the producer performance.
optional arguments:
-h, --help show this help message and exit
--topic TOPIC produce messages to this topic
--num-records NUM-RECORDS
number of messages to produce
--payload-delimiter PAYLOAD-DELIMITER
provides delimiter to be used when --payload-file is provided.
Defaults to new line. Note that this parameter will be ignored if --
payload-file is not provided. (default: \n)
--throughput THROUGHPUT
throttle maximum message throughput to *approximately* THROUGHPUT
messages/sec. Set this to -1 to disable throttling.
--producer-props PROP-NAME=PROP-VALUE [PROP-NAME=PROP-VALUE ...]
kafka producer related configuration properties like bootstrap.
servers,client.id etc. These configs take precedence over those
passed via --producer.config.
--producer.config CONFIG-FILE
producer config properties file.
--print-metrics print out metrics at the end of the test. (default: false)
--transactional-id TRANSACTIONAL-ID
The transactionalId to use if transaction-duration-ms is > 0. Useful
when testing the performance of concurrent transactions. (default:
performance-producer-default-transactional-id)
--transaction-duration-ms TRANSACTION-DURATION
The max age of each transaction. The commitTransaction will be called
after this time has elapsed. Transactions are only enabled if this
value is positive. (default: 0)
either --record-size or --payload-file must be specified but not both.
--record-size RECORD-SIZE
message size in bytes. Note that you must provide exactly one of --
record-size or --payload-file.
--payload-file PAYLOAD-FILE
file to read the message payloads from. This works only for UTF-8
encoded text files. Payloads will be read from this file and a
payload will be randomly selected when sending messages. Note that
you must provide exactly one of --record-size or --payload-file.
kafka-consumer-groups.sh¶
This tool gets a list of the active groups in the cluster. For examples, see Manage Consumer Groups.
Usage details
Option Description
------ -----------
--all-groups Apply to all consumer groups.
--all-topics Consider all topics assigned to a
group in the `reset-offsets` process.
--bootstrap-server <String: server to REQUIRED: The server(s) to connect to.
connect to>
--by-duration <String: duration> Reset offsets to offset by duration
from current timestamp. Format:
'PnDTnHnMnS'
--command-config <String: command Property file containing configs to be
config property file> passed to Admin Client and Consumer.
--delete Pass in groups to delete topic
partition offsets and ownership
information over the entire consumer
group. For instance --group g1 --
group g2
--delete-offsets Delete offsets of consumer group.
Supports one consumer group at the
time, and multiple topics.
--describe Describe consumer group and list
offset lag (number of messages not
yet processed) related to given
group.
--dry-run Only show results without executing
changes on Consumer Groups.
Supported operations: reset-offsets.
--execute Execute operation. Supported
operations: reset-offsets.
--export Export operation execution to a CSV
file. Supported operations: reset-
offsets.
--from-file <String: path to CSV file> Reset offsets to values defined in CSV
file.
--group <String: consumer group> The consumer group we wish to act on.
--help Print usage information.
--list List all consumer groups.
--members Describe members of the group. This
option may be used with '--describe'
and '--bootstrap-server' options
only.
Example: --bootstrap-server localhost:
9092 --describe --group group1 --
members
--offsets Describe the group and list all topic
partitions in the group along with
their offset lag. This is the
default sub-action of and may be
used with '--describe' and '--
bootstrap-server' options only.
Example: --bootstrap-server localhost:
9092 --describe --group group1 --
offsets
--reset-offsets Reset offsets of consumer group.
Supports one consumer group at the
time, and instances should be
inactive
Has 2 execution options: --dry-run
(the default) to plan which offsets
to reset, and --execute to update
the offsets. Additionally, the --
export option is used to export the
results to a CSV format.
You must choose one of the following
reset specifications: --to-datetime,
--by-period, --to-earliest, --to-
latest, --shift-by, --from-file, --
to-current.
To define the scope use --all-topics
or --topic. One scope must be
specified unless you use '--from-
file'.
--shift-by <Long: number-of-offsets> Reset offsets shifting current offset
by 'n', where 'n' can be positive or
negative.
--state [String] When specified with '--describe',
includes the state of the group.
Example: --bootstrap-server localhost:
9092 --describe --group group1 --
state
When specified with '--list', it
displays the state of all groups. It
can also be used to list groups with
specific states.
Example: --bootstrap-server localhost:
9092 --list --state stable,empty
This option may be used with '--
describe', '--list' and '--bootstrap-
server' options only.
--timeout <Long: timeout (ms)> The timeout that can be set for some
use cases. For example, it can be
used when describing the group to
specify the maximum amount of time
in milliseconds to wait before the
group stabilizes (when the group is
just created, or is going through
some changes). (default: 5000)
--to-current Reset offsets to current offset.
--to-datetime <String: datetime> Reset offsets to offset from datetime.
Format: 'YYYY-MM-DDTHH:mm:SS.sss'
--to-earliest Reset offsets to earliest offset.
--to-latest Reset offsets to latest offset.
--to-offset <Long: offset> Reset offsets to a specific offset.
--topic <String: topic> The topic whose consumer group
information should be deleted or
topic whose should be included in
the reset offset process. In `reset-
offsets` case, partitions can be
specified using this format: `topic1:
0,1,2`, where 0,1,2 are the
partition to be included in the
process. Reset-offsets also supports
multiple topic inputs.
--verbose Provide additional information, if
any, when describing the group. This
option may be used with '--
offsets'/'--members'/'--state' and
'--bootstrap-server' options only.
Example: --bootstrap-server localhost:
9092 --describe --group group1 --
members --verbose
--version Display Kafka version.
kafka-consumer-perf-test.sh¶
This tool tests the consumer performance for the Kafka cluster.
Usage details
Option Description
------ -----------
--bootstrap-server <String: server to REQUIRED unless --broker-list
connect to> (deprecated) is specified. The server
(s) to connect to.
--broker-list <String: broker-list> DEPRECATED, use --bootstrap-server
instead; ignored if --bootstrap-
server is specified. The broker
list string in the form HOST1:PORT1,
HOST2:PORT2.
--consumer.config <String: config file> Consumer config properties file.
--date-format <String: date format> The date format to use for formatting
the time field. See java.text.
SimpleDateFormat for options.
(default: yyyy-MM-dd HH:mm:ss:SSS)
--fetch-size <Integer: size> The amount of data to fetch in a
single request. (default: 1048576)
--from-latest If the consumer does not already have
an established offset to consume
from, start with the latest message
present in the log rather than the
earliest message.
--group <String: gid> The group id to consume on. (default:
perf-consumer-50334)
--help Print usage information.
--hide-header If set, skips printing the header for
the stats
--messages <Long: count> REQUIRED: The number of messages to
send or consume
--num-fetch-threads <Integer: count> DEPRECATED AND IGNORED: Number of
fetcher threads. (default: 1)
--print-metrics Print out the metrics.
--reporting-interval <Integer: Interval in milliseconds at which to
interval_ms> print progress info. (default: 5000)
--show-detailed-stats If set, stats are reported for each
reporting interval as configured by
reporting-interval
--socket-buffer-size <Integer: size> The size of the tcp RECV size.
(default: 2097152)
--threads <Integer: count> DEPRECATED AND IGNORED: Number of
processing threads. (default: 10)
--timeout [Long: milliseconds] The maximum allowed time in
milliseconds between returned
records. (default: 10000)
--topic <String: topic> REQUIRED: The topic to consume from.
--version Display Kafka version.
Connect tools¶
connect-distributed.sh¶
Runs Connect workers in Distributed mode, meaning on multiple, distributed, machines. Distributed mode handles automatic balancing of work, allows you to scale up (or down) dynamically, and offers fault tolerance both in the active tasks and for configuration and offset commit data.
Usage details
USAGE: ./connect-distributed.sh [-daemon] connect-distributed.properties
connect-standalone.sh¶
Runs Connect in standalone mode meaning all work is performed in a single process. This is good for getting started but lacks fault tolerance. For more information, see Kafka Connect
Usage details
USAGE: ./connect-standalone.sh [-daemon] connect-standalone.properties
connect-mirror-maker.sh¶
Replicates topics from one cluster to another using the Connect framework.
You must pass an an mm2.properties MM2 configuration file. For more information, see
KIP-382: MirrorMaker 2.0
Usage details
usage: connect-mirror-maker [-h] [--clusters CLUSTER [CLUSTER ...]] mm2.properties
MirrorMaker 2.0 driver
positional arguments:
mm2.properties MM2 configuration file.
optional arguments:
-h, --help show this help message and exit
--clusters CLUSTER [CLUSTER ...]
Target cluster to use for this node.
Kafka Streams tools¶
kafka-streams-application-reset.sh¶
For Kafka Streams applications, this tool resets the application and forces it to reprocess its data from the beginning. Useful for debugging and testing. For more information, see Kafka Streams Application Reset Tool in the Confluent documentation.
Usage details
This tool helps to quickly reset an application in order to reprocess its data from scratch.
* This tool resets offsets of input topics to the earliest available offset (by default), or to a specific
defined position and it skips to the end of intermediate topics (topics that are input and output topics,
e.g., used by deprecated through() method).
* This tool deletes the internal topics that were created by Kafka Streams (topics starting with "<application.id>-").
The tool finds these internal topics automatically. If the topics flagged automatically for deletion by
the dry-run are unsuitable, you can specify a subset with the "--internal-topics" option.
* This tool will not delete output topics (if you want to delete them, you need to do it yourself with the bin/kafka-topics.sh command).
* This tool will not clean up the local state on the stream application instances (the persisted stores
used to cache aggregation results). You need to call KafkaStreams#cleanUp() in your application
or manually delete them from the directory specified by "state.dir"
configuration (${java.io.tmpdir}/kafka-streams/<application.id> by default).
* When long session timeout has been configured, active members could take longer to get expired on the
broker thus blocking the reset job to complete. Use the "--force" option could remove those
left-over members immediately. Make sure to stop all stream applications when this option
is specified to avoid unexpected disruptions.
*** Important! You will get wrong output if you don't clean up the local stores after running the reset tool!
*** Warning! This tool makes irreversible changes to your application. It is strongly recommended that you run this once with "--dry-run" to preview your changes before making them.
Option (* = required) Description
--------------------- -----------
* --application-id <String: id> The Kafka Streams application ID
(application.id).
--bootstrap-servers <String: urls> Comma-separated list of broker urls with
format: HOST1:PORT1,HOST2:PORT2
(default: localhost:9092)
--by-duration <String> Reset offsets to offset by duration from
current timestamp. Format: 'PnDTnHnMnS'
--config-file <String: file name> Property file containing configs to be
passed to admin clients and embedded
consumer.
--dry-run Display the actions that would be
performed without executing the reset
commands.
--force Force the removal of members of the
consumer group (intended to remove
stopped members if a long session
timeout was used). Make sure to shut
down all stream applications when this
option is specified to avoid unexpected
rebalances.
--from-file <String> Reset offsets to values defined in CSV
file.
--help Print usage information.
--input-topics <String: list> Comma-separated list of user input
topics. For these topics, the tool by
default will reset the offset to the
earliest available offset. Reset to
other offset position by appending
other reset offset option, ex: --input-
topics foo --shift-by 5
--intermediate-topics <String: list> Comma-separated list of intermediate user
topics (topics that are input and
output topics, e.g., used in the
deprecated through() method). For these
topics, the tool will skip to the end.
--internal-topics <String: list> Comma-separated list of internal topics
to delete. Must be a subset of the
internal topics marked for deletion by
the default behaviour (do a dry-run
without this option to view these
topics).
--shift-by <Long: number-of-offsets> Reset offsets shifting current offset by
'n', where 'n' can be positive or
negative
--to-datetime <String> Reset offsets to offset from datetime.
Format: 'YYYY-MM-DDTHH:mm:SS.sss'
--to-earliest Reset offsets to earliest offset.
--to-latest Reset offsets to latest offset.
--to-offset <Long> Reset offsets to a specific offset.
--version Print version information and exit.
Security tools¶
kafka-acls¶
Tool for adding, removing, and listing ACLs ACL Command-line interface
Usage details
This tool helps to manage acls on kafka.
Option Description
------ -----------
--add Indicates you are trying to add ACLs.
--allow-host <String: allow-host> Host from which principals listed in --
allow-principal will have access. If
you have specified --allow-principal
then the default for this option
will be set to '*' which allows
access from all hosts.
--allow-principal <String: allow- principal is in principalType:name
principal> format. Note that principalType must
be supported by the Authorizer being
used. For example, User:'*' is the
wild card indicating all users.
--authorizer <String: authorizer> DEPRECATED: Fully qualified class name
of the authorizer, which defaults to
kafka.security.authorizer.
AclAuthorizer if --bootstrap-server
is not provided. Warning: support
for ACL configuration directly
through the authorizer is deprecated
and will be removed in a future
release. Please use --bootstrap-
server instead to set ACLs through
the admin client.
--authorizer-properties <String: DEPRECATED: The properties required to
authorizer-properties> configure an instance of the
Authorizer specified by --
authorizer. These are key=val pairs.
For the default authorizer, example
values are: zookeeper.
connect=localhost:2181. Warning:
support for ACL configuration
directly through the authorizer is
deprecated and will be removed in a
future release. Please use --
bootstrap-server instead to set ACLs
through the admin client.
--bootstrap-server <String: server to A list of host/port pairs to use for
connect to> establishing the connection to the
Kafka cluster. This list should be
in the form host1:port1,host2:
port2,... This config is required
for acl management using admin
client API.
--cluster Add/Remove cluster ACLs.
--command-config [String: command- A property file containing configs to
config] be passed to Admin Client.
--consumer Convenience option to add/remove ACLs
for consumer role. This will
generate ACLs that allows READ,
DESCRIBE on topic and READ on group.
--delegation-token <String: delegation- Delegation token to which ACLs should
token> be added or removed. A value of '*'
indicates ACL should apply to all
tokens.
--deny-host <String: deny-host> Host from which principals listed in --
deny-principal will be denied
access. If you have specified --deny-
principal then the default for this
option will be set to '*' which
denies access from all hosts.
--deny-principal <String: deny- principal is in principalType:name
principal> format. By default anyone not added
through --allow-principal is denied
access. You only need to use this
option as negation to already
allowed set. Note that principalType
must be supported by the Authorizer
being used. For example if you
wanted to allow access to all users
in the system but not test-user you
can define an ACL that allows access
to User:'*' and specify --deny-
principal=User:test@EXAMPLE.COM. AND
PLEASE REMEMBER DENY RULES TAKES
PRECEDENCE OVER ALLOW RULES.
--force Assume Yes to all queries and do not
prompt.
--group <String: group> Consumer Group to which the ACLs
should be added or removed. A value
of '*' indicates the ACLs should
apply to all groups.
--help Print usage information.
--idempotent Enable idempotence for the producer.
This should be used in combination
with the --producer option. Note
that idempotence is enabled
automatically if the producer is
authorized to a particular
transactional-id.
--link-id <String: link id> Cluster link ID associated with the
ACLs for add/remove/list.
--list List ACLs for the specified resource,
use --topic <topic> or --group
<group> or --cluster to specify a
resource.
--operation <String> Operation that is being allowed or
denied. Valid operation names are:
Describe
DescribeConfigs
Alter
IdempotentWrite
Read
Delete
Create
ClusterAction
All
Write
AlterConfigs
CreateTokens
DescribeTokens
(default: All)
--principal [String: principal] List ACLs for the specified principal.
principal is in principalType:name
format. Note that principalType must
be supported by the Authorizer being
used. Multiple --principal option
can be passed.
--producer Convenience option to add/remove ACLs
for producer role. This will
generate ACLs that allows WRITE,
DESCRIBE and CREATE on topic.
--remove Indicates you are trying to remove
ACLs.
--resource-id List the ACLs with principal as
resourceID. This is only applicable
for Confluent Cloud.
--resource-pattern-type The type of the resource pattern or
<ANY|MATCH|LITERAL|PREFIXED|CONFLUENT_ALL_TENANT_LITERAL|CONFLUENT_ALL_TENANT_PREFIXED|CONFLUENT_ALL_TENANT_ANY|CONFLUENT_ONLY_TENANT_MATCH> pattern filter. When adding acls,
this should be a specific pattern
type, e.g. 'literal' or 'prefixed'.
When listing or removing acls, a
specific pattern type can be used to
list or remove acls from specific
resource patterns, or use the filter
values of 'any' or 'match', where
'any' will match any pattern type,
but will match the resource name
exactly, where as 'match' will
perform pattern matching to list or
remove all acls that affect the
supplied resource(s). WARNING:
'match', when used in combination
with the '--remove' switch, should
be used with care. (default: LITERAL)
--topic <String: topic> topic to which ACLs should be added or
removed. A value of '*' indicates
ACL should apply to all topics.
--transactional-id <String: The transactionalId to which ACLs
transactional-id> should be added or removed. A value
of '*' indicates the ACLs should
apply to all transactionalIds.
--user-principal <String: user- Specifies a user principal as a
principal> resource in relation with the
operation. For instance one could
grant CreateTokens or DescribeTokens
permission on a given user principal.
--version Display Kafka version.
--zk-tls-config-file <String: DEPRECATED: Identifies the file where
Authorizer ZooKeeper TLS ZooKeeper client TLS connectivity
configuration> properties are defined for the
default authorizer kafka.security.
authorizer.AclAuthorizer. Any
properties other than the following
(with or without an "authorizer."
prefix) are ignored: zookeeper.
clientCnxnSocket, zookeeper.ssl.
cipher.suites, zookeeper.ssl.client.
enable, zookeeper.ssl.crl.enable,
zookeeper.ssl.enabled.protocols,
zookeeper.ssl.endpoint.
identification.algorithm, zookeeper.
ssl.keystore.location, zookeeper.ssl.
keystore.password, zookeeper.ssl.
keystore.type, zookeeper.ssl.ocsp.
enable, zookeeper.ssl.protocol,
zookeeper.ssl.truststore.location,
zookeeper.ssl.truststore.password,
zookeeper.ssl.truststore.type. Note
that if SASL is not configured and
zookeeper.set.acl is supposed to be
true due to mutual certificate
authentication being used then it is
necessary to explicitly specify --
authorizer-properties zookeeper.set.
acl=true. Warning: support for ACL
configuration directly through the
authorizer is deprecated and will be
removed in a future release. Please
use --bootstrap-server instead to
set ACLs through the admin client.
kafka-delegation-tokens.sh¶
This tool creates, renews, expires, and describes delegation tokens. For more information, see Authentication using Delegation Tokens. See Authentication using Delegation Tokens in the Confluent Documentation.
Usage details
This tool helps to create, renew, expire, or describe delegation tokens.
Option Description
------ -----------
--bootstrap-server <String> REQUIRED: server(s) to use for bootstrapping.
--command-config <String> REQUIRED: A property file containing configs to
be passed to Admin Client. Token management
operations are allowed in secure mode only.
This config file is used to pass security
related configs.
--create Create a new delegation token. Use --renewer-
principal option to pass renewers principals.
--describe Describe delegation tokens for the given
principals. Use --owner-principal to pass
owner/renewer principals. If --owner-principal
option is not supplied, all the user owned
tokens and tokens where user have Describe
permission will be returned.
--expire Expire delegation token. Use --expiry-time-
period option to expire the token.
--expiry-time-period [Long] Expiry time period in milliseconds. If the value
is -1, then the token will get invalidated
immediately.
--help Print usage information.
--hmac [String] HMAC of the delegation token
--max-life-time-period [Long] Max life period for the token in milliseconds.
If the value is -1, then token max life time
will default to a server side config value
(delegation.token.max.lifetime.ms).
--owner-principal [String] owner is a kafka principal. It is should be in
principalType:name format.
--renew Renew delegation token. Use --renew-time-period
option to set renew time period.
--renew-time-period [Long] Renew time period in milliseconds. If the value
is -1, then the renew time period will default
to a server side config value (delegation.
token.expiry.time.ms).
--renewer-principal [String] renewer is a kafka principal. It is should be in
principalType:name format.
--version Display Kafka version.
zookeeper-security-migration.sh¶
This tool is used to restrict or provide access to ZooKeeper metadata. The tool updates the ACLs of znodes.
Usage details
ZooKeeper Migration Tool Help. This tool updates the ACLs of znodes as part of the process of setting up ZooKeeper authentication.
Option Description
------ -----------
--enable.path.check Checks if all the root paths exist in
ZooKeeper before migration. If not,
exit the command.
--help Print usage information.
--version Display Kafka version.
--zk-tls-config-file <String: Identifies the file where ZooKeeper
ZooKeeper TLS configuration> client TLS connectivity properties
are defined. Any properties other
than zookeeper.ssl.keystore.
password, zookeeper.ssl.truststore.
location, zookeeper.ssl.truststore.
type, zookeeper.ssl.ocsp.enable,
zookeeper.ssl.crl.enable, zookeeper.
ssl.truststore.password, zookeeper.
ssl.protocol, zookeeper.ssl.keystore.
type, zookeeper.ssl.client.enable,
zookeeper.ssl.cipher.suites,
zookeeper.clientCnxnSocket,
zookeeper.ssl.endpoint.
identification.algorithm, zookeeper.
ssl.enabled.protocols, zookeeper.ssl.
keystore.location are ignored.
--zookeeper.acl <String> Indicates whether to make the Kafka
znodes in ZooKeeper secure or
unsecure. The options are 'secure'
and 'unsecure'
--zookeeper.connect <String> Sets the ZooKeeper connect string
(ensemble). This parameter takes a
comma-separated list of host:port
pairs. (default: localhost:2181)
--zookeeper.connection.timeout Sets the ZooKeeper connection timeout.
<Integer> (default: 30000)
--zookeeper.session.timeout <Integer> Sets the ZooKeeper session timeout.
(default: 30000)
Addition testing and miscellaneous tools¶
This section contains tools you can use for testing and troubleshooting your applications.
kafka-dump-log.sh¶
This tool helps to parse a log file and dump its contents to the console; useful for debugging a corrupt log segment.
Requires a comma separated list of data and index log files to dump. Example ./kafka-log-dump.sh --files file1, file2
Usage details
This tool helps to parse a log file and dump its contents to the console, useful for debugging a seemingly corrupt log segment.
Option Description
------ -----------
--cluster-metadata-decoder if set, log data will be parsed as cluster
metadata records.
--deep-iteration if set, uses deep instead of shallow
iteration. Automatically set if print-
data-log is enabled.
--files <String: file1, file2, ...> REQUIRED: The comma separated list of
files to be dumped. The supported file
types include log files, index files,
time index files, transaction index
files, producer snapshots and tier state
files.
--help Print usage information.
--index-sanity-check if set, just checks the index sanity
without printing its content. This is
the same check that is executed on
broker startup to determine if an index
needs rebuilding or not.
--key-decoder-class [String] if set, used to deserialize the keys. This
class should implement kafka.serializer.
Decoder trait. Custom jar should be
available in kafka/libs directory.
(default: kafka.serializer.StringDecoder)
--max-bytes <Integer: size> Limit the amount of total batches read in
bytes avoiding reading the whole .log
file(s). (default: 2147483647)
--max-message-size <Integer: size> Size of largest message. (default: 5242880)
--offsets-decoder if set, log data will be parsed as offset
data from the __consumer_offsets topic.
--print-data-log if set, printing the messages content when
dumping data logs. Automatically set if
any decoder option is specified.
--quotas-decoder if set, log data will be parsed as quotas
data from the _confluent-quotas topic
--skip-record-metadata whether to skip printing metadata for each
record.
--transaction-log-decoder if set, log data will be parsed as
transaction metadata from the
__transaction_state topic.
--value-decoder-class [String] if set, used to deserialize the messages.
This class should implement kafka.
serializer.Decoder trait. Custom jar
should be available in kafka/libs
directory. (default: kafka.serializer.
StringDecoder)
--verify-index-only if set, just verify the index log without
printing its content.
--version Display Kafka version.
trogdor.sh¶
Trogdor is a test framework for Kafka. Trogdor can run benchmarks and other workloads. Trogdor can also inject faults in order to stress test the system. For more information, see Trogdor and TROGDOR
Usage details
The Trogdor fault injector.
Usage:
./trogdor [action] [options]
Actions:
agent: Run the trogdor agent.
coordinator: Run the trogdor coordinator.
client: Run the client which communicates with the trogdor coordinator.
agent-client: Run the client which communicates with the trogdor agent.
help: This help message.
kafka-run-class.sh¶
This tool is a thin wrapper around the Kafka Java class. It is called by other tools, and should not be run or modified directly.
Usage details
USAGE: ./kafka-run-class [-daemon] [-name servicename] [-loggc] classname [opts]