MS Lesson23: Apache Kafka

                                                            Fundamental bilikler

 

1. Broker - Kafka-nin ishlediyi server prosesidir. Broker poct merkezi kimidir. Broker mesajlari qebul edir, mesajlari diske yazir, consumerlere istedikde gonderir. 

Real production-da hecvaxt 1 broker olmur, bir nece broker birlikde Cluster-i teshkil edir. 
Bir nece brokerin olmasi:
* Fault Tolerance : 1 broker cokse digerleri ishleyir
* Scalability : yuk bolunur, performans artir
* Replication : data kopyalanir, itmir

Her brokerin unikal ID-si var. Cluster-de diger brokerlerler emekdashliq edir. 

2. Topic - Kafka-da melumatlarin saxlandigi mentiqi kanal (log)dir. 
Poct merkezinde (Broker) mektublar kateqoriyalara bolunur. 

Topic-in xususiyyetleri:
* Unikal adi olur
* Partitions : nece hisseye bolunur (default 1)
* Replication factor : nece broker-de kopyalanir
* Retention period : melumat nece muddet saxlanir (default 7 gun)

Retention vacib meqamlardan biridir. Kafka diger message brokerlerden (RabbitMQ kimi) bu cehetden ferqlenir.

RabbitMQ: consumer mesaji oxudu -> mesaj silinir
Kafka: consumer mesaji oxudu -> mesaj qalir

Bu ne demekdir? 
Eyni melumati 10 ferqli Consumer oxuya biler.
Yeni bir servis qoshulsa, kecmish melumatlarida oxuya biler

3. Partition - bunu supermarketin kassasi kimi fikirleshek, 1 kassada 1000 nefer novbede gozlese bu zaman sistem cox yavash ishleyer, lakin 10 kassa olsa 1000 neferi 10 bour ve suret artir. 

Topic-i bir nece fiziki hisseye bolursen - her bir hisse musteqil ishleyir. 

Her partition:
* Musteqil bir log faylidir - diskde ayrica saxlanilir
* Ordered - oz icindeki mesajlar ardicildir
* Immutable - yazilan mesaj deyishdirilmir

Bes neye gore partition lazimdir?

* Paralellik (Parallelism)
1 Partition - yalniz 1 Consumer oxuya biler
3 Partition - 3 Consumer parallel oxuya biler.

* Scalibility 
Bugun: 10000 mesaj/saniye  -> 3 partition kifayetdir
Sabah: 100000 mesaj/saniye  -> 30 partition elave et

* Fault tolerance
Her partition ferqli brokerde replikasiya olunur:

Partition 0 → Leader: Broker 1, Replica: Broker 2 Partition 1 → Leader: Broker 2, Replica: Broker 3 Partition 2 → Leader: Broker 3, Replica: Broker 1 Broker 1 çökdü? → Broker 2 dərhal LEADER olur ✅ Data itirilmir!

Bes mesajlar hansi partition-a dushur?

// 3 ssenari var: // 1. Key yoxdur → Round Robin (növbəli bölünmə) producer.send("order-created", message); // msg1 → P0, msg2 → P1, msg3 → P2, msg4 → P0 ... // 2. Key var → eyni key həmişə eyni partition-a düşür producer.send("order-created", "userId-123", message); // "userId-123" olan BÜTÜN mesajlar → həmişə P1-ə // 3. Manual → sən özün seçirsən producer.send("order-created", 2, message); // Bu mesaj → P2-yə

Sual: Niyə eyni key-i eyni partition-a göndəririk? Cavab: ORDERING (sıra) zəmanəti üçün! ❌ Key olmadan: userId-123 → "sifariş verildi" → Partition 0 userId-123 → "ödəniş edildi" → Partition 2 userId-123 → "çatdırıldı" → Partition 1 Consumer bu 3 mesajı FƏRQLI sırada oxuya bilər! ❌ ✅ Key ilə (userId-123): userId-123 → "sifariş verildi" → Partition 1 (offset 0) userId-123 → "ödəniş edildi" → Partition 1 (offset 1) userId-123 → "çatdırıldı" → Partition 1 (offset 2) Consumer həmişə DÜZGÜN sırada oxuyur! ✅

