🆓 👩🏾‍🔬 👻 Satu Kafka bagus, beberapa lebih baik 👕 🦃 🔸

Halo semuanya! Nama saya Alexander dan saya adalah insinyur di tim yang bertanggung jawab atas pengembangan layanan TI terpusat yang digunakan oleh tim produk di X5 Retail Group.

Apache Kafka , . , , , Kafka, , , , .

14 (1 13 ) 15 .

Kafka

, Kafka – Elasticsearch.

- 5
– 179
consumer – 77
^[1] – 555.1

_{[1] 90}

. , , , : , , , ( ). , Debug Trace. , , , - , , , . , X5 . - , 600 , .

, .

, , Elasticsearch . , , , . , , , . , , . , , .

Apache Kafka :

.

Kafka – acknowledgements. acks , () producer . , acks, , , , , . , , , Kafka;
. ( Logstash, Kafka) - , ;
.

Kafka , , retention. , - Elasticsearch ;
.

Kafka, . , , producer .

, , .

#	vCPU	RAM	Storage^[2]	Kafka	Kafka Uptime	Zookeeper	ZK Uptime
1	4	16	290	+	11	+	15
2	4	16	270	+	11	+	15
3	4	16	290	+	11	+	15
4	4	16	270	+	4	-	-
5	4	16	270	+	11	-	-

[2] – , Kafka

, 2 , 1-3, 4- , , - .

, Kafka Security Manager (https://github.com/simplesteph/kafka-security-manager). ACL. :

User:projectprodwrite@srelogs,Topic,PREFIXED,projectprod,Create,Allow,
User:projectprodread@srelogs,Topic,PREFIXED,projectprod,Read,Allow,
User:projectprodread@srelogs,Group,PREFIXED,projectprod,All,Allow,

User – CN , ,

srelogs – ,

Topic/Group – , ,

PREFIXED/LITERAL – , Kafka ( ),

project_prod – , .

Producer/consumer SSL , Vault.

Ansible. ( ), , , ( Elasticsearch, Logstash)

, , , Kafka auto.create.topics.enable=True

, Kafka . , , Elasticsearch .

, , , ( ), dev , , .

Kafka , Logstash ( ) , Kafka, , . , consumer , Elasticsearch Logstash.

, , Kafka . , , – , , .

, , , , . «» Kafka. :

auto.create.topics.enable=True
delete.topic.enable=True
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connection.timeout.ms=6000
auto.leader.rebalance.enable=true

, Ansible , , X5 .

, , . , :

- name: 01|Set Kafka replication factor
  set_fact:
    kafka_cfg_default_replication_factor: "{{  kafka_cfg_default_replication_factor | default(kafka_hosts|length) }}"
    kafka_cfg_offsets_topic_replication_factor: "{{ kafka_cfg_offsets_topic_replication_factor | default(kafka_hosts|length) }}"
    kafka_cfg_transaction_state_log_replication_factor: "{{ kafka_cfg_transaction_state_log_replication_factor | default(kafka_hosts|length) }}"
  run_once: True

- name: 01|Set kafka ISR
  set_fact:
    kafka_cfg_min_insync_replicas: "{{ kafka_cfg_min_insync_replicas | default([kafka_cfg_default_replication_factor|int - 1 , 1] | max) }}"
    kafka_cfg_transaction_state_log_min_isr: "{{ kafka_cfg_transaction_state_log_min_isr | default([kafka_cfg_transaction_state_log_replication_factor|int - 1 , 1] | max) }}"
  run_once: True

30 , :

log.retention.hours=720

, , - . :

Project.yml
---
project: name
. . .
kafka_scala_version: "2.11"
kafka_zk_chroot: '/'
kafka_enable_protocol: ['PLAINTEXT']
kafka_cfg_default_replication_factor: 2
kafka_cfg_log_retention_hours: 6
kafka_cfg_log_segment_bytes: 52428800

, , SSL . kafkaenableprotocol: ['SSL'], , .

- name: Lookup for ssl data in Vault
  set_fact:
    jks_b64: "{{ lookup('hashi_vault', 'secret=sre/{{ env }}/{{ project }}/kafka/{{ inventory_hostname }}:kafka.keystore.jks.b64') }}"
- name: Copy keystore data from Vault
  copy:
    dest: "/opt/kafka/ssl/{{ inventory_hostname }}/kafka.keystore.jks"
    content: "{{ jks_b64 | b64decode }}"

Kafka Zookeeper , . Kafka, Ansible :

[Unit]
Description=Kafka Daemon
After=zookeeper.service

[Service]
Type=simple
User={{ kafka_user }}
Group={{ kafka_group }}
LimitNOFILE={{ kafka_nofiles_limit }}
Restart=on-failure
EnvironmentFile=/etc/default/kafka
ExecStart={{ kafka_bin_path }}/kafka-server-start.sh {{ kafka_config_path }}/server.properties
ExecStop={{ kafka_bin_path }}/kafka-server-stop.sh
WorkingDirectory={{ kafka_bin_path }}

[Install]
WantedBy=multi-user.target