On Premise Kafka Kurulum

Windows

Örnek

Şöyle yaparız

1. KAFKA_HOME isimli bir ortam değişkeni tanımla. Buna bir değer ata örneğin C:\kafka_2.13–2.8.0

2. Path değişkenine "%KAFKA_HOME%\bin" değerini ilave et

Sonra şöyle yaparız

cd %KAFKA_HOME%

# Start ZooKeeper
.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

# Start Kafka
.\bin\windows\kafka-server-start.bat .\config\server.properties

WSL

Örnek

Şöyle yaparız

# Download latest kafka
wget https://downloads.apache.org/kafka/3.5.0/kafka_2.12-3.5.0.tgz

# Untar kafka and go to kafka folder
tar -xzf kafka_2.13-2.6.0.tgz
cd kafka_2.13-2.6.0

# Start Zookeeper on new tab
bin/zookeeper-server-start.sh config/zookeeper.properties

# Start kafka server on new tab
cd kafka_2.13-2.6.0
bin/kafka-server-start.sh config/server.properties


# Create topic as quickstart-events on new tab
cd kafka_2.13-2.6.0
bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

# Create event producer
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic quickstart-events
>>
Hi
>>

# Start event listener on topic quickstart-event
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic quickstart-events --from-beginning
>>
Hi
>>

# clean up
rm -rf /tmp/kafka-logs /tmp/zookeeper

Kafka Connect Mimarisi

Deployment Model

1. Standalone

2. Distributed 

Eğer dağıtık ise bir cluster şeklinde çalışır. Şeklen şöyle

Connector Mimarisi

Bileşenler şöyle

1. Connector Arayüzü

 Connector yazısına bakabilirsiniz

Açıklaması şöyle

Connectors are instances responsible for instantiating and defining/configuring a set of child processes called Tasks.

Açıklaması şöyle

Example to understand Connector and Tasks Difference

Say we have a Postgres database source connector which is responsible to pull data from 5 different tables into Kafka.

The connector would come up first and will figure out that there are 5 tables that need to be polled.

Based on that it will spin up new tasks and each task will have responsibility to poll data from a subset of tables.

2. Tasks

Açıklaması şöyle

Tasks are the most basic unit of pulling/pushing data into/out-of Kafka.

SourceTask : Poll işlemini yapar

SinkTask  : Put işlemini yapar

3. Workers

Açıklaması şöyle

As a user when you submit a certain connector configuration to the connect worker. This config would create the type of connector plugin and in addition to that some more plugin specific configs, a connector instance would be instantiated on one of the workers in the connect cluster.

in-sync replica - ISR

Giriş

Açıklaması şöyle

An in-sync replica is a replica that fully catches up with the leader in the last 10 seconds

Açıklaması şöyle

Kafka continues to leverage its legacy protocol, called ISR, to manage data replication within the cluster. 

RocksDB Nedir ?

Giriş

Açıklaması şöyle

RocksDB is an embedded database forked from Google’s LevelDB in 2012. It was initially created on Facebook by Dhruba Borthakur with the goal of improving performance for server workloads. Currently, RocksDB is developed and maintained by Meta.

Written in C++, RocksDB provides support for embedding in applications written in various languages like C, C++, Rust, Go, and Java through С binding. This flexibility allows developers to integrate RocksDB into their applications regardless of the programming language they are using.

Kafka Connect JdbcSinkConnector Sınıfı

Örnek

Şöyle yaparız

{
  "name": "postgres-sink",
  "config": {"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max":"1",
    "topics": "RESEARCH_AVE_BOOST",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "connection.url": "jdbc:postgresql://postgres:5432/students?user=postgres&password=postgres",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "true",
    "auto.create": "true",
    "auto.evolve": "true",
    "insert.mode": "upsert",
    "pk.fields": "RESEARCH",
    "pk.mode": "record_key"
  }
}

Docker Compose ve ksql

Örnek

Şöyle yaparız