Architect qaydası: Partition sayı = max(gözlənilən Consumer sayı, throughput tələbi) Məsələn: - 6 Consumer paralel işləyəcək → minimum 6 partition - Gələcəkdə böyüyəcək → 12 partition qoy (2x ehtiyat) ⚠️ Partition sayı artırıla bilər ⚠️ Amma AZALDILA BİLMƏZ — bunu həmişə yadda saxla! ⚠️ Çox partition da problem yaradır (metadata yükü)

                                                                    Docker compose yaml fayli

services:

  kafka-1:

    image: apache/kafka:3.9.0

    container_name: kafka-1

    ports:

      - "9092:9092"

      - "29092:29092"

    environment:

      KAFKA_NODE_ID: 1

      CLUSTER_ID: 97exVv29T2icAhT2SNEEDQ

      KAFKA_PROCESS_ROLES: broker,controller

      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9091,2@kafka-2:9091,3@kafka-3:9091

      KAFKA_LISTENERS: PLAINTEXT://:9090,CONTROLLER://:9091,EXTERNAL://:9092,EXTERNAL_CLIENT://:29092

      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:9090,EXTERNAL://kafka-1:9092,EXTERNAL_CLIENT://localhost:29092

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL_CLIENT:PLAINTEXT

      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT

      KAFKA_LOG_DIRS: /var/lib/kafka/data

    healthcheck:

      test: ["CMD", "/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9090", "--list"]

      interval: 10s

      timeout: 5s

      retries: 5

      start_period: 30s

    volumes:

      - ./volumes/kafka/server-1:/var/lib/kafka/data

  kafka-2:

    image: apache/kafka:3.9.0

    container_name: kafka-2

    ports:

      - "9094:9094"

      - "29094:29094"

    environment:

      KAFKA_NODE_ID: 2

      CLUSTER_ID: 97exVv29T2icAhT2SNEEDQ

      KAFKA_PROCESS_ROLES: broker,controller

      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9091,2@kafka-2:9091,3@kafka-3:9091

      KAFKA_LISTENERS: PLAINTEXT://:9090,CONTROLLER://:9091,EXTERNAL://:9094,EXTERNAL_CLIENT://:29094

      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:9090,EXTERNAL://kafka-2:9094,EXTERNAL_CLIENT://localhost:29094

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL_CLIENT:PLAINTEXT

      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT

      KAFKA_LOG_DIRS: /var/lib/kafka/data

    healthcheck:

      test: ["CMD", "/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9090", "--list"]

      interval: 10s

      timeout: 5s

      retries: 5

      start_period: 30s

    volumes:

      - ./volumes/kafka/server-2:/var/lib/kafka/data

  kafka-3:

    image: apache/kafka:3.9.0

    container_name: kafka-3

    ports:

      - "9096:9096"

      - "29096:29096"

    environment:

      KAFKA_NODE_ID: 3

      CLUSTER_ID: 97exVv29T2icAhT2SNEEDQ

      KAFKA_PROCESS_ROLES: broker,controller

      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9091,2@kafka-2:9091,3@kafka-3:9091

      KAFKA_LISTENERS: PLAINTEXT://:9090,CONTROLLER://:9091,EXTERNAL://:9096,EXTERNAL_CLIENT://:29096

      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-3:9090,EXTERNAL://kafka-3:9096,EXTERNAL_CLIENT://localhost:29096

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL_CLIENT:PLAINTEXT

      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT

      KAFKA_LOG_DIRS: /var/lib/kafka/data

    healthcheck:

      test: ["CMD", "/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9090", "--list"]

      interval: 10s

      timeout: 5s

      retries: 5

      start_period: 30s

    volumes:

      - ./volumes/kafka/server-3:/var/lib/kafka/data

  kafka-ui:

    image: provectuslabs/kafka-ui:v0.7.2

    container_name: kafka-ui

    ports:

      - "9999:8080"

    environment:

      KAFKA_CLUSTERS_0_NAME: local

      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-1:9090,kafka-2:9090,kafka-3:9090

    depends_on:

      kafka-1:

        condition: service_healthy

      kafka-2:

        condition: service_healthy

      kafka-3:

        condition: service_healthy

