Bu yazım içerisinde Kafka’nın kurulumu ve iyi uygulamalarından bahsedeceğim eğer Kafka nedir ve Kafka’nın terminolojisini öğrenmek istiyorsanız bir önceki yazımı okuyabilirsiniz.

Bu yazıdan kurulum olarak hem Zookeeper’lı Kafka kurulumu hem de Kafka 2.8 versiyonu itibariyle Zookeeper’a alternatif olarak Kafka’ya dahil ettiği KRaft’lı kurulum yönergelerinide paylaşacağım. Bunların kurulum adımları haricinde Kafka’nın kapasite planlaması, Kafka kümesinin kurulması, performans testleri, güvenlik ayarları ve farklı Kafka kümeleri arasında veri transferi konularında ciddi bir değişiklik olmamaktadır.

Senaryo:

Senaryo olarak iki farklı veri merkezine 3’er adet Kafka broker kurup veri merkezleri arasında senkronizasyonu sağlamaycağız. Tabii bunu yaparken Kafka’ların hem kendi veri merkezleri içerisinde yedekli olmasını garanti altına almak istiyor hem de Kafka üzerinde ACL, SSL gibi güvenlik önlemlerini almak istiyoruz.

Bu yazıda Kafka’nın kurulumunu tamamlayıp bir sonraki adımdaysa SSL kurulumu, SCRAM ayarları, MirrorMaker 2 ve performans testlerine odaklanıyor olacağım.

Son olarak Kafka’yı 3 adet Zookeeper ve sadece 2 adet Broker ile kurmak mümkün ama daha gerçekci bir senaryo oluşturabilmek için 3 adet Broker kurulumu yapıyor olacağım. Bunun iki sebebi var birincisi kümeleri planlarken birden fazla node’un devre dışı kalma ihtimaline göre planlama yapmak öbürüyse min-ISR değerini 2 yaptığınız taktirde tek bir node’u kaybettiğinizde kümeniz duracak, min-ISR 1 yaptığınız taktirde de veri yedekliliğini kullanamayacaksınız.

Kapasite Planlaması:

Her kurulumun ilk adımı kapasite planlaması olmalıdır. Kafka ve Zookeeper sunucularını kuracağımız sistemlerin CPU, Memory, Disk ve network ihtiyaçlarını önceden hesaplamak ve ona göre kurulumlarımızı yapmamız her zaman operasyonlarımız için daha sağlıklı olacaktır.

CPU: Kafka çok fazla CPU tüketen ve performans dar boğazlarına genelde sebep olan bileşen değildir. Buna rağmen CPU ihtiyaçlarını planlarken dikkat edilmesi gereken bir kaç kritik nokta var.

• Kafka’nın bileşenleri multi-threaded çalışmasından dolayı genel sistem performansı CPU’nun hızından çok core sayısına bağlıdır.
• Eğer consumer’larınızı max.fetch.wait=0 olarak ayarlarsanız bu CPU tüketimine olumsuz olarak yansıyacaktır. Eğer uygulamanız izin veriyorsan consumer ve producer’ların işlemlerini batch olarak yapmasını tercih edebilirsiniz.
• Kafka mesajları şifrelemek(encryption) veya sıkıştırma(compression) özelliğine sahiptir. Eğer bunları kullanacaksanız mutlaka ek CPU tüketimini göz önüne alınız.

CPU olarak orta yoğunlukta bir Kafka kümesi için Broker’lara minimum 4 CPU ve Zookeeper’lar için min 2 CPU önermekteyim.

Memory: Tüm Kafka ve zookeeper bileşenleri JVM kullanmaktadır, minimum 4GB’lık bir heap size hem Zookeeper hem de Kafka Broker’ları için yeterlidir. Bunun altında bir heap size; çok fazla garbace collection’a sebep olacağından dolayı CPU kullanımını artırır.

Başka önemli noktaysa Kafka’ya yazılan tüm veriler eş zamanlı olarak partition’lara(disk) yazılsada aynı zamanda memory’de yer olduğu müddetçe OS Page Cache’de de tutulur, bu sayede mesaj çağrıldığında diske gitmek yerine direk memory’den çağrılabilir. Bu genel sistem performansını etkileyen bir parametredir. Memory ihtiyacını kabaca aşağıda bulunan formülü kullanarak hesaplayabilirsiniz. 

(avg-msg size) * (Saat) * (Saatlik mesaj sayısı)*(toplam kopya sayısı) / Broker sayısı

