KCraft kullanarak çok düğümlü bir Kafka kümesi nasıl kurulur?

giriiş

Apache Kafka, gerçek zamanlı veri akışlarını işlemek için Java dilinde yazılmış, açık kaynaklı, dağıtılmış bir olay ve akış işleme platformudur. Yüksek verimlilik ve kullanılabilirliğe sahip, doğası gereği ölçeklenebilir bir yapıdadır. Küme başına yüzlerce düğümü destekleyecek şekilde tasarlanmıştır.

Bu eğitimde, KRaft konsensüs protokolünü kullanan bir Kafka kümesi oluşturacaksınız. Düğümleri bir kümenin parçası olacak şekilde nasıl yapılandıracağınızı ve konu bölümlerinin farklı düğümlere nasıl atandığını öğreneceksiniz. Ayrıca, kümedeki belirli aracılara nasıl konu atayacağınızı da öğreneceksiniz.

Ön koşullar

• En az 4 GB RAM ve 2 CPU'ya sahip üç damla
• Üç drop'a işaret eden üç alt domain'e sahip tam kayıtlı bir alan adı.
• Apache Kafka Droplet’lerinize kuruldu ve yapılandırıldı.

Adım 1 – Kafka Düğümlerini Yapılandırın

Bu adımda, ön koşullar kapsamında oluşturduğunuz üç Kafka sunucusunu aynı KRaft kümesinin parçası olacak şekilde yapılandıracaksınız. KRaft ile düğümler, ek yük ve Apache ZooKeeper'a bağımlılık olmadan yönetim görevlerini organize edebilir ve gerçekleştirebilir.

İlk düğüm yapılandırması

İlk düğümü yapılandırarak başlayacaksınız. İlk olarak, aşağıdaki komutu çalıştırarak ilk Droplet'teki hizmeti durdurun:

sudo systemctl stop kafka

Kafka kullanıcısı olarak, Kafka'nın bulunduğu dizine gidin ve aşağıdaki komutu çalıştırarak yapılandırma dosyasını düzenlemeye açın:

vi /config/kraft/server.properties

Aşağıdaki satırları bulun:

...
# The role of this server. Setting this puts us in KRaft mode
process.roles=broker,controller
# The node id associated with this instance's roles
node.id=1
# The connect string for the controller quorum
controller.quorum.voters=1@localhost:9092
...

Bu üç parametre, Kafka düğümünü hem aracı hem de denetleyici olarak çalışacak şekilde yapılandırır; yani veri alıp tüketir (aracı) ve yönetim görevlerini gerçekleştirir (denetleyici). Bu ayrım, denetleyicilerin daha fazla verimlilik ve yedeklilik için ayrı tutulabildiği büyük dağıtımlarda kullanışlıdır.

node.id, kümedeki düğüm kimliğini belirtir. Bu ilk düğüm olduğundan, 1 olarak kalmalıdır. Tüm düğümlerin benzersiz bir düğüm kimliği olmalıdır, bu nedenle ikinci ve üçüncü düğümlerin kimlikleri sırasıyla 2 ve 3 olacaktır.

controller.quorum.voters, iletişim için düğüm kimliklerini ilgili adres ve portlara eşler. Bu, tüm küme düğümlerinin adreslerini belirttiğiniz yerdir, böylece her düğüm diğer düğümlerden haberdar olur. Satırı şu şekilde değiştirin:

...
[email protected]_domain:9093,[email protected]_domain:9093,[email protected]_domain:9093
...

Burada, kümedeki üç düğümü ilgili kimlikleriyle listeleyeceksiniz. your_domain ifadesini, ön koşullar sırasında belirlediğiniz alan adı adresinizle değiştirmeyi unutmayın.

Daha sonra dosyada şu satırları bulun:

...
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
# Name of listener used for communication between brokers.
inter.broker.listener.name=PLAINTEXT
# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for "listeners".
advertised.listeners=PLAINTEXT://localhost:9092
...