*Konteynerin daxiline girmek ucun: docker exec -it kafka-1 bash

*Topic yaratmaq ucun: ./kafka-topics.sh --bootstrap-server localhost:9092, localhost:9094, localhost:9094 --create --topic test-topic --partitions 3 --replication-factor 3

*Topic haqqinda melumat almaq ucun: ./kafka-topics.sh \

  --bootstrap-server localhost:9092 \

  --describe \

  --topic test-topic

*Mesaj produce elemek ucun: ./kafka-console-producer.sh \

  --bootstrap-server kafka-1:9090,kafka-2:9090,kafka-3:9090 \

  --topic test-topic \

  --property parse.key=true \

  --property key.separator=:

*Mesaj consume elemek ucun: ./kafka-console-consumer.sh \

  --bootstrap-server kafka-1:9090,kafka-2:9090,kafka-3:9090 \

  --topic test-topic \

  --from-beginning \

  --property print.key=true \

  --property key.separator=:

* Eger Group id teyin etmesek o zaman Round Robin alqoritmi ile ferqli partitionlara dushecek. 

./kafka-console-producer.sh \

  --bootstrap-server kafka-1:9090,kafka-2:9090,kafka-3:9090 \

  --topic test-topic \

  --producer-property linger.ms=0 \

  --producer-property batch.size=1

                                                      Spring Boot project Kafka Producer

Step1: yaml fayli konfiqurasiya etmek:

server:
  port: 0

spring:
  application:
    name: kafka-producer
  kafka:
    producer:
      bootstrap-servers: localhost:9092,localhost:9094,localhost:9096
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

Step2: Kafka konfiq fayli yaratmaq:

package az.etibarli.kafkaproducer.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

import java.util.Map;

@Configuration
public class KafkaConfig {

    @Bean
    public NewTopic studentEventsTopic() {
        return TopicBuilder.name("student-events-topic")
                .partitions(3)
                .replicas(3)
                .configs(Map.of("min.insync.replicas", "2"))
                .build();
    }

}

Step3: Sade bir model yaratmaq:

package az.etibarli.kafkaproducer.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Student {

    private Long id;
    private String name;
    private int age;

}

Step4: Sade bir service yaratmaq:

package az.etibarli.kafkaproducer.service;

import az.etibarli.kafkaproducer.model.Student;
import lombok.RequiredArgsConstructor;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class StudentProducerService {

    private static final String TOPIC = "student-events-topic";

    private final KafkaTemplate<String, Student> kafkaTemplate;

    public void send(Student student) {
        kafkaTemplate.send(TOPIC, String.valueOf(student.getId()), student);
    }

}

Step5: Sade bir controller yaratmaq:

package az.etibarli.kafkaproducer.controller;

import az.etibarli.kafkaproducer.model.Student;
import az.etibarli.kafkaproducer.service.StudentProducerService;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api/students")
@RequiredArgsConstructor
public class StudentController {

    private final StudentProducerService studentProducerService;

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public void createStudent(@RequestBody Student student) {
        studentProducerService.send(student);
    }

}

* Eger her hansisa bir serializasiya problemi bash verse o zaman dependency kimi bu elave edilmelidir:

        <dependency>

            <groupId>com.fasterxml.jackson.core</groupId>

            <artifactId>jackson-databind</artifactId>

        </dependency>

                                                    Indi ise biz sinxron ve asinxron ferqine baxaq