Mesela 3 broker’lı saatte 1.000.000 mesaj yazan ve ortalama mesaj büyüklüğü 10KB olan bir kümede mesajları 2 saat kadar Page’de tutmak istiyoruz ve replication factor’ün 2(leader ve follower) olması durumunda formülümüzü uyarladığımızda;

(10kb)*2*(1000000)*2 /3 = 12.8 GB

13GB civarı Broker başına bir memory ihtiyaçı çıkacaktır. Buna heap size ve OS ihtiyaçlarını da eklediğiniz zaman kabaca memory ihtiyacınız çıkacaktır.

Storage: Disklerin performansı hem Zookeeper hem de Kafka Broker’ların performansını doğrudan etkileyen bileşenlerdir. Zookeeper veya yerine gelen Kraft sürekli bir metadata transferi yapacağından dolayı ve loğları çok hızlı şekilde yazmak zorundadır; bu bileşenler disk kapasitesinden çok yüksek performans ihtiyacına sahiptir ve genelde SSD/Flash diskler tercih edilmelidir.

Broker tarafındaysa hem kapasite hem de hız önemli faktörlerdir. İyi uygulama örneği olarak Kafka Broker’lara mümkün olduğu kadar hızlı diskler verilmelidir, bu sayede log(mesaj) geldiğinde page’de tutulmasının yanında commit edebilmek için Partition’a yazması gerekmektedir, bu işlemi ne kadar hızlı yaparsa producer bir sonraki mesajını yollamaya başlayabilir, bu da genel sistem performansını ciddi şekilde etkilemektedir. Aynı zamanda yoğun sistemlerde memory hızla dolabilir ve consumer loğlara erişebileceği zaman disklerden okunması gereklidir.

Peki disk kapasite ihtiyacını neye göre karar veriyoruz; kapasite aşağıdaki parametrelere bağımlıdır

• Ortalama mesaj büyüklüğü
• Günlük mesaj sayısı
• Retention süresi(gün cinsinden)
• Replication sayısı

Kabaca formülü;

(ortalama mesaj büyüklüğü) * (günlük mesaj sayısı) * (retention süresi) * (replication sayısı)

Örnek olarak: 10KB’lık mesajımızı 3 gün süreyle tutmak istiyoruz ve günde 1000000 kadar mesaj gelsin. Replication factorümüz 3

10KB * 3 * 3 * 1000000 = 85GB’lık bir disk ihtiyacımız ortaya çıkacaktır.

Kafka Kurulumu

Kurulum adımlarını yazarken hem Kraft’lı versiyona hem de Zookeeper’lı versiyona uygun olmasına özen gösterdim. Kraft kafka için metadata yönetimini yapan ve Zookeeper’ın yerine 3.0 versiyonu itibariyle geçen bir araçtır. Kafka’nın tüm ihtiyaç duyduğu metadataları internal bir topic üzerinde tutmaktadır, çalışma mekanizması olarak __consumer_offtopics ile benzerlik gösterir.

Kafka 3.0 hala içerisinde Zookeeper binary ve konfigurasyon dosyaları ile gelmektedir, yani istediğinizi kullanabilirsiniz.

Genel kurulum önerileri:

Kafka ve zookeeper servislerini çalıştırmak için ayrı bir kullanıcı yaratmanızı güvenlik sebebiyle öneririm, her iki servis için tek kullanıcı yeterlidir. Aynı zamanda Kafka ve Zookeeper’ın verilerini yazabilmesi için ayrı bir disk ekleyip bunu EXT4 veya XFS ile formatlamanızı öneririm.

Son olarak Kafka ve Zookeeper; Scala ve Java tabanlı uygulamalar olduğundan JVM kurulu olması gereklidir.

Kafka Download:

Kafka tar.gz olarak gelmekte ve tek yapmanız gereken dosyayı istediğiniz bir alana açmaktır. Kafka içerisinde aynı zamanda Zookeeper veya Kraft içinde tüm binary ve configurasyon dosyaları bulunmaktadır.

Ben örnek boyunca binary’ler için /opt/kafka klasörünü ama verinin yazılacağı alanlar için /datadisk dosyasını kullanacağım.

Zookeeper kurulumu:

1)/datadisk içerisinde zookeeper isimli bir klasör oluşturun, bu klasörü zookeeper’in kendi log(mesajları) ve veri tabanı için kullanacağız.

2)Her sunucu için Zookeeper ID yaratılması gerekmektedir, ID sadece sayılardan oluşabilir. Aşağıdaki komutu çalıştırarak ID ve dosyayı yaratalım. ID her sunucu için ayrı olmalıdır.
echo “1” > /datadisk/zookeeper/myid