listeners, Kafka düğümünün dinlediği adresleri tanımlarken, advertised.listeners, istemcilere düğüme bağlanmaları için gönderilen adresleri belirtir. Bu, istemcilerin kullanması gereken gerçek adreslerin bir alt kümesini belirtmenize olanak tanır.

Satırları şu şekilde değiştirin, your_domain kısmını gerçek alan adınızla değiştirin:

...
listeners=PLAINTEXT://kafka1.your_domain:9092,CONTROLLER://kafka1.your_domain:9093
# Name of listener used for communication between brokers.
inter.broker.listener.name=PLAINTEXT
# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for "listeners".
advertised.listeners=PLAINTEXT://kafka1.your_domain:9092
...

Bu düğüm bir kümede olacağından, çalışacağı droplet'e adresleri açıkça işaret ettiniz.

Daha sonra num.partitions ayarını bulun:

...
# 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
...

Yorumda belirtildiği gibi, bu, her yeni iş parçacığı için yapılandırılan varsayılan bölüm sayısıdır. Üç düğümünüz olduğundan, bunu ikinin katına ayarlayın:

...
num.partitions=6
...

Buradaki 6 değeri, her düğümün varsayılan olarak iki konu bölümünü korumasını sağlar.

Ardından, tüketici ofsetlerini ve işlem durumunu koruyan dahili iş parçacıkları için yineleme faktörünü yapılandırın. Aşağıdaki satırları bulun:

...
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
...

Bunları aşağıdaki değerlere ayarlayın:

...
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2
...

Burada, en az iki düğümün dahili meta veriler açısından senkronize olması gerektiğini belirtiyorsunuz. İşiniz bittiğinde dosyayı kaydedip kapatın.

Varsayılan bölüm numarasını ayarladıktan sonra, günlük depolama alanını yeniden yapılandırmanız gerekir. Öncelikle, şu komutu çalıştırarak mevcut günlük dosyalarını silin:

