One Kafka is good, several are better

Hello everyone! My name is Alexander and I am an engineer on the team responsible for the development of centralized IT services used by the product teams at 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  . 

  
  
  
  
  
  
  
  
  
  

  , , . 

[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
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

  :

• . 

  
  
  
  
  

  , , , - , . , , - - ;

  
  
  
  
  

• .

  
  
  
  
  

  , - Kafka, .

  
  
  
  
  

     

  
  
  
  
  

, :

, ( ).

,  Kafka  , . 

Kafka

, , , .

 Grafana  : 

jmx-exporter –  Java Virtual Machine,

kafka-exporter, zookeeper-exporter – , ,

telegraf – ,  Kafka.

. , ,  kafka-minion exporter (https://github.com/cloudworkz/kafka-minion), , , , ..

 Kafka, , , , .  Kafdrop (https://github.com/obsidiandynamics/kafdrop).  Kafdrop,  CI pipeline,  OpenShift  : Kafdrop  nginx.  web UI  basic ,  nginx.

, , , (PostgreSQL Connector, MongoDB Kafka Connector), ksqlDB Kafka REST Proxy REST API.

, Kafka  , , .  Kafka  , . , , , . 

, - , .  RabbitMQ,  .