CLI (Command Line Interface) 환경에서 kafka 명령어와 동작 방식에 대해 학습해보도록 하겠습니다.
가장 먼저 Topic을 생성, 삭제, 변경, 설명(출력) 해보는 명령어를 학습해보도록 하겠습니다.
kafka의 Topic을 제어할 수 있는 파일은 kafka-topic.(sh | bat) 입니다.
확장자가 2개인 이유는 ubuntu의 경우는 sh, window의 경우는 bat파일이기 때문입니다.
Kafka에서 Topic 제어하기
Ubuntu: kafka-topics.sh Window: kafka-topics.bat
각각 환경에 따라 실행 파일이 다르지만, 실행 파일의 경로 또한 다릅니다.
ubuntu의 경우 /bin 폴더에 존재하지만, window의 경우에는 /bin/windows/ 폴더에 bat 파일이 존재합니다.
좌측: linux 우측: window
kafka-topics를 실행할 때 여러 옵션을 주어야 하지만, 어떤 옵션이 있는지 모르기 때문에 먼저 옵션을 확인합니다.
// ubuntu (linux)
# cd C:\dev\kafka\kafka_2.12-3.3.1\bin
# kafka-topics.sh
// window
# cd C:\dev\kafka\kafka_2.12-3.3.1\bin\windows
# kafka-topics.bat
kafka-topics를 실행하면 아래와 같이 해당 파일의 설명과 함께 옵션 정보가 출력이 됩니다.
파일 설명에 따르면 kafka-topics를 통해 topic을 생성/삭제/설명(출력) 또는 변경한다고 합니다.
이제 각 옵션을 하나씩 확인해보도록 하겠습니다.
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
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.
--list List all available topics.
--partitions <Integer: # of 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 , ...>
--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.
--bootstrap-server REQUIRED: The Kafka server to connect to
→ 연결할 kafka 서버를 작성하며 이 옵션은 필수입니다.
kafka-topics.bat --list
Exception in thread "main" java.lang.IllegalArgumentException: --bootstrap-server must be specified at kafka.admin.TopicCommand$TopicCommandOptions.checkArgs(TopicCommand.scala:619) at kafka.admin.TopicCommand$.main(TopicCommand.scala:48) at kafka.admin.TopicCommand.main(TopicCommand.scala)
--command-config Property file containing configs to be passed to Admin Client. This is used only with --bootstrap-server option for describing and altering broker configs.
→ 관리자 클라이언트에게 전달할 구성이 포함된 속성 파일입니다. → 브로커 구성을 설명하고 변경하기 위해 --bootstrap-server 옵션과 함께만 사용됩니다.
--config A topic configuration override for the topic being created or altered. The following is a list of valid configurations: cleanup.policy compression.type 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).
→ 생성 또는 변경 중인 Topic에 대한 Topic Config 재정의입니다 → 나열한 항목들을 사용할 수 있으며, --bootstap-server 옵션이 사용되는 경우 --create와 함께 사용하는 경우에만 지원됩니다.
자세한 내용은 공식 문서를 참고하시면 되며, 부담스러우신 분들은 공식 문서 내용을 번역해주신 토리맘님의 링크를 참고하시면 좋을 것 같습니다.
--delete-config 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.
kafka-topics.bat --bootstrap-server localhost:9092 --delete --topic test2 Error while executing topic command : Topic 'test2' does not exist as expected [2023-03-28 09:16:51,850] ERROR java.lang.IllegalArgumentException: Topic 'test2' does not exist as expected at kafka.admin.TopicCommand$.kafka$admin$TopicCommand$$ensureTopicExists(TopicCommand.scala:401) at kafka.admin.TopicCommand$TopicService.deleteTopic(TopicCommand.scala:361) at kafka.admin.TopicCommand$.main(TopicCommand.scala:63) at kafka.admin.TopicCommand.main(TopicCommand.scala) (kafka.admin.TopicCommand$)
현재 test Topic만 있는 상황에서 없는 Topic을 지우려 할때, --if-exists 옵션 없이 실행하면 Topic이 존재하지 않기 때문에 에러가 발생하지만, --if-exists 옵션을 설정하게 되면 Topic이 없기 때문에 해당 명령어를 실행하지 않아 에러가 발생하지 않습니다.
현재 test Topic이 존재하는 상황에서 다시 test로 Topic을 생성하려 할 때 --if-not-exists 옵션을 설정하게 되면 test Topic이 이미 있기 때문에 해당 명령어를 실행하지 않아 에러가 발생하지 않지만, --if-not-exists 옵션을 설정하지 않을 경우 그대로 명령어를 실행하여 에러가 발생합니다.
--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.
→ topic을 생성/수정할 때 partition 개수를 설정합니다. → WARNING: key가 있는 topic의 partition 개수를 늘리면 메세지의 동작 또는 순서에 영향을 받을 것이라고 합니다. → --partitions 옵션을 설정하지 않으면 cluster의 기본값으로 설정한다고 합니다.
kafka-topics.bat --bootstrap-server localhost:9092 --create --topic test Created topic test
cluster의 default partitions를 확인하려면 window는 config/server.properties, ubuntu는 /etc/kafka/server.properties에서 확인할 수 있습니다.
# The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across # the brokers. num.partitions=1
--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 입니다. → --create 옵션을 제외하고 정규 표현식도 허용합니다. → Topic 이름을 ""로 묶고 \접두사를 사용하여 정규식 기호를 escape 처리합니다.
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both. Error while executing topic command : Topic name "test\..topic" is illegal, it contains a character other than ASCII alphanumerics, '.', '_' and '-' [2023-03-28 08:49:20,860] ERROR org.apache.kafka.common.errors.InvalidTopicException: Topic name "test\..topic" is illegal, it contains a character other than ASCII alphanumerics, '.', '_' and '-' (kafka.admin.TopicCommand$)
Topic: test TopicId: xvA3RH4GQACOJYfUQP1Wag PartitionCount: 1 ReplicationFactor: 1 Configs: Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
하지만 topic-id를 설정하지 않아도 --describe를 통해 topic 정보를 출력해보면 topic-id가 설정되어 있습니다.
확인해보니, topic-id를 설정하지 않을 경우 임의의 UUID로 topic-id를 설정해준다고 합니다.
The controller will maintain local in-memory state containing a mapping from topic name to topic ID. On controller startup, the topic ID will automatically be loaded into memory along with the topics and partitions. A random UUID will be generated on topic creation or on migration of an existing topic without topic IDs.
댓글