Kafka Connect SourceConnector Sınıfı

Giriş

Şu satırı dahil ederiz. Soyut bir sınıftır

import org.apache.kafka.connect.source.SourceConnector;

config metodu

Medotun için şuna benzer

@Override
public ConfigDef config() {
    return CONFIG_DEF;
}

Açıklaması şöyle

create fewer tasks if it cannot achieve this level of parallelism.aaThe ConfigDef exposes what configuration you require for your connector and hopefully how it manipulates your connector.

context metodu

Metodu şöyle

@Override
protected SourceConnectorContext context() {
  return (SourceConnectorContext) context;
}

taskConfigs metodu

İmzası şöyle

public abstract List<Map<String, String>> taskConfigs(int maxTasks);

Açıklaması şöyle

The maximum number of tasks that should be created for this connector. The connector may create fewer tasks if it cannot achieve this level of parallelism.

maxTasks alanı kullanılan connector tipine göre bazen dikkate alınıyor bazen alınmıyor. 

Örneğin io.debezium.connector.mysql.MySqlConnector, io.debezium.connector.postgresql.PostgresConnector bu değeri dikkate almıyor. Açıklaması şöyle

Q : How can I increase the tasks.max for debezium sql connnector?

A : It's not possible.

The database bin log must be read sequentially by only one task.

Run multiple connectors for different tables if you want to distribute workload

Ama mesela io.confluent.connect.jdbc.JdbcSourceConnector dikkate alıyor. O yüzden dokümantasyona bakmak lazım

Docker ve Debezium

Örnek

Şöyle yaparız

docker run -it --name connect --net=host -p 8083:8083 \
-e GROUP_ID=1 \
-e CONFIG_STORAGE_TOPIC=my-connect-configs \
-e OFFSET_STORAGE_TOPIC=my-connect-offsets \
-e BOOTSTRAP_SERVERS=localhost:9092 \
-e CONNECT_TOPIC_CREATION_ENABLE=true \
-v ~/.aws/config:/kafka/.aws/config \
-v ~/Downloads/kafka-connect-iceberg-sink-0.1.3-shaded.jar:/kafka/connect/kafka-connect-iceberg-sink-0.1.3-shaded.jar \
debezium/connect

Docker Compose ve Kafka Connect

Bir örnek burada

Kafka Producer KafkaProducer Sınıfı transaction metodları

Giriş

Kafka Producer Exactly-once yazısında örnek var

Örnek - Chained Transaction

Bir örnek burada

mutual Transport Layer Security - mTLS

Giriş

Açıklaması şöyle

mTLS enables clients to authenticate servers, and servers to reciprocally authenticate clients.

Kafka supports other authentication mechanisms, like OAuth, or Salted Challenge Response Authentication Mechanism (SCRAM), but we chose mTLS because it is able to verify the peer’s identity offline. This verification ability means that systems do not need an active connection to an authentication server to ascertain the identity of a peer. This enables operating in disparate network environments, where all parties do not necessarily have access to such a central authority.

Producer Ayarları - Serializer

Örnek

Şöyle yaparız

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);

Docker ve Kafka Connect

Örnek

Şöyle yaparız

docker pull confluentinc/cp-kafka-connect:latest

docker run \
  -it \
  --rm \
  --name es-sink-connector \
  -e CONNECT_BOOTSTRAP_SERVERS=kafka:9092 \
  -e CONNECT_REST_PORT=8083 \
  -e CONNECT_GROUP_ID="connect-cluster" \
  -e CONNECT_CONFIG_STORAGE_TOPIC="connect-configs" \
  -e CONNECT_OFFSET_STORAGE_TOPIC="connect-offsets" \
  -e CONNECT_STATUS_STORAGE_TOPIC="connect-status" \
  -e CONNECT_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_REST_ADVERTISED_HOST_NAME="localhost" \
  -e CONNECT_PLUGIN_PATH="/usr/share/java,/etc/kafka-connect/jars" \
  -p 8083:8083 
  confluentinc/cp-kafka-connect:latest