3) /opt/kafka altında bulunan “config” klasörü altındaki zookeeper.properties dosyasını açalım ve içindeki bilgileri şu şekilde dolduralım.

# Zookeeper loğları ve snapshotları nereye saklanacak aynı zamanda myid’i hangi klasörde arayacak

dataDir=/datadisk/zookeeper

# Zookeeper Client portu

clientPort=2181

# Eş zamanlı bağlantı sayısını sınırlamak için kullanılır “0” sınırsız için kullanılır

maxClientCnxns=0

# MS cinsinden heatbeat

tickTime=2000

# Senkronizasyon için ne kadar süre ayrıldığı(ms), bu süre içerisinde sync sağlanamazsa bağlantı timeout oluyor

initLimit=10

# Timeout olmadan önce(küme dışına çıkmadan önce) kaç kere ulaşamayacağını belirler.

syncLimit=5

# Zookeeper kümesinde olan sunucuların FQDN olarak domain adları ve portları.

server.1=zookeeper1:2888:3888

server.2=zookeeper2:2888:3888

server.3=zookeeper3:2888:3888

En son satırlarda bulunan server.1, server.2, server.3 olan kısma dikkatinizi çekmek isterim. server.<id> kısmında ki id numarası ile FQDN’i girilmiş zookeeper sunucusunun myid dosyasında yazan id numaraları aynı olmalıdır.

Bu adımlardan sonra Zookeeper’ın çalışıp çalışmadığını ve diğer node’lara bağlanıp bağlanmadığını test edebiliriz. Aşağıda ki komut ile sunucuları başlatalım.

/opt/kafka/zookeeper-server-start.sh –deamon /opt/kafka/config/zookeeper.properties

Loglar içerisinde Leader’ın seçildiğini görürseniz ve kümede election işlemleri sorunsuz biterse Zookeeper’ı kapatıp Kafka kurulumuna geçebiliriz.

Kafka Kurulumu:

Kafka kurulumunu Zookeeper’lı ve Zookeeper’sız olarak ayrı ayrı anlatacağım. En başta Zookeeper’lı anlatım.

/opt/kafka/config altında bulunan server.properties dosyasını açalım. Burada bulunan tüm ayarlardan bahsetmek yerine sadece kritik olanlardan bahsedeceğim.

1)Broker.id = 1; Broker ID her bir broker için eşsiz olmalıdır, sunucularınızın id’lerini sırayla 1,2,3,4 veya 10,15,20 şeklinde verebilirsiniz. Sadece numaraları kabul etmektedir ve her kafka broker’ı için eşsiz olmalıdır.

2)listeners=PLAINTEXT://:9092
Listeners Kafka’nın hangi porttan hangi protokol ile trafiği dinleyeceğini belirtmektedir. Kafka an itibariyle aşağıdaki 4 protokolü destekliyor;

• PLAINTEXT: Consumer ve Producer’ların kafka’yla yaptığı trafik clear text’dir ve Topic’ler için herhangi bir kullanıcı adı şifreye zorlanmaz
• SSL: Consumer ve Producer’ların kafka’yla yaptığı trafik SSL ile şifrelenir ve Topic’ler için herhangi bir kullanıcı adı şifreye zorlanmaz
• SASL_PLAINTEXT: Consumer ve Producer’ların kafka’yla yaptığı trafik clear text’dir ama Topic’ler için kullanıcı adı ve şifre ile gelinmesini zorumlu kılar
• SASL_SSL: Consumer ve Producer’ların kafka’yla yaptığı trafik SSL ile şifrelenir ve Topic’ler için kullanıcı adı ve şifre ile gelinmesini zorumlu kılar

Canlı ortamlarda SASL_SSL kullanılmasını öneririm ama ilk kurulum için şimdilik sadece PLAINTEXT olarak ayarlayabilirsiniz.

3)log.dirs =/dataDisk/kafka
Logların hangi fiziksel alana yazılacağını belirler, partitionlar burada oluşturulur.

4)num.partitions: Topic’ler için varsayılan otomatik oluşturulan partition sayısını belirler, minimum broker sayısı kadar oluşturmanız performans açısından en iyi sonucu verecektir.

5) offsets.topic.replication.factor, transaction.state.log.replication.factor ve transaction.state.log.min.isr parametrelerini bir bütün olarak ele almak lazım. Kafka her partition için hangi offset’te kalındığını ve küme içinde metadata transferini _consumer_offsets ve __transaction_state isimli iç topic’leriyle yönetir.

Kafka’yı ilk çalıştırdığınızda bunları yaratır ve yaratırken bu topiclerin kaç partition olacağını belirleyen parametrelerdir.