Test case1:

package az.etibarli.kafkaproducer.model;

public record StudentSendResponse(
        String mode,
        Long studentId,
        String message,
        Integer partition,
        Long offset
) {

    public static StudentSendResponse async(Long studentId) {
        return new StudentSendResponse(
                "ASYNC",
                studentId,
                "Message is being sent in the background. Check application logs for result.",
                null,
                null
        );
    }

    public static StudentSendResponse sync(Long studentId, int partition, long offset) {
        return new StudentSendResponse(
                "SYNC",
                studentId,
                "Message sent successfully.",
                partition,
                offset
        );
    }

}

package az.etibarli.kafkaproducer.service;

import az.etibarli.kafkaproducer.model.Student;
import az.etibarli.kafkaproducer.model.StudentSendResponse;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Slf4j
@Service
@RequiredArgsConstructor
public class StudentProducerService {

    private static final String TOPIC = "student-events-topic";

    private final KafkaTemplate<String, Student> kafkaTemplate;

    public StudentSendResponse sendAsync(Student student) {
        CompletableFuture<SendResult<String, Student>> future =
                kafkaTemplate.send(TOPIC, String.valueOf(student.getId()), student);

        future.whenComplete((result, exception) -> {
            if (exception != null) {
                log.error("ASYNC - Failed to send message for student {}: {}",
                        student.getId(), exception.getMessage());
            } else {
                RecordMetadata metadata = result.getRecordMetadata();
                log.info("ASYNC - Message sent successfully. topic={}, partition={}, offset={}",
                        metadata.topic(), metadata.partition(), metadata.offset());
            }
        });

        log.info("ASYNC - Request accepted immediately. studentId={}", student.getId());
        return StudentSendResponse.async(student.getId());
    }

    public StudentSendResponse sendSync(Student student) {
        CompletableFuture<SendResult<String, Student>> future =
                kafkaTemplate.send(TOPIC, String.valueOf(student.getId()), student);

        SendResult<String, Student> result = future.join();

        RecordMetadata metadata = result.getRecordMetadata();
        log.info("SYNC - Message sent successfully. topic={}, partition={}, offset={}",
                metadata.topic(), metadata.partition(), metadata.offset());

        return StudentSendResponse.sync(student.getId(), metadata.partition(), metadata.offset());
    }

}

package az.etibarli.kafkaproducer.controller;

import az.etibarli.kafkaproducer.model.Student;
import az.etibarli.kafkaproducer.model.StudentSendResponse;
import az.etibarli.kafkaproducer.service.StudentProducerService;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api/students")
@RequiredArgsConstructor
public class StudentController {

    private final StudentProducerService studentProducerService;

    @PostMapping("/async")
    @ResponseStatus(HttpStatus.ACCEPTED)
    public StudentSendResponse createStudentAsync(@RequestBody Student student) {
        return studentProducerService.sendAsync(student);
    }

    @PostMapping("/sync")
    @ResponseStatus(HttpStatus.CREATED)
    public StudentSendResponse createStudentSync(@RequestBody Student student) {
        return studentProducerService.sendSync(student);
    }

}

Test case2:

package az.etibarli.kafkaproducer.model;

public record StudentSendResponse(
        String mode,
        Long studentId,
        String message,
        Integer partition,
        Long offset
) {

    public static StudentSendResponse async(Long studentId) {
        return new StudentSendResponse(
                "ASYNC",
                studentId,
                "HTTP response returned immediately. Kafka send continues in background. Check logs.",
                null,
                null
        );
    }

    public static StudentSendResponse sync(Long studentId, int partition, long offset) {
        return new StudentSendResponse(
                "SYNC",
                studentId,
                "Message sent successfully.",
                partition,
                offset
        );
    }

    public static StudentSendResponse syncFailed(Long studentId, String error) {
        return new StudentSendResponse(
                "SYNC",
                studentId,
                "Failed to send message: " + error,
                null,
                null
        );
    }

    public boolean isSuccess() {
        return message.startsWith("Message sent successfully");
    }

}

