Apache Kafka, y\u00fcksek performansl\u0131, da\u011f\u0131t\u0131k bir ak\u0131\u015f platformu olarak g\u00fcn\u00fcm\u00fcz modern veri mimarilerinin temel ta\u015flar\u0131ndan biridir. B\u00fcy\u00fck \u00f6l\u00e7ekli veri ak\u0131\u015flar\u0131n\u0131 i\u015flemek, ger\u00e7ek zamanl\u0131 uygulamalar geli\u015ftirmek ve mikroservis mimarilerinde mesajla\u015fma sa\u011flamak i\u00e7in yayg\u0131n olarak kullan\u0131l\u0131r. Ancak Kafka’n\u0131n g\u00fcc\u00fcnden tam anlam\u0131yla faydalanabilmek i\u00e7in, onu sadece bir mesaj kuyru\u011fu olarak kullanmak yeterli de\u011fildir; ayn\u0131 zamanda operasyonel s\u00fcre\u00e7lerini de etkin bir \u015fekilde y\u00f6netmek gerekir. Konu (topic) olu\u015fturma, silme, yap\u0131land\u0131rma, t\u00fcketici gruplar\u0131n\u0131 izleme ve y\u00f6netme gibi g\u00f6revler, manuel olarak yap\u0131ld\u0131\u011f\u0131nda zaman al\u0131c\u0131, hataya a\u00e7\u0131k ve \u00f6l\u00e7eklenemez hale gelebilir. \u0130\u015fte bu noktada Kafka’y\u0131 programatik olarak y\u00f6netme ihtiyac\u0131 ortaya \u00e7\u0131kar.<\/p>\n
Bu makale, Kafka k\u00fcmelerini programatik olarak y\u00f6netmenin temel prensiplerini, kullan\u0131lan API’leri ve k\u00fct\u00fcphaneleri, yayg\u0131n y\u00f6netim g\u00f6revlerini ve en iyi uygulamalar\u0131 derinlemesine inceleyecektir. Java’n\u0131n Kafka gibi da\u011f\u0131t\u0131k bir sistemin y\u00f6netimi, \u00f6zellikle b\u00fcy\u00fck ve karma\u015f\u0131k ortamlarda, manuel i\u015flemlerle s\u00fcrd\u00fcr\u00fclemez hale gelir. Programatik y\u00f6netim, bu zorluklar\u0131n \u00fcstesinden gelmek i\u00e7in bir dizi \u00f6nemli avantaj sunar:<\/p>\n Programatik y\u00f6netim, tekrarlayan g\u00f6revleri otomatikle\u015ftirmeyi sa\u011flar. Yeni bir uygulaman\u0131n devreye al\u0131nmas\u0131yla birlikte gerekli Kafka konular\u0131n\u0131 ve ACL’leri otomatik olarak olu\u015fturmak, art\u0131k kullan\u0131lmayan konular\u0131 temizlemek veya belirli ko\u015fullar alt\u0131nda partisyon say\u0131lar\u0131n\u0131 ayarlamak gibi i\u015flemler, script’ler veya otomasyon ara\u00e7lar\u0131 arac\u0131l\u0131\u011f\u0131yla ger\u00e7ekle\u015ftirilebilir. Bu, insan hatas\u0131 riskini azalt\u0131r ve operasyonel verimlili\u011fi art\u0131r\u0131r.<\/p>\n \u0130\u015f y\u00fckleri zamanla de\u011fi\u015febilir. Programatik y\u00f6netim, Kafka k\u00fcmenizin bu de\u011fi\u015fikliklere dinamik olarak uyum sa\u011flamas\u0131na olanak tan\u0131r. \u00d6rne\u011fin, belirli bir konudaki mesaj hacmi beklenenden fazla artt\u0131\u011f\u0131nda, programatik olarak daha fazla partisyon eklenebilir. Bu, sistemin performans\u0131n\u0131 ve kullan\u0131labilirli\u011fini korumak i\u00e7in kritik \u00f6neme sahiptir.<\/p>\n S\u00fcrekli Entegrasyon\/S\u00fcrekli Da\u011f\u0131t\u0131m (CI\/CD) s\u00fcre\u00e7lerinde, Kafka kaynaklar\u0131n\u0131n otomatik olarak olu\u015fturulmas\u0131 ve yap\u0131land\u0131r\u0131lmas\u0131 b\u00fcy\u00fck kolayl\u0131k sa\u011flar. Uygulama da\u011f\u0131t\u0131mlar\u0131 s\u0131ras\u0131nda gerekli Kafka konular\u0131n\u0131n ve di\u011fer kaynaklar\u0131n otomatik olarak haz\u0131rlanmas\u0131, da\u011f\u0131t\u0131m s\u00fcre\u00e7lerini h\u0131zland\u0131r\u0131r ve tutarl\u0131l\u0131\u011f\u0131 garanti eder.<\/p>\n Programatik API’ler, ekiplerin kendi \u00f6zel Kafka y\u00f6netim ara\u00e7lar\u0131n\u0131 veya kontrol panellerini geli\u015ftirmelerine olanak tan\u0131r. Bu ara\u00e7lar, \u015firketin \u00f6zel ihtiya\u00e7lar\u0131na g\u00f6re tasarlanabilir, mevcut izleme ve uyar\u0131 sistemleriyle entegre edilebilir ve operasyon ekiplerine daha derinlemesine g\u00f6r\u00fcn\u00fcrl\u00fck ve kontrol sa\u011flayabilir.<\/p>\n Programatik y\u00f6netim, hata durumlar\u0131nda otomatik kurtarma mekanizmalar\u0131 geli\u015ftirmek i\u00e7in de kullan\u0131labilir. \u00d6rne\u011fin, bir broker’\u0131n \u00e7evrimd\u0131\u015f\u0131 kalmas\u0131 durumunda, replikasyon fakt\u00f6rlerini veya partisyon da\u011f\u0131l\u0131mlar\u0131n\u0131 yeniden d\u00fczenlemek gibi \u00f6nlemler otomatik olarak al\u0131nabilir. T\u00fcketici gruplar\u0131n\u0131n offset’lerini belirli bir noktaya s\u0131f\u0131rlama gibi kurtarma i\u015flemleri de programatik olarak ger\u00e7ekle\u015ftirilebilir.<\/p>\n Kafka’y\u0131 programatik olarak y\u00f6netmek i\u00e7in \u00e7e\u015fitli API’ler ve k\u00fct\u00fcphaneler mevcuttur. Bunlar\u0131n en temel ve kapsaml\u0131 olan\u0131, Kafka’n\u0131n kendi Java API’sinde bulunan Kafka ekosisteminde Java merkezi bir rol oynar. Konu Olu\u015fturma:<\/strong><\/p>\n Konu Silme:<\/strong><\/p>\n Konular\u0131 Listeleme:<\/strong><\/p>\n Kafka ekosistemi, Java d\u0131\u015f\u0131ndaki diller i\u00e7in de g\u00fc\u00e7l\u00fc istemci k\u00fct\u00fcphaneleri sunar. Bu k\u00fct\u00fcphaneler genellikle C\/C++ tabanl\u0131 * * * Kafka’y\u0131 programatik olarak y\u00f6netirken ger\u00e7ekle\u015ftirilebilecek ba\u015fl\u0131ca g\u00f6revler \u015funlard\u0131r:<\/p>\n Konular, Kafka’daki veri ak\u0131\u015flar\u0131n\u0131n temel birimleridir ve y\u00f6netimsel i\u015flemlerin \u00f6nemli bir b\u00f6l\u00fcm\u00fcn\u00fc olu\u015fturur.<\/p>\n * Konu Olu\u015fturma<\/strong>: Yeni bir veri ak\u0131\u015f\u0131 i\u00e7in bir konu olu\u015fturmak en temel y\u00f6netim g\u00f6revidir. Partisyon say\u0131s\u0131 (mesajlar\u0131n paralel i\u015flenmesini sa\u011flar) ve replikasyon fakt\u00f6r\u00fc (veri dayan\u0131kl\u0131l\u0131\u011f\u0131n\u0131 sa\u011flar) gibi \u00f6nemli parametreler bu a\u015famada belirlenir. Ayr\u0131ca, Kafka k\u00fcmesindeki broker’lar hakk\u0131nda bilgi almak, k\u00fcmenin sa\u011fl\u0131\u011f\u0131n\u0131 ve durumunu anlamak i\u00e7in \u00f6nemlidir.<\/p>\n * Broker’lar\u0131 Listeleme<\/strong>: K\u00fcmedeki aktif broker’lar\u0131n listesini ve ID’lerini almak. T\u00fcketici gruplar\u0131, Kafka’daki mesajlar\u0131n paralel ve hataya dayan\u0131kl\u0131 bir \u015fekilde t\u00fcketilmesini sa\u011flar.<\/p>\n * T\u00fcketici Gruplar\u0131n\u0131 Listeleme<\/strong>: K\u00fcmedeki t\u00fcm aktif t\u00fcketici gruplar\u0131n\u0131 listelemek. Kafka’da g\u00fcvenlik, \u00f6zellikle \u00e7ok kullan\u0131c\u0131l\u0131 veya \u00e7oklu uygulama ortamlar\u0131nda kritik \u00f6neme sahiptir. ACL’ler, belirli kullan\u0131c\u0131lar\u0131n veya uygulamalar\u0131n belirli Kafka kaynaklar\u0131na (konular, broker’lar, t\u00fcketici gruplar\u0131) eri\u015fimini kontrol etmek i\u00e7in kullan\u0131l\u0131r.<\/p>\n * ACL Olu\u015fturma<\/strong>: Belirli bir prensibe (kullan\u0131c\u0131\/uygulama) belirli bir kaynak \u00fczerinde belirli bir izni (okuma\/yazma\/y\u00f6netim) veren ACL’ler olu\u015fturmak. Kafka k\u00fcmesinin performans\u0131n\u0131 ve sa\u011fl\u0131\u011f\u0131n\u0131 izlemek, operasyonel s\u00fcreklilik i\u00e7in kritik \u00f6neme sahiptir. * JMX Metriklerini Programatik Olarak \u00c7ekme<\/strong>: JMX exporter’lar (\u00f6rn. Prometheus JMX exporter) kullan\u0131larak bu metrikler toplanabilir ve izleme sistemlerine aktar\u0131labilir. Programatik Kafka y\u00f6netimi sadece temel g\u00f6revlerle s\u0131n\u0131rl\u0131 de\u011fildir; daha karma\u015f\u0131k senaryolar\u0131 ele almak i\u00e7in geli\u015fmi\u015f yakla\u015f\u0131mlar da mevcuttur.<\/p>\n Da\u011f\u0131t\u0131k sistemlerde API \u00e7a\u011fr\u0131lar\u0131 s\u0131ras\u0131nda a\u011f sorunlar\u0131, broker \u00e7\u00f6kmesi veya zaman a\u015f\u0131mlar\u0131 gibi hatalar meydana gelebilir. Programatik y\u00f6netimde, \u00f6zellikle otomasyon script’lerinde, i\u015flemlerin idempotent (birden fazla kez \u00e7al\u0131\u015ft\u0131r\u0131ld\u0131\u011f\u0131nda ayn\u0131 sonucu veren) olmas\u0131 \u00f6nemlidir. \u00d6rne\u011fin, bir konu olu\u015fturma iste\u011fi, konu zaten mevcutsa hata vermemeli veya mevcut konuyu de\u011fi\u015ftirmemelidir (veya belirli bir politika dahilinde de\u011fi\u015ftirmelidir). Kafka k\u00fcmesini programatik olarak y\u00f6netirken g\u00fcvenlik g\u00f6z ard\u0131 edilmemelidir.<\/p>\n * SSL\/TLS<\/strong>: Y\u00f6netim istemcisi ile Kafka broker’lar\u0131 aras\u0131ndaki ileti\u015fimi \u015fifrelemek i\u00e7in SSL\/TLS kullan\u0131lmal\u0131d\u0131r. Programatik olarak yap\u0131lan de\u011fi\u015fikliklerin kal\u0131c\u0131l\u0131\u011f\u0131 ve konfig\u00fcrasyon dosyalar\u0131yla ili\u015fkisi iyi anla\u015f\u0131lmal\u0131d\u0131r. Kafka’y\u0131 programatik olarak y\u00f6netmenin pratik uygulamalar\u0131 olduk\u00e7a geni\u015ftir.<\/p>\n Her test ko\u015fusundan \u00f6nce yeni, izole Kafka konular\u0131 olu\u015fturmak ve test bittikten sonra bunlar\u0131 temizlemek, testlerin tutarl\u0131l\u0131\u011f\u0131n\u0131 ve tekrarlanabilirli\u011fini sa\u011flar. CI\/CD pipeline’lar\u0131nda bu yakla\u015f\u0131m, entegrasyon testlerini h\u0131zland\u0131rabilir.<\/p>\n Yeni bir mikroservis devreye al\u0131nd\u0131\u011f\u0131nda, onun i\u00e7in gerekli Kafka konular\u0131n\u0131, t\u00fcketici gruplar\u0131n\u0131 ve ACL’leri otomatik olarak olu\u015fturmak, da\u011f\u0131t\u0131m s\u00fcre\u00e7lerini basitle\u015ftirir ve hata olas\u0131l\u0131\u011f\u0131n\u0131 azalt\u0131r. Bu, \u00f6zellikle h\u0131zla b\u00fcy\u00fcyen mikroservis mimarilerinde \u00e7ok de\u011ferlidir.<\/p>\n * Otomatik Konu Bak\u0131m\u0131<\/strong>: Belirli bir s\u00fcre kullan\u0131lmayan veya mesaj i\u00e7ermeyen konular\u0131 otomatik olarak tespit edip silmek. Programatik Kafka y\u00f6netimi bir\u00e7ok avantaj sunsa da, baz\u0131 zorluklarla kar\u015f\u0131la\u015fmak m\u00fcmk\u00fcnd\u00fcr:<\/p>\n * API Versiyon Uyumlulu\u011fu<\/strong>: Kafka API’leri zamanla geli\u015fir. Kulland\u0131\u011f\u0131n\u0131z Kafka istemci k\u00fct\u00fcphanesinin ve Kafka’y\u0131 programatik olarak y\u00f6netmek, modern veri platformlar\u0131nda operasyonel verimlili\u011fi, \u00f6l\u00e7eklenebilirli\u011fi ve g\u00fcvenilirli\u011fi art\u0131rmak i\u00e7in vazge\u00e7ilmez bir yakla\u015f\u0131md\u0131r. Bu makalede ele al\u0131nan prensipleri ve \u00f6rnekleri uygulayarak, ekipler Kafka altyap\u0131lar\u0131n\u0131 daha \u00e7evik, g\u00fcvenilir ve s\u00fcrd\u00fcr\u00fclebilir bir \u015fekilde y\u00f6netebilirler. Programatik y\u00f6netim, sadece operasyonel y\u00fck\u00fc azaltmakla kalmaz, ayn\u0131 zamanda Kafka’n\u0131n dinamik do\u011fas\u0131ndan tam olarak yararlanarak de\u011fi\u015fen i\u015f ihtiya\u00e7lar\u0131na h\u0131zla adapte olabilen sistemler in\u015fa etmenin de anahtar\u0131d\u0131r. Gelecekte, Kafka operasyonlar\u0131n\u0131n daha da otomatik hale gelmesi ve yapay zeka destekli y\u00f6netim ara\u00e7lar\u0131n\u0131n y\u00fckseli\u015fiyle, programatik y\u00f6netimin \u00f6nemi daha da artacakt\u0131r.<\/body><\/p>\n","protected":false},"excerpt":{"rendered":"Kafka’y\u0131 Programatik Olarak Y\u00f6netmek: Kapsaml\u0131 Bir Rehber\nApache Kafka, y\u00fcksek performansl\u0131, da\u011f\u0131t\u0131k bir ak\u0131\u015f platformu olarak g\u00fcn\u00fcm\u00fcz m","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"csco_page_header_type":"","csco_page_load_nextpost":"","csco_page_subscribe_form":"","csco_page_contact_form":"","footnotes":""},"categories":[1],"tags":[],"class_list":{"0":"post-32902","1":"post","2":"type-post","3":"status-publish","4":"format-standard","6":"category-genel","7":"cs-entry","8":"cs-video-wrap"},"yoast_head":"\nAdminClient<\/code> API’si ba\u015fta olmak \u00fczere, farkl\u0131 programlama dillerindeki yakla\u015f\u0131mlara de\u011finecek ve ger\u00e7ek d\u00fcnya senaryolar\u0131nda nas\u0131l kullan\u0131labilece\u011fine dair \u00f6rnekler sunacakt\u0131r. Amac\u0131m\u0131z, geli\u015ftiricilere ve operasyon ekiplerine, Kafka altyap\u0131lar\u0131n\u0131 daha otomatize, esnek ve g\u00fcvenilir bir \u015fekilde y\u00f6netmeleri i\u00e7in gerekli bilgi ve ara\u00e7lar\u0131 sa\u011flamakt\u0131r.<\/p>\nKafka Y\u00f6netimi Neden Programatik Olmal\u0131?<\/h3>\n
Otomasyon ve Verimlilik<\/h4>\n
Dinamik \u00d6l\u00e7eklendirme ve Esneklik<\/h4>\n
CI\/CD Entegrasyonu<\/h4>\n
\u00d6zel Ara\u00e7lar ve Kontrol Panelleri Geli\u015ftirme<\/h4>\n
Hata Y\u00f6netimi ve Kurtarma S\u00fcre\u00e7leri<\/h4>\n
Kafka Y\u00f6netimi i\u00e7in Temel API’ler ve K\u00fct\u00fcphaneler<\/h3>\n
AdminClient<\/code>‘t\u0131r.<\/p>\nJava API’leri<\/h4>\n
AdminClient<\/code> s\u0131n\u0131f\u0131, Kafka k\u00fcmelerini programatik olarak y\u00f6netmek i\u00e7in tasarlanm\u0131\u015f ana aray\u00fczd\u00fcr.<\/p>\nAdminClient<\/h5>\n
AdminClient<\/code>, Kafka k\u00fcmesiyle etkile\u015fim kurarak y\u00f6netimsel g\u00f6revleri ger\u00e7ekle\u015ftirmek i\u00e7in kullan\u0131l\u0131r. Bu g\u00f6revler aras\u0131nda konu olu\u015fturma, silme, listeleme, yap\u0131land\u0131rma de\u011fi\u015fiklikleri, broker bilgisi alma ve t\u00fcketici grubu y\u00f6netimi gibi i\u015flemler bulunur. AdminClient<\/code> asenkron bir API’ye sahiptir, yani \u00e7a\u011fr\u0131lar hemen bir sonu\u00e7 d\u00f6nd\u00fcrmez; bunun yerine bir KafkaFuture<\/code> nesnesi d\u00f6nd\u00fcr\u00fcr, bu da i\u015flemin tamamlanmas\u0131n\u0131 beklemek veya geri \u00e7a\u011f\u0131r\u0131m (callback) mekanizmalar\u0131 kullanmak i\u00e7in kullan\u0131labilir.<\/p>\nAdminClient<\/code> Olu\u015fturma:<\/strong><\/p>\nAdminClient<\/code> olu\u015fturmak i\u00e7in temel Kafka yap\u0131land\u0131rma \u00f6zelliklerini i\u00e7eren bir Properties<\/code> nesnesi gereklidir. En az\u0131ndan bootstrap.servers<\/code> \u00f6zelli\u011fi belirtilmelidir.<\/p>\nimport org.apache.kafka.clients.admin.AdminClient;\nimport org.apache.kafka.clients.admin.AdminClientConfig;\nimport java.util.Properties;\n\npublic class KafkaAdminClientExample {\n\n private AdminClient adminClient;\n\n public KafkaAdminClientExample(String bootstrapServers) {\n Properties props = new Properties();\n props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);\n \/\/ \u0130ste\u011fe ba\u011fl\u0131 olarak g\u00fcvenlik ve di\u011fer ayarlar eklenebilir\n \/\/ props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, \"SASL_SSL\");\n \/\/ props.put(AdminClientConfig.SASL_MECHANISM, \"PLAIN\");\n \/\/ ...\n this.adminClient = AdminClient.create(props);\n }\n\n public void close() {\n if (adminClient != null) {\n adminClient.close();\n }\n }\n\n \/\/ ... y\u00f6netim metotlar\u0131 buraya gelecek\n}<\/code><\/pre>\ncreateTopics<\/code> metodu, bir veya daha fazla konu olu\u015fturmak i\u00e7in kullan\u0131l\u0131r. NewTopic<\/code> nesneleri, olu\u015fturulacak konular\u0131n ad\u0131n\u0131, partisyon say\u0131s\u0131n\u0131 ve replikasyon fakt\u00f6r\u00fcn\u00fc belirtir.<\/p>\nimport org.apache.kafka.clients.admin.NewTopic;\nimport java.util.Collections;\nimport java.util.concurrent.ExecutionException;\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void createTopic(String topicName, int numPartitions, short replicationFactor) {\n NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);\n try {\n adminClient.createTopics(Collections.singletonList(newTopic)).all().get();\n System.out.println(\"Konu '\" + topicName + \"' ba\u015far\u0131yla olu\u015fturuldu.\");\n } catch (InterruptedException | ExecutionException e) {\n System.err.println(\"Konu olu\u015fturulurken hata olu\u015ftu: \" + e.getMessage());\n }\n}<\/code><\/pre>\ndeleteTopics<\/code> metodu, belirtilen konular\u0131 silmek i\u00e7in kullan\u0131l\u0131r.<\/p>\nimport java.util.Arrays;\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void deleteTopic(String topicName) {\n try {\n adminClient.deleteTopics(Arrays.asList(topicName)).all().get();\n System.out.println(\"Konu '\" + topicName + \"' ba\u015far\u0131yla silindi.\");\n } catch (InterruptedException | ExecutionException e) {\n System.err.println(\"Konu silinirken hata olu\u015ftu: \" + e.getMessage());\n }\n}<\/code><\/pre>\nlistTopics<\/code> metodu, k\u00fcmedeki t\u00fcm konular\u0131 listeler.<\/p>\nimport org.apache.kafka.clients.admin.ListTopicsOptions;\nimport org.apache.kafka.clients.admin.TopicListing;\nimport java.util.Set;\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic SetKafkaProducer ve KafkaConsumer<\/h5>\n
KafkaProducer<\/code> ve KafkaConsumer<\/code> do\u011frudan y\u00f6netim API’leri olmasa da, belirli y\u00f6netimsel g\u00f6revlerde dolayl\u0131 olarak kullan\u0131labilirler. \u00d6rne\u011fin, bir konuya test mesajlar\u0131 g\u00f6ndermek veya bir t\u00fcketici grubunun mevcut offset’lerini izlemek i\u00e7in kullan\u0131labilirler. Ayr\u0131ca, bu s\u0131n\u0131flar\u0131n metrikleri, Kafka k\u00fcmesinin sa\u011fl\u0131k durumunu izlemek i\u00e7in programatik olarak toplanabilir.<\/p>\nKafkaStreams<\/h5>\n
KafkaStreams<\/code> k\u00fct\u00fcphanesi, Kafka’da ak\u0131\u015f i\u015fleme uygulamalar\u0131 geli\u015ftirmek i\u00e7in kullan\u0131l\u0131r. Do\u011frudan y\u00f6netimle ilgili olmasa da, karma\u015f\u0131k ak\u0131\u015f i\u015fleme topolojilerini da\u011f\u0131t\u0131rken veya izlerken y\u00f6netimsel g\u00f6revlerle entegre edilebilir.<\/p>\nDi\u011fer Diller i\u00e7in K\u00fct\u00fcphaneler<\/h4>\n
librdkafka<\/code> \u00fczerine in\u015fa edilmi\u015ftir ve AdminClient<\/code> benzeri i\u015flevsellik sa\u011flarlar.<\/p>\nPython<\/h5>\n
kafka-python<\/code><\/strong>: Saf Python ile yaz\u0131lm\u0131\u015f pop\u00fcler bir Kafka istemcisidir. Y\u00f6netimsel i\u015flemler i\u00e7in KafkaAdminClient<\/code> s\u0131n\u0131f\u0131n\u0131 sunar.
\n* confluent-kafka-python<\/code><\/strong>: librdkafka<\/code> \u00fczerine in\u015fa edilmi\u015f y\u00fcksek performansl\u0131 bir istemcidir. AdminClient<\/code> s\u0131n\u0131f\u0131, konu olu\u015fturma, silme ve di\u011fer y\u00f6netimsel g\u00f6revler i\u00e7in kullan\u0131l\u0131r.<\/p>\nconfluent-kafka-python<\/code> ile Konu Olu\u015fturma \u00d6rne\u011fi:<\/strong><\/p>\nfrom confluent_kafka.admin import AdminClient, NewTopic, NewPartitions\nfrom confluent_kafka import KafkaException\nimport sys\n\ndef create_topic_python(bootstrap_servers, topic_name, num_partitions, replication_factor):\n admin_client = AdminClient({'bootstrap.servers': bootstrap_servers})\n\n new_topic = NewTopic(topic_name, num_partitions=num_partitions, replication_factor=replication_factor)\n fs = admin_client.create_topics([new_topic])\n\n for topic, f in fs.items():\n try:\n f.result() # \u0130\u015flemin tamamlanmas\u0131n\u0131 bekle\n print(f\"Konu '{topic}' ba\u015far\u0131yla olu\u015fturuldu.\")\n except KafkaException as e:\n print(f\"Konu '{topic}' olu\u015fturulurken hata olu\u015ftu: {e}\")\n except Exception as e:\n print(f\"Beklenmeyen hata: {e}\")\n admin_client.poll(0) # Bekleyen olaylar\u0131 i\u015fle\n # admin_client.close() # AdminClient'\u0131 kapatmak genellikle gerekli de\u011fildir, kendisi kapan\u0131r\n\nif __name__ == '__main__':\n bootstrap_servers = \"localhost:9092\"\n topic_name = \"my_python_topic\"\n num_partitions = 3\n replication_factor = 1\n create_topic_python(bootstrap_servers, topic_name, num_partitions, replication_factor)<\/code><\/pre>\nGo<\/h5>\n
sarama<\/code><\/strong>: Go i\u00e7in pop\u00fcler ve olgun bir Kafka istemcisidir. Y\u00f6netimsel i\u015flemler i\u00e7in ClusterAdmin<\/code> aray\u00fcz\u00fcn\u00fc sa\u011flar.
\n* confluent-kafka-go<\/code><\/strong>: librdkafka<\/code> \u00fczerine in\u015fa edilmi\u015f Confluent’in Go istemcisidir. AdminClient<\/code> benzeri i\u015flevsellik sunar.<\/p>\nNode.js<\/h5>\n
kafkajs<\/code><\/strong>: Saf JavaScript ile yaz\u0131lm\u0131\u015f modern bir Kafka istemcisidir. Y\u00f6netimsel i\u015flemler i\u00e7in admin<\/code> aray\u00fcz\u00fcn\u00fc sa\u011flar.<\/p>\nTemel Programatik Y\u00f6netim G\u00f6revleri<\/h3>\n
Konu (Topic) Y\u00f6netimi<\/h4>\n
retention.ms<\/code> (mesajlar\u0131n ne kadar s\u00fcre saklanaca\u011f\u0131) veya cleanup.policy<\/code> (mesajlar\u0131n nas\u0131l temizlenece\u011fi – delete\/compact) gibi konu seviyesi yap\u0131land\u0131rmalar da ayarlanabilir.
\n* Konu Silme<\/strong>: Art\u0131k kullan\u0131lmayan veya test ama\u00e7l\u0131 olu\u015fturulmu\u015f konular\u0131n silinmesi, kaynak israf\u0131n\u0131 \u00f6nler.
\n* Konu Listeleme ve Detaylar\u0131n\u0131 Alma<\/strong>: K\u00fcmedeki mevcut konular\u0131 ve her bir konunun partisyon say\u0131s\u0131, replikasyon fakt\u00f6r\u00fc, yap\u0131land\u0131rma ayarlar\u0131 gibi detaylar\u0131n\u0131 almak, k\u00fcmenin genel durumu hakk\u0131nda bilgi sa\u011flar.
\n* Konu Yap\u0131land\u0131rmas\u0131n\u0131 G\u00fcncelleme<\/strong>: Mevcut bir konunun baz\u0131 yap\u0131land\u0131rma ayarlar\u0131n\u0131 (\u00f6rn. retention.ms<\/code>) \u00e7al\u0131\u015fma zaman\u0131nda g\u00fcncellemek m\u00fcmk\u00fcnd\u00fcr.
\n* Partisyon Say\u0131s\u0131n\u0131 Art\u0131rma<\/strong>: Bir konunun mesaj hacmi artt\u0131\u011f\u0131nda, i\u015fleme kapasitesini art\u0131rmak i\u00e7in partisyon say\u0131s\u0131 art\u0131r\u0131labilir. \u00d6nemli Not<\/strong>: Partisyon say\u0131s\u0131 genellikle azalt\u0131lamaz. Bu nedenle ba\u015flang\u0131\u00e7ta dikkatli planlama yapmak \u00f6nemlidir.<\/p>\nimport org.apache.kafka.clients.admin.ConfigEntry;\nimport org.apache.kafka.clients.admin.Config;\nimport org.apache.kafka.clients.admin.AlterConfigsResult;\nimport org.apache.kafka.common.config.ConfigResource;\nimport org.apache.kafka.common.config.ConfigResource.Type;\n\nimport java.util.Collections;\nimport java.util.HashMap;\nimport java.util.Map;\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void updateTopicConfig(String topicName, String configKey, String configValue) {\n ConfigResource configResource = new ConfigResource(Type.TOPIC, topicName);\n ConfigEntry configEntry = new ConfigEntry(configKey, configValue);\n Config config = new Config(Collections.singletonList(configEntry));\n\n MapBroker Y\u00f6netimi ve Bilgi Alma<\/h4>\n
\n* Broker Konfig\u00fcrasyonlar\u0131n\u0131 Alma<\/strong>: Her bir broker’\u0131n yap\u0131land\u0131rma ayarlar\u0131n\u0131 (\u00f6rn. log.dirs<\/code>, listeners<\/code>) incelemek.
\n* Kontrol\u00f6r Broker’\u0131 Bulma<\/strong>: K\u00fcmenin lider broker’\u0131n\u0131 (controller) belirlemek, k\u00fcme y\u00f6netiminde \u00f6nemli bir rol oynar.<\/p>\nimport org.apache.kafka.clients.admin.DescribeClusterResult;\nimport org.apache.kafka.clients.admin.Node;\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void describeCluster() {\n DescribeClusterResult result = adminClient.describeCluster();\n try {\n System.out.println(\"K\u00fcme ID: \" + result.clusterId().get());\n System.out.println(\"Kontrol\u00f6r: \" + result.controller().get().id());\n System.out.println(\"Broker'lar:\");\n for (Node node : result.nodes().get()) {\n System.out.println(\" - ID: \" + node.id() + \", Host: \" + node.host() + \", Port: \" + node.port());\n }\n } catch (InterruptedException | ExecutionException e) {\n System.err.println(\"K\u00fcme bilgileri al\u0131n\u0131rken hata olu\u015ftu: \" + e.getMessage());\n }\n}<\/code><\/pre>\nT\u00fcketici Grubu (Consumer Group) Y\u00f6netimi<\/h4>\n
\n* T\u00fcketici Grubu Detaylar\u0131n\u0131 Alma<\/strong>: Belirli bir t\u00fcketici grubunun hangi konular\u0131 t\u00fcketti\u011fini, hangi partisyonlar\u0131 atand\u0131\u011f\u0131n\u0131 ve her bir partisyon i\u00e7in en son i\u015flenen offset’leri \u00f6\u011frenmek. Bu, t\u00fcketici gecikmesini (consumer lag) izlemek i\u00e7in hayati \u00f6neme sahiptir.
\n* T\u00fcketici Grubunun Offset’lerini S\u0131f\u0131rlama<\/strong>: Hata durumlar\u0131nda veya belirli bir noktadan yeniden i\u015fleme gerekti\u011finde, bir t\u00fcketici grubunun offset’lerini belirli bir zamana, en ba\u015fa, en sona veya belirli bir offset’e s\u0131f\u0131rlamak (seek) m\u00fcmk\u00fcnd\u00fcr. Bu i\u015flem dikkatli yap\u0131lmal\u0131d\u0131r, \u00e7\u00fcnk\u00fc veri kayb\u0131na veya tekrar i\u015flenmesine neden olabilir.<\/p>\nimport org.apache.kafka.clients.admin.ListConsumerGroupsResult;\nimport org.apache.kafka.clients.admin.ConsumerGroupListing;\nimport org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;\nimport org.apache.kafka.clients.admin.MemberAssignment;\nimport org.apache.kafka.clients.admin.MemberDescription;\nimport org.apache.kafka.clients.admin.ConsumerGroupDescription;\nimport org.apache.kafka.clients.admin.OffsetSpec;\nimport org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;\nimport org.apache.kafka.clients.admin.TopicPartitionReplica;\nimport org.apache.kafka.common.TopicPartition;\n\nimport java.util.HashSet;\nimport java.util.Set;\nimport java.util.Map;\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void listConsumerGroups() {\n ListConsumerGroupsResult result = adminClient.listConsumerGroups();\n try {\n System.out.println(\"Aktif T\u00fcketici Gruplar\u0131:\");\n for (ConsumerGroupListing listing : result.all().get()) {\n System.out.println(\" - \" + listing.groupId() + (listing.isSimpleConsumerGroup() ? \" (basit)\" : \"\"));\n }\n } catch (InterruptedException | ExecutionException e) {\n System.err.println(\"T\u00fcketici gruplar\u0131 listelenirken hata olu\u015ftu: \" + e.getMessage());\n }\n}\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void describeConsumerGroup(String groupId) {\n DescribeConsumerGroupsResult result = adminClient.describeConsumerGroups(Collections.singletonList(groupId));\n try {\n ConsumerGroupDescription description = result.all().get().get(groupId);\n if (description != null) {\n System.out.println(\"T\u00fcketici Grubu: \" + description.groupId());\n System.out.println(\" Durum: \" + description.state());\n System.out.println(\" Koordinat\u00f6r: \" + description.coordinator().id());\n System.out.println(\" \u00dcyeler (\" + description.members().size() + \"):\");\n for (MemberDescription member : description.members()) {\n System.out.println(\" - Client ID: \" + member.clientId() + \", Host: \" + member.host());\n for (TopicPartition tp : member.assignment().topicPartitions()) {\n System.out.println(\" - Konu: \" + tp.topic() + \", Partisyon: \" + tp.partition());\n }\n }\n }\n } catch (InterruptedException | ExecutionException e) {\n System.err.println(\"T\u00fcketici grubu '\" + groupId + \"' a\u00e7\u0131klan\u0131rken hata olu\u015ftu: \" + e.getMessage());\n }\n}\n\n\/\/ T\u00fcketici grubunun offsetlerini alma\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void listConsumerGroupOffsets(String groupId) {\n ListConsumerGroupOffsetsResult result = adminClient.listConsumerGroupOffsets(groupId);\n try {\n MapACL (Eri\u015fim Kontrol Listesi) Y\u00f6netimi<\/h4>\n
\n* ACL Silme<\/strong>: Art\u0131k ge\u00e7erli olmayan ACL’leri kald\u0131rmak.
\n* ACL Listeleme<\/strong>: K\u00fcmedeki t\u00fcm tan\u0131ml\u0131 ACL’leri veya belirli bir kaynakla ili\u015fkili ACL’leri listelemek.<\/p>\nimport org.apache.kafka.common.acl.AclBinding;\nimport org.apache.kafka.common.acl.AccessControlEntry;\nimport org.apache.kafka.common.acl.AclOperation;\nimport org.apache.kafka.common.acl.AclPermissionType;\nimport org.apache.kafka.common.resource.ResourcePattern;\nimport org.apache.kafka.common.resource.ResourcePatternFilter;\nimport org.apache.kafka.common.resource.ResourceType;\nimport org.apache.kafka.common.resource.PatternType;\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void createTopicAcl(String principal, String host, AclOperation operation, AclPermissionType permissionType, String topicName) {\n ResourcePattern resource = new ResourcePattern(ResourceType.TOPIC, topicName, PatternType.LITERAL);\n AccessControlEntry entry = new AccessControlEntry(principal, host, operation, permissionType);\n AclBinding aclBinding = new AclBinding(resource, entry);\n\n try {\n adminClient.createAcls(Collections.singletonList(aclBinding)).all().get();\n System.out.println(\"Konu '\" + topicName + \"' i\u00e7in ACL ba\u015far\u0131yla olu\u015fturuldu.\");\n } catch (InterruptedException | ExecutionException e) {\n System.err.println(\"ACL olu\u015fturulurken hata olu\u015ftu: \" + e.getMessage());\n }\n}\n\n\/\/ KafkaAdminClientExample s\u0131n\u0131f\u0131 i\u00e7inde bir metot\npublic void deleteTopicAcl(String principal, String topicName) {\n ResourcePatternFilter resourceFilter = new ResourcePatternFilter(ResourceType.TOPIC, topicName, PatternType.LITERAL);\n AccessControlEntry entry = new AccessControlEntry(principal, null, AclOperation.ANY, AclPermissionType.ANY); \/\/ null host, ANY operation\/permissionType ile t\u00fcm e\u015fle\u015fenleri sil\n AclBinding aclBinding = new AclBinding(resourceFilter, entry);\n\n try {\n adminClient.deleteAcls(Collections.singletonList(aclBinding.toFilter())).all().get();\n System.out.println(\"Konu '\" + topicName + \"' i\u00e7in ACL ba\u015far\u0131yla silindi.\");\n } catch (InterruptedException | ExecutionException e) {\n System.err.println(\"ACL silinirken hata olu\u015ftu: \" + e.getMessage());\n }\n}<\/code><\/pre>\nMetrik ve \u0130zleme<\/h4>\n
AdminClient<\/code> do\u011frudan zengin metrikler sunmasa da, Kafka broker’lar\u0131, \u00fcreticileri ve t\u00fcketicileri JMX (Java Management Extensions) arac\u0131l\u0131\u011f\u0131yla zengin metrikler yay\u0131mlar.<\/p>\n
\n* Kafka API’leri \u00dczerinden Temel Metrikler<\/strong>: KafkaProducer<\/code> ve KafkaConsumer<\/code> gibi istemciler, kendi i\u00e7 metriklerini (\u00f6rn. gecikme, throughput, hata oranlar\u0131) metrics()<\/code> metodu arac\u0131l\u0131\u011f\u0131yla sunarlar. Bu metrikler programatik olarak toplanarak \u00f6zel izleme \u00e7\u00f6z\u00fcmlerine entegre edilebilir.
\n* describeLogDirs<\/code><\/strong>: Bu metot, her bir broker’daki log dizinlerinin kullan\u0131m durumunu ve partisyonlar\u0131n bu dizinlere nas\u0131l da\u011f\u0131ld\u0131\u011f\u0131n\u0131 g\u00f6sterir. Disk alan\u0131 y\u00f6netimi i\u00e7in faydal\u0131d\u0131r.<\/p>\nGeli\u015fmi\u015f Y\u00f6netim Senaryolar\u0131 ve \u0130pu\u00e7lar\u0131<\/h3>\n
Hata Y\u00f6netimi ve Tekrar Deneme Mekanizmalar\u0131<\/h4>\n
AdminClient<\/code> \u00e7a\u011fr\u0131lar\u0131 i\u00e7in sa\u011flam hata y\u00f6netimi ve tekrar deneme (retry) mekanizmalar\u0131 uygulamak \u00f6nemlidir. KafkaFuture<\/code> nesneleri, whenComplete<\/code> veya exceptionally<\/code> gibi metotlarla hata i\u015fleme zincirleri olu\u015fturmaya olanak tan\u0131r. Ge\u00e7ici hatalar i\u00e7in \u00fcstel geri \u00e7ekilme (exponential backoff) stratejileri ile tekrar denemeler uygulanmal\u0131d\u0131r.<\/p>\nIdempotency ve \u0130\u015flem Garantileri<\/h4>\n
AdminClient<\/code>‘\u0131n baz\u0131 metotlar\u0131 do\u011fal olarak idempotenttir (\u00f6rn. createTopics<\/code> ayn\u0131 konuyu tekrar olu\u015fturmaya \u00e7al\u0131\u015f\u0131rken hata vermez), ancak di\u011ferleri i\u00e7in dikkatli olunmal\u0131d\u0131r.<\/p>\nG\u00fcvenlik<\/h4>\n
\n* SASL<\/strong>: \u0130stemcinin kimli\u011fini do\u011frulamak i\u00e7in SASL (Simple Authentication and Security Layer) mekanizmalar\u0131 (\u00f6rn. PLAIN, SCRAM) kullan\u0131lmal\u0131d\u0131r.
\n* ACL’ler<\/strong>: Y\u00f6netim istemcisinin sadece ihtiya\u00e7 duydu\u011fu kaynaklara ve i\u015flemlere eri\u015fim izni oldu\u011fundan emin olmak i\u00e7in ACL’ler do\u011fru bir \u015fekilde yap\u0131land\u0131r\u0131lmal\u0131d\u0131r. Y\u00f6netim istemcisi, AdminClient<\/code>‘\u0131n belirli i\u015flemleri ger\u00e7ekle\u015ftirebilmesi i\u00e7in ClusterAction<\/code> iznine sahip olmal\u0131d\u0131r.<\/p>\nAsenkron \u0130\u015flemler<\/h4>\n
AdminClient<\/code> API’si asenkron bir yap\u0131ya sahiptir ve KafkaFuture<\/code> nesneleri d\u00f6nd\u00fcr\u00fcr. Bu, birden fazla y\u00f6netimsel g\u00f6revi paralel olarak \u00e7al\u0131\u015ft\u0131rman\u0131za olanak tan\u0131r ve uygulaman\u0131z\u0131n engellenmesini \u00f6nler. CompletableFuture<\/code> ile birle\u015ftirilerek karma\u015f\u0131k asenkron i\u015f ak\u0131\u015flar\u0131 olu\u015fturulabilir.<\/p>\nKonfig\u00fcrasyon Y\u00f6netimi<\/h4>\n
AdminClient<\/code> ile yap\u0131lan baz\u0131 de\u011fi\u015fiklikler (\u00f6rn. konu yap\u0131land\u0131rmalar\u0131) k\u00fcme genelinde kal\u0131c\u0131d\u0131r. Bu de\u011fi\u015fikliklerin bir konfig\u00fcrasyon y\u00f6netim sistemi (\u00f6rn. Git, Ansible, Terraform) ile takip edilmesi, de\u011fi\u015fikliklerin izlenebilirli\u011fini ve geri al\u0131nabilirli\u011fini sa\u011flar.<\/p>\n\u00d6rnek Kullan\u0131m Senaryolar\u0131<\/h3>\n
Otomatik Test Ortam\u0131 Kurulumu<\/h4>\n
\/\/ Test \u00f6ncesi konular\u0131 olu\u015ftur\nkafkaAdminClient.createTopic(\"test-topic-1\", 1, (short) 1);\nkafkaAdminClient.createTopic(\"test-topic-2\", 3, (short) 1);\n\n\/\/ Test verisi g\u00f6nder\/t\u00fcket\n\/\/ Test sonras\u0131 konular\u0131 sil\nkafkaAdminClient.deleteTopic(\"test-topic-1\");\nkafkaAdminClient.deleteTopic(\"test-topic-2\");<\/code><\/pre>\nDinamik Mikroservis Konfig\u00fcrasyonu<\/h4>\n
Operasyonel Otomasyon<\/h4>\n
\n* Partisyon Yeniden Dengeleme<\/strong>: Broker’lar aras\u0131nda partisyonlar\u0131n dengeli da\u011f\u0131l\u0131m\u0131n\u0131 sa\u011flamak veya bir broker’\u0131n devre d\u0131\u015f\u0131 kalmas\u0131 durumunda partisyonlar\u0131 yeniden atamak.
\n* T\u00fcketici Gecikmesi Y\u00f6netimi<\/strong>: Belirli bir t\u00fcketici grubunun gecikmesi (lag) kritik bir e\u015fi\u011fi a\u015ft\u0131\u011f\u0131nda otomatik olarak uyar\u0131 g\u00f6ndermek veya duruma g\u00f6re offset’leri s\u0131f\u0131rlama gibi kurtarma i\u015flemleri \u00f6nermek.<\/p>\nS\u0131k Kar\u015f\u0131la\u015f\u0131lan Zorluklar ve \u00c7\u00f6z\u00fcmleri<\/h3>\n
AdminClient<\/code>‘\u0131n, hedef Kafka k\u00fcmesinin versiyonuyla uyumlu oldu\u011fundan emin olun.
\n* Da\u011f\u0131t\u0131k Sistem Hatalar\u0131<\/strong>: A\u011f kesintileri, broker \u00e7\u00f6kmesi veya lider se\u00e7imi gibi da\u011f\u0131t\u0131k sistem olaylar\u0131 y\u00f6netim \u00e7a\u011fr\u0131lar\u0131n\u0131 etkileyebilir. Sa\u011flam hata i\u015fleme ve tekrar deneme mant\u0131\u011f\u0131 bu durumlar\u0131 ele almak i\u00e7in kritik \u00f6neme sahiptir.
\n* Hatal\u0131 Konfig\u00fcrasyonlar<\/strong>: Yanl\u0131\u015f partisyon say\u0131s\u0131 veya replikasyon fakt\u00f6r\u00fc gibi hatal\u0131 konu konfig\u00fcrasyonlar\u0131 performans sorunlar\u0131na veya veri kayb\u0131na yol a\u00e7abilir. Konfig\u00fcrasyonlar\u0131 dikkatlice planlay\u0131n ve test edin.
\n* G\u00fcvenlik Konfig\u00fcrasyonlar\u0131<\/strong>: SSL\/TLS, SASL ve ACL’lerin do\u011fru \u015fekilde yap\u0131land\u0131r\u0131lmas\u0131 karma\u015f\u0131k olabilir. Yanl\u0131\u015f konfig\u00fcrasyonlar eri\u015fim sorunlar\u0131na veya g\u00fcvenlik a\u00e7\u0131klar\u0131na neden olabilir.<\/p>\nSonu\u00e7<\/h3>\n
AdminClient<\/code> gibi g\u00fc\u00e7l\u00fc API’ler sayesinde, konu olu\u015fturmadan ACL y\u00f6netimine kadar bir\u00e7ok kritik g\u00f6revi otomatikle\u015ftirmek ve \u00f6zelle\u015ftirilmi\u015f y\u00f6netim ara\u00e7lar\u0131 geli\u015ftirmek m\u00fcmk\u00fcnd\u00fcr. Java, Python, Go gibi farkl\u0131 dillerdeki istemci k\u00fct\u00fcphaneleri, geli\u015ftiricilere esneklik sunar.<\/p>\n