Önerilen sayı minimum 3 olmasıdır ama benim önerim broker sayısına eşit yapılmasıdır. Eğer bu sayıyı 1 olarak ayarladığınız durumda offsets bilgisi sadece bir adet broker’da kalacaktır. Bu broker’ın devre dışı kalması durumunda tüm consumer trafiği duracaktır.

6) log.retention.hours=24 ve log.retention.bytes=1073741824; Kafka’ya gelen mesajların ne kadar süre saklanacağını belirleyen parametrelerdir. Saat veya büyüklük cinsindendir, hangi kondisyon önce karşılanırsa ona göre retention politikası devreye girer.

7) zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181
Kafka’nın kümede bulunan node, partition, topic ve ACLs bilgilerini takip edebilmek ve kontrol edebilmesi için kullandığı zookeeper’ların FQDN olarak girilmesi gereklidir.

Bu adımları takip ettiğimiz taktirde; SSL ve SASL desteği olmadan Kafka kümesini çalıştırabilirsiniz ve aşağıdaki komutu her bir broker sunucusu için çalıştırmanız yeterlidir.

/opt/kafka/kafka-server-start.sh –deamon /opt/kafka/config/server.properties

Kraft’lı kurulum:

Kraft ile kurulumda bir önceki kısımda ki adımlardan 7. adım hariç hepsi bire bir aynıdır sadece bu sefer işlemleri /opt/kafka/config/Kraft/server.properties dosyasında yapmamız gerekli.

1)process.roles=broker, controller
Kafka node’un rolünü belirliyoruz, 2.8 itibariyle bir kafka node’u broker ve/veya controller olabiliyor. Eğer yukardaki örnekte olduğu gibi her ikisini de yazarsak node her iki role’de aynı anda sahip olabilir.

2)listeners= PLAINTEXT://:9092,CONTROLLER://:19092
Yukarda belirtilen listener protokollerine ek olarak Controller isimli yeni bir protokolde eklenmiştir, bunu tüm controller rolüne sahip node’lara eklememiz gerekli.

3)controller.quorum.voters=1@kcontroller1:19092,2@kcontroller2:19092,3@kcontroller3:19092

Zookeeper connect parametresinin çalışmasına benzer şekilde Kraft controller’larının bulunduğu sunucuları FQDN olarak ve port bilgileriyle beraber giriyoruz.

4)Kraft’da quorum olarak üzerinde bulunduğu sunucuların disklerini kullandığından ve Zookeeper yerine küme id’sini belirleyeeğinden dolayı. Kafka küme ID’sini yaratmamız gereklidir.
/opt/kafka/bin/kafka-storage.sh random-uuid
Çıkan ID’yi bir yere kopyalayın.

5)Son olarak yarattığımız ID ve kafka-storage’ı kullanarak kafka’nın kullanacağı disk alanını formatlamamız lazım. Burada formatlanacak alan logs.dir parametresiyle aynıdır.

./bin/kafka-storage.sh format -t <id> -c /opt/kafka/config/kraft/server.properties

Bu adımlardan sonra kafka servislerinizi başlatabilirsiniz. Kraft’ın kurulumu ve detayları hakkında config altında bulunan Readme.md dosyasının içeriğine bakmanızı öneririm.

Kafka’yı test etmek:

Kafka’nın sorunsuz çalıştığını test etmenin en iyi yöntemi bir topic yaratıp içerisine mesaj yazmaktan geçer.

1)Topic yaratmak için herhangi bir kafka broker sunucusuna girip aşağıdaki komutu giriniz;

 ./bin/kafka-topics.sh --create --topic kafka-test --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092

Bu komut kafka-test isimli 3 partitiondan oluşan ve her partition’nın 3 kopyası olan bir topic yaratacaktır.

2)Kafka üzerindeki tüm topic’leri listelemek için aşağıdaki komutu kullanabilirsiniz, bunun çıktısı olarak kafka-topic’i göreceksiniz.
bin/kafka-topics.sh --bootstrap-server localhost:9093 --list

3)Eğer topic hakkında detaylı bilgi istersek;

bin/kafka-topics.sh --bootstrap-server localhost:9093 --describe --topic kafka-test

Bu komutun çıktısı sonucunda her partition’nın liderinin(aktif) hangi broker üzerinde olduğu, replication’larının hangi broker’larda olduğu ve senkronizasyon durumunu görebiliriz.

Bir sonraki yazımda Kafka için güvenlik ayarlarını yapacağız, performans testleri ve opensource arayüzlerden bahsedeceğim.

Okudunuz için teşekkür ederim.