package az.etibarli.kafkaproducer.service;

import az.etibarli.kafkaproducer.model.Student;
import az.etibarli.kafkaproducer.model.StudentSendResponse;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Slf4j
@Service
@RequiredArgsConstructor
public class StudentProducerService {

    private static final String TOPIC = "student-events-topic";

    private final KafkaTemplate<String, Student> kafkaTemplate;

    public StudentSendResponse sendAsync(Student student) {
        CompletableFuture.runAsync(() -> {
            try {
                CompletableFuture<SendResult<String, Student>> future =
                        kafkaTemplate.send(TOPIC, String.valueOf(student.getId()), student);

                future.whenComplete((result, exception) -> {
                    if (exception != null) {
                        log.error("ASYNC - Failed to send message for student {}: {}",
                                student.getId(), exception.getMessage());
                    } else {
                        RecordMetadata metadata = result.getRecordMetadata();
                        log.info("ASYNC - Message sent successfully. topic={}, partition={}, offset={}",
                                metadata.topic(), metadata.partition(), metadata.offset());
                    }
                });
            } catch (Exception exception) {
                log.error("ASYNC - Failed to send message for student {}: {}",
                        student.getId(), exception.getMessage());
            }
        });

        log.info("ASYNC - HTTP response returned immediately. studentId={}", student.getId());
        return StudentSendResponse.async(student.getId());
    }

    public StudentSendResponse sendSync(Student student) {
        try {
            CompletableFuture<SendResult<String, Student>> future =
                    kafkaTemplate.send(TOPIC, String.valueOf(student.getId()), student);

            SendResult<String, Student> result = future.join();

            RecordMetadata metadata = result.getRecordMetadata();
            log.info("SYNC - Message sent successfully. topic={}, partition={}, offset={}",
                    metadata.topic(), metadata.partition(), metadata.offset());

            return StudentSendResponse.sync(student.getId(), metadata.partition(), metadata.offset());
        } catch (Exception exception) {
            log.error("SYNC - Failed to send message for student {}: {}",
                    student.getId(), exception.getMessage());
            return StudentSendResponse.syncFailed(student.getId(), exception.getMessage());
        }
    }

}

package az.etibarli.kafkaproducer.controller;

import az.etibarli.kafkaproducer.model.Student;
import az.etibarli.kafkaproducer.model.StudentSendResponse;
import az.etibarli.kafkaproducer.service.StudentProducerService;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api/students")
@RequiredArgsConstructor
public class StudentController {

    private final StudentProducerService studentProducerService;

    @PostMapping("/async")
    public ResponseEntity<StudentSendResponse> createStudentAsync(@RequestBody Student student) {
        return ResponseEntity.status(HttpStatus.ACCEPTED)
                .body(studentProducerService.sendAsync(student));
    }

    @PostMapping("/sync")
    public ResponseEntity<StudentSendResponse> createStudentSync(@RequestBody Student student) {
        StudentSendResponse response = studentProducerService.sendSync(student);
        HttpStatus status = response.isSuccess() ? HttpStatus.CREATED : HttpStatus.SERVICE_UNAVAILABLE;
        return ResponseEntity.status(status).body(response);
    }

}

                                                                min.insync.replicas vs acks

*** min.insync.replicas - Kafkanin qaydasi. Topic-a mesaj yazmaq ucun min nece broker ishlemelidir

3 Broker var A, B, C

min.insync.replicas = 2

Kafka deyir : "En azi 2 broker online olmalidir, yoxsa mesaj qebul etmirem".  Bu bir nov qapidir - kifayet qeder broker yoxdursa, Kafka mesaju umumiyyetle qebul etmir. 