rm -rf /home/kafka/kafka-logs/*

Daha sonra yeni bir küme kimliği oluşturun ve bunu bir ortam değişkeninde saklayın:

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

Terminalde göster:

echo $KAFKA_CLUSTER_ID

Çıktı kimliği şu şekilde olacaktır:

OutputMjj4bch9Q3-B0TEXv8_zPg

Bu değeri not edin; ikinci ve üçüncü düğümleri yapılandırmak için buna ihtiyacınız olacak.

Son olarak rapor deposunu oluşturmak için aşağıdaki komutu çalıştırın:

./bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

Çıktı şuna benzer olacaktır:

Output...
Formatting /home/kafka/kafka-logs with metadata.version 3.7-IV4.

İkinci ve üçüncü düğümlerin yapılandırılması

Diğer düğümleri yapılandırmak, ilk düğüm için yaptığınıza çok benzer. Ayrıca node.id'yi de güncellemeniz gerektiğini unutmayın:

...
node.id=node_number
...

İkinci ve üçüncü düğümler için uygun değerler sırasıyla 2 ve 3'tür, dinleyiciler ve duyurulan dinleyiciler için uygun adresleri belirler.

Günlük depolamasını yeniden oluştururken, ilk düğümdeki küme kimliğini yeniden kullanın:

KAFKA_CLUSTER_ID="your_cluster_id"

İşiniz bittiğinde, aşağıdaki komutu çalıştırarak üç düğümde de Kafka hizmetini başlatın:

sudo systemctl start kafka

Bu noktada, üç Kafka düğümünü bir KRaft kümesinin parçası olacak şekilde yapılandırmış olursunuz. Bir konu oluşturur ve kümenizde iletiler üretip tüketirsiniz.

Adım 2 – Kümeye Bağlanma

Bu adımda, Kafka ile birlikte gelen kabuk betiklerini kullanarak Kafka kümesine bağlanacaksınız. Ayrıca bir konu başlığı oluşturup kümeden veri üretip tüketmeye çalışacaksınız. Ardından, düğümlerden birini devre dışı bırakıp Kafka'nın veri kaybını nasıl azalttığını göreceksiniz.

Kafka, küme ve üyeleri hakkında bilgi görüntüleyen kafka-metadata-quorum.sh adlı bir betik sağlar. Çalıştırmak için aşağıdaki komutu çalıştırın:

./bin/kafka-metadata-quorum.sh --bootstrap-controller kafka1.your_domain:9093 describe --status

Denetleyicinin uç noktası olan (ancak aracının uç noktası olmayan) 9093 numaralı bağlantı noktasındaki düğümlerden birine bağlanacaksınız. kafka1.your_domain ifadesini, Kafka düğümlerinizden birine işaret eden etki alanıyla değiştirmeyi unutmayın.

Çıktı şuna benzer olacaktır:

OutputClusterId: G3TeIZoeTSCvG2YOWvPE2w
LeaderId: 3
LeaderEpoch: 2
HighWatermark: 383
MaxFollowerLag: 0
MaxFollowerLagTimeMs: 55
CurrentVoters: [1,2,3]
CurrentObservers: []

Betik, küme durumu hakkında ilk bilgileri listeler. Gösterilen çıktıda, 3 numaralı düğümün lider olarak seçildiğini ve üç düğümün de ([1,2,3]) oylama kümesinde olduğunu ve bu kararda hemfikir olduğunu görebilirsiniz.

Aşağıdakileri çalıştırarak İlk Konu adında bir konu oluşturun:

./bin/kafka-topics.sh --create --topic first-topic --bootstrap-server kafka1.your_domain:9092 --replication-factor 2

Çıktı şu şekilde olacaktır:

Created topic first-topic.

Daha sonra kafka-topics.sh betiğini çalıştırarak bölümlerin düğümlerde nasıl düzenlendiğini görün:

./bin/kafka-topics.sh --describe --bootstrap-server kafka1.your_domain:9092 --topic first-topic

Tekrarlama faktörünü 2'ye ayarlamak, konunun en az iki düğümde mevcut olmasını sağlar.

Çıktı şuna benzer olacaktır:

OutputTopic: first-topic TopicId: 4kVImoFNTQeyk3r2zQbdvw PartitionCount: 6 ReplicationFactor: 2 Configs: segment.bytes=1073741824
Topic: first-topic Partition: 0 Leader: 3 Replicas: 3,1 Isr: 3,1
Topic: first-topic Partition: 1 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: first-topic Partition: 2 Leader: 2 Replicas: 2,3 Isr: 2,3
Topic: first-topic Partition: 3 Leader: 1 Replicas: 1,3 Isr: 1,3
Topic: first-topic Partition: 4 Leader: 3 Replicas: 3,2 Isr: 3,2
Topic: first-topic Partition: 5 Leader: 2 Replicas: 2,1 Isr: 2,1

Her bölümün bir lideri, iki kopyası ve iki eşzamanlanmış kopya kümesi (ISR) olduğunu görebilirsiniz. Bölüm lideri, bölümün verilerini istemcilere sunan bir sunucu düğümüdür; kopyalar ise yalnızca kopyaları tutar. Bir kopya düğümü, varsayılan olarak son on saniyedir liderle eşzamanlanmışsa ISR olarak kabul edilir. Bu aralık, konu bazında yapılandırılabilir.

Artık bir konu oluşturduğunuza göre, kafka-console-producer.sh betiğini kullanarak mesajlarını üreteceksiniz. Üreticiyi başlatmak için aşağıdaki komutu çalıştırın:

./bin/kafka-console-producer.sh --topic first-topic --bootstrap-server kafka1.your_domain:9092

Boş bir bildirim göreceksiniz:

>

Sağlayıcı SMS'inizi bekliyor. Testi girin ve ENTER'a basın. Bildirim şu şekilde görünecektir:

>Hello World!
>

Üretici artık bir sonraki mesajı bekliyor; bu, önceki mesajın Kafka'ya başarıyla iletildiği anlamına geliyor. Test için istediğiniz sayıda mesaj girebilirsiniz. Üreticiden çıkmak için CTRL+C tuşlarına basın.

Konu mesajlarını tekrar okumak için bir tüketiciye ihtiyacınız var. Kafka, kafka-console-consumer.sh adında basit bir tüketici sağlar. Çalıştırmak için şunu çalıştırın:

./bin/kafka-console-consumer.sh --topic first-topic --from-beginning --bootstrap-server kafka1.your_domain:9092

Konudan okunan mesajları göreceksiniz:

OutputHello World!
...

Düğüm arızası simülasyonu

Üçüncü Kafka düğümünde, aşağıdaki komutu çalıştırarak hizmeti durdurun:

sudo systemctl stop kafka

Daha sonra konuyu uygulamalı olarak anlatalım:

./bin/kafka-topics.sh --describe --bootstrap-server kafka1.your_domain:9092 --topic first-topic

Çıktı şuna benzer olacaktır:

OutputTopic: first-topic TopicId: 4kVImoFNTQeyk3r2zQbdvw PartitionCount: 6 ReplicationFactor: 2 Configs: segment.bytes=1073741824
Topic: first-topic Partition: 0 Leader: 1 Replicas: 3,1 Isr: 1
Topic: first-topic Partition: 1 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: first-topic Partition: 2 Leader: 2 Replicas: 2,3 Isr: 2
Topic: first-topic Partition: 3 Leader: 1 Replicas: 1,3 Isr: 1
Topic: first-topic Partition: 4 Leader: 2 Replicas: 3,2 Isr: 2
Topic: first-topic Partition: 5 Leader: 2 Replicas: 2,1 Isr: 2,1

3. düğüm bir kopya olarak listelenmiş olsa da, erişilemez olduğu için ISR kümelerinde yer almıyor. Kümeye yeniden katıldığında, diğer düğümlerle senkronize oluyor ve önceki konumunu geri kazanmaya çalışıyor.

İlk konudaki mesajları tekrar okumayı deneyin:

./bin/kafka-console-consumer.sh --topic first-topic --from-beginning --bootstrap-server kafka1.your_domain:9092

Her zamanki gibi erişilebilir olduklarını göreceksiniz:

OutputHello World!
...

Replikaların varlığı sayesinde ilk iki düğüm kontrolü ele alır ve tüketiciye hizmet verir. Artık Kafka'yı üçüncü sunucuda başlatabilirsiniz:

sudo systemctl start kafka

Bu noktada, Kafka'nın kümedeki bir düğümün kullanılamama durumunu nasıl hafiflettiğini gördünüz. Şimdi bir düğümü kümeden nasıl kaldıracağınızı öğreneceksiniz.

Adım 3 – Düğümler arasında veri aktarımı

Bu adımda, bir Kafka kümesindeki düğümler arasında konuları nasıl taşıyacağınızı öğreneceksiniz. Konuları olan mevcut bir kümeye bir düğüm eklerken, Kafka otomatik olarak herhangi bir bölümü taşımaz; ancak siz bunu yapmak isteyebilirsiniz. Bu, düğümleri kaldırmak için de kullanışlıdır, çünkü mevcut bölümler kalan düğümlere otomatik olarak taşınmaz.

Kafka, geçiş planlarını oluşturabilen, çalıştırabilen ve doğrulayabilen kafka-reassign-partitions.sh adlı bir betik sağlar. Bunu, ilk konunun bölümlerini ilk iki düğüme geçirmek için bir plan oluşturmak amacıyla kullanacaksınız.

Öncelikle hangi konuların taşınacağını belirtmeniz gerekiyor. Komut dosyası, konu tanımlarını içeren bir JSON dosyası kabul ediyor, bu nedenle dosyayı oluşturup düzenleme için açın:

vi topics-to-move.json

Aşağıdaki satırları ekleyin:

{
"topics": [
{
"topic": "first-topic"
}
],
"version": 1
}

Başlıklar altında, ilk konuya başvuran bir nesne tanımlarsınız. İşiniz bittiğinde dosyayı kaydedip kapatın.

Göç planını oluşturmak için kafka1.your_domain ifadesini Kafka düğümlerinizden birine işaret eden etki alanıyla değiştirerek aşağıdaki komutu çalıştırın:

./bin/kafka-reassign-partitions.sh --bootstrap-server kafka1.your_domain:9092 --topics-to-move-json-file topics-to-move.json --broker-list "1,2" --generate

Hedef brokerların kimliğini temsil eden –broker-list'e “1,2” değerini geçirirsiniz.

Çıktı şuna benzer olacaktır:

OutputCurrent partition replica assignment
{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[3,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[2,1],"log_dirs":["any","any"]}]}
Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[1,2],"log_dirs":["any","any"]}]}

Bu betik, sırasıyla mevcut ve önerilen bölüm düzenlerini açıklayan toplam iki plan oluşturdu. Değişiklikleri daha sonra geri almak isterseniz, ilk plan size sunulmuştur. migration-plan.json adlı ayrı bir dosyaya kaydedeceğiniz ikinci plana dikkat edin. Oluşturun ve düzenlemek için açın:

vi migration-plan.json

İkinci çalıştırılabilir dosyayı ekleyin:

{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[1,2],"log_dirs":["any","any"]}]}

Dosyayı kaydedip kapatın. Ardından çalıştırmak için aşağıdaki komutu çalıştırın:

./bin/kafka-reassign-partitions.sh --bootstrap-server kafka1.your_domain:9092 --reassignment-json-file migration-plan.json --execute

Çıktı şu şekilde olacaktır:

OutputCurrent partition replica assignment
{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[3,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[2,1],"log_dirs":["any","any"]}]}
Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for first-topic-0,first-topic-1,first-topic-2,first-topic-3,first-topic-4,first-topic-5

Betik, geçişin başladığını gösteriyordu. Geçişin ilerleme durumunu görmek için bunun yerine --verify girin:

./bin/kafka-reassign-partitions.sh --bootstrap-server kafka1.your_domain:9092 --reassignment-json-file migration-plan.json --verify

Bir süre sonra çıktı şu şekilde görünecektir:

OutputStatus of partition reassignment:
Reassignment of partition first-topic-0 is completed.
Reassignment of partition first-topic-1 is completed.
Reassignment of partition first-topic-2 is completed.
Reassignment of partition first-topic-3 is completed.
Reassignment of partition first-topic-4 is completed.
Reassignment of partition first-topic-5 is completed.
Clearing broker-level throttles on brokers 1,2,3
Clearing topic-level throttles on topic first-topic

Şimdi Broker 3'te hiçbir bölüm olmadığını kontrol etmek için ilk sorunu açıklayabilirsiniz:

./bin/kafka-topics.sh --describe --bootstrap-server kafka1.your_domain:9092 --topic first-topic

Çıktı şu şekilde olacaktır:

OutputTopic: first-topic TopicId: 4kVImoFNTQeyk3r2zQbdvw PartitionCount: 6 ReplicationFactor: 2 Configs: segment.bytes=1073741824
Topic: first-topic Partition: 0 Leader: 2 Replicas: 2,1 Isr: 1,2
Topic: first-topic Partition: 1 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: first-topic Partition: 2 Leader: 2 Replicas: 2,1 Isr: 2,1
Topic: first-topic Partition: 3 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: first-topic Partition: 4 Leader: 2 Replicas: 2,1 Isr: 2,1
Topic: first-topic Partition: 5 Leader: 1 Replicas: 1,2 Isr: 2,1

Yalnızca 1 ve 2 numaralı brokerlar replika ve ISR olarak mevcut olduğundan, geçiş başarılı olmuştur.

Bu noktada, ilk konuyu broker 3'ten kalan konulara taşımak için bir geçiş planı oluşturdunuz ve geçişin sorunsuz bir şekilde gerçekleştiğini nasıl doğrulayacağınızı öğrendiniz.

Sonuç

Artık KRaft protokolünü kullanarak iletişim kuran üç düğümden oluşan bir Kafka kümeniz var. Ayrıca kümeyi ve bölüm düzenini nasıl denetleyeceğinizi de öğrendiniz. Bir düğümü kapatıp bir konudan okuyarak küme yedekliliğini test ettiniz. Son olarak, konuları küme düğümlerine nasıl geçireceğinizi öğrendiniz.