version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.1.2
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:5.1.2
    depends_on:
      - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  schema-registry:
    image: confluentinc/cp-schema-registry:5.1.2
    hostname: schema-registry
    ports:
      - 8081:8081
    depends_on:
      - zookeeper
      - kafka
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper:2181
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

  ksql-server:
    image: confluentinc/cp-ksql-server:5.1.2
    depends_on:
      - kafka
      - schema-registry
    environment:
      KSQL_BOOTSTRAP_SERVERS: kafka:9092
      KSQL_LISTENERS: http://0.0.0.0:8088
      KSQL_KSQL_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      KSQL_KSQL_SERVICE_ID: ksql-server

Şöyle yaparız

docker run — network postgres-kafka-demo_default 
           — interactive — tty — rm 
           confluentinc/cp-ksql-cli:latest 
           http://ksql-server:8088

# To see your updates, a few settings need to be configured by first running:

set ‘commit.interval.ms’=’2000';
set ‘cache.max.bytes.buffering’=’10000000';
set ‘auto.offset.reset’=’earliest’;

Kafka Consumer Schema Registry Mesaj Okuma

Maven

Örnek

Şu satırı dahil ederiz. Schema Registry Sunucu için istemci ve Avro kullanıyoruz

<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-schema-registry-client</artifactId>
  <version>6.2.0</version>
</dependency>
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-avro-serializer</artifactId>
  <version>6.2.0</version>
</dependency>

Örnek

Şöyle yaparız. Toplam 5 tane özelliği atamak gerekiyor. 

VALUE_DESERIALIZER_CLASS_CONFIG : KafkaAvroDeserializer

SCHEMA_REGISTRY_URL_CONFIG : Schema Registry Sunucusu Adresi

@Configuration
public class KafkaConfig {
  @Value("${spring.kafka.bootstrap-servers}")
  private String bootstrapServers;
    
  @Value("${spring.kafka.schema-registry-url}")
  private String schemaRegistryUrl;
    
  @Bean
  public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
      KafkaAvroDeserializer.class);
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
    return props;
  }
  // Define your consumer factory, Kafka listener container factory, etc.
}

Örnek - Topic Naming Strategies

Şöyle yaparız

Map<String, Object> props = new HashMap<>();
props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "SCHEMA_REGISTRY_URL");
// Your prefferred deserializer
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class); 
// Your prefferred deserializer
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);

try {
  props.put("key.subject.name.strategy", 
    Class.forName("YOUR_NAMING_STRATEGY_FULLY_QUALIFIED_NAME"));
} catch (ClassNotFoundException e) {
  log.info("Key subject naming strategy not found {} ", e);
}
try {
  props.put("value.subject.name.strategy", 
    Class.forName("YOUR_NAMING_STRATEGY_FULLY_QUALIFIED_NAME"));
} catch (ClassNotFoundException e) {
  log.info("Value subject naming strategy not found {} ", e);
}

1. TopicNameStrategy

Açıklaması şöyle

This is the default strategy and is implemented if no strategy has been specified. In this strategy, we are expected to create the subjects with appended suffixes such as -key and -value.

So, if you are using schema registry and if you have a topic named student_information on your kafka broker, then you should have subjects named student_information-key and student_information-value with their respective schema definitions in schema registry.

Örnek

Şöyledir

{
  "subject":"student_information-key",
  "version":1,
  "id":1,
  "schema":"\"string\""
}

{
  "subject":"student_information-value",
  "version":1,
  "id":1,
  "schema":"
    {
     \"type\":\"record\",
      \"name\":\"StudentDTO\",
      \"namespace\":\"com.example.demo.models\",
      \"fields\":[
        {\"name\":\"firstname\",
         \"type\":[\"null\",\"string\"]
        },
        {\"name\":\"lastname\",\
          "type\":[\"null\",\"string\"]
        },
        {\"name\":\"age\",
         \"type\":[\"null\",\"int\"]
        }
   ]
  }"
}

2. RecordNameStrategy

Açıklaması şöyle

That is also called as fully qualified record name.

Şöyledir

com.example.demo.models.StudentDTO

3. TopicRecordNameStrategy

Açıklaması şöyle

In this strategy, the topic name and the fully qualified record name are combined together and are suffixed with -key and -value.

Şöyledir

student_information-com.example.demo.models.StudentDTO-key
student_information-com.example.demo.models.StudentDTO-value