*** acks - Producerin qaydasi. "Mesaj gonderende ne qeder gozleyim?". 

acks = all

Producer deyir : "Mesaj heqiqeten yazilana qeder gozleyecem".

acks = 1

Producer deyir: "Leader 'aldim' dese, kifayet edir".

acks = 0

Producer deyir: "Gonderdim, cavab gozlemirem".

*** Analogiya

3 poct shirketi var: A, B, C. Mektub gonderirsen, her shobede suret saxlanilir. 

min.insync.replicas - poct sistemine lazimdir. "Menim 2 shobem aciq olmalidir, yoxsa mektub qebul etmirem".

acks - gonderene lazimdir. "Mektub catana qeder gozleyim, yoxsa gedim?".

*** Qizil qayda min.insync.replicas yalniz acks= all duzgun sayilir

*** resplicas - eyni datani bir nece brokerde saxlamaq ucun nezerde tutulub. 

Mes, replicas = 3 o demekdirki her partition 3 brokerde kopyalanir. 

Broker sayi >= resplicas. Yeni broker sayi hemishe replika sayindan ya boyuk olmalidir ya da beraber. 

Mes, biz 3 brokerli sistemde yazsaq ki, replicas = 4 bu zaman topic yaranmayacaq ve exception atacaq.

*** max.in.flight.requests.per.connection: 5

Producer mesaj gonderende cavab gozlemeden nece mesaji eyni anda "yolda" saxlaya bilerem? 

max.in.flight.requests=5:

Producer → [msg1] [msg2] [msg3] [msg4] [msg5] → Broker

           ←←←←←←←←←←←←←←←←←←←←←←←←←←←←←

           hamisinin cavabini gozleyir

msg6 gondermek istəyir?

→ Evvelki 5-den biri cavab gelmelidir!

→ Gelenden sonra msg6 gonderilir

Bu deyer 5 den boyuk olduqda idempotence = true ile ishlemir. Kafka yalniz 5 qeder destekleyir.

* idempotency = true arxa planda nece ishleyir:
1. Producer-e unikal ID verilir
2. Her mesaja SEQ nomresi elave olunur
3. Broker PID + SEQ nomresini yadda saxlayir
4. Eyni PID + SEQ gelende saxlamir, OK qaytarir
5. SEQ ardicil deyilse xeta verir

step6: 

package az.etibarli.demokafkaproducer.config;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import az.etibarli.demokafkaproducer.kafka.dto.StudentDto;

@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.producer.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.producer.key-serializer}")
    private String keySerializer;

    @Value("${spring.kafka.producer.value-serializer}")
    private String valueSerializer;

    @Value("${spring.kafka.producer.acks}")
    private String acks;

    @Value("${spring.kafka.producer.properties.delivery.timeout.ms}")
    private String deliveryTimeout;

    @Value("${spring.kafka.producer.properties.linger.ms}")
    private String linger;

    @Value("${spring.kafka.producer.properties.request.timeout.ms}")
    private String requestTimeout;

    Map<String, Object> producerConfigs() {
        Map<String, Object> config = new HashMap<>();

        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        config.put(ProducerConfig.ACKS_CONFIG, acks);
        config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeout);
        config.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeout);

        return config;
    }

    @Bean
    ProducerFactory<String, StudentDto> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    KafkaTemplate<String, StudentDto> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public NewTopic studentEventsTopic() {
        return TopicBuilder.name("ms34-topic")
                .partitions(3)
                .replicas(3)
                .configs(Map.of("min.insync.replicas", "2"))
                .build();
    }

}

===========================================================================

                                                                Kafka Consumer

Qayda:

Consumer sayi < Partition sayi  →  bezi consumerler birden cox partition oxuyur

Consumer sayı = Partition sayı  →  ideal bolgu

Consumer sayı > Partition sayı  →  bəzi consumer-ler bosh oturur