Sink Connectors

Sink connector читает данные из Kafka и записывает их во внешнюю систему. Это зеркальная задача по отношению к source connector: вместо чтения из внешней системы и записи в Kafka — чтение из Kafka и запись в хранилище, поисковый движок или объектное хранилище.

Жизненный цикл sink connector

Понимание жизненного цикла sink connector объясняет семантику доставки и поведение при сбоях.

1. Инициализация. Worker вызывает SinkConnector.start(props) — коннектор проверяет конфигурацию и устанавливает соединение с целевой системой.
2. Подписка. Каждая задача (SinkTask) подписывается на партиции исходного топика как обычный Kafka-консьюмер.
3. Получение записей. Worker вызывает SinkTask.put(records) с пакетом SinkRecord. Размер пакета определяется consumer.max.poll.records.
4. Запись во внешнюю систему. Задача записывает пакет в целевую систему (базу данных, Elasticsearch, S3).
5. Фиксация offset. После успешного возврата из put() Worker коммитит offset в __consumer_offsets. Если put() бросает исключение — offset не коммитится, пакет будет обработан повторно (at-least-once).

JDBC Sink Connector

JdbcSinkConnector записывает данные из Kafka в реляционную базу данных. Поддерживает PostgreSQL, MySQL, Oracle, SQL Server.

{
  "name": "jdbc-orders-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:postgresql://db:5432/warehouse",
    "connection.user": "kafka_sink",
    "connection.password": "secret",

    "topics": "db-orders",
    "auto.create": "true",
    "auto.evolve": "true",

    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "id",

    "tasks.max": "2"
  }
}

Режимы вставки

Режим upsert обеспечивает идемпотентность записи — при повторной обработке одного и того же события (at-least-once доставка) таблица остаётся консистентной.

Конфигурация первичного ключа

Auto-create и auto-evolve

• auto.create=true — автоматически создаёт таблицу при первом запуске, если её нет.
• auto.evolve=true — автоматически добавляет новые столбцы при изменении схемы сообщений.

Elasticsearch Sink Connector

ElasticsearchSinkConnector индексирует сообщения из Kafka в Elasticsearch. Полезен для полнотекстового поиска, аналитических дашбордов (Kibana), log aggregation.

{
  "name": "elasticsearch-events-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",

    "topics": "user-events",
    "key.ignore": "false",

    "schema.ignore": "false",
    "compact.map.fixed.schema": "false",

    "tasks.max": "3"
  }
}

Параметр key.ignore:

• false (по умолчанию) — использует ключ сообщения Kafka как _id документа Elasticsearch. Повторная запись того же ключа обновляет документ (идемпотентно).
• true — использует комбинацию topic+partition+offset как _id. Каждое сообщение — уникальный документ.

Для режима полнотекстового поиска, где одно событие = один документ, используйте key.ignore=false с бизнес-ключом (user_id, event_id) — это обеспечивает идемпотентность при повторных доставках.

S3 Sink Connector

S3SinkConnector записывает пакеты сообщений из Kafka в Amazon S3 (или S3-совместимое хранилище: MinIO, GCS). Подходит для data lake ingestion, долгосрочного хранения, аналитических конвейеров.

{
  "name": "s3-events-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "s3.bucket.name": "my-data-lake",
    "s3.region": "us-east-1",

    "topics": "user-events",
    "topics.dir": "kafka-data",

    "flush.size": "1000",
    "rotate.interval.ms": "600000",

    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.avro.AvroFormat",
    "schema.compatibility": "FULL",

    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "locale": "ru_RU",
    "timezone": "Europe/Moscow",

    "tasks.max": "4"
  }
}

Форматы файлов

Партиционеры S3

Параметр partitioner.class определяет структуру директорий в S3:

Когда создаётся новый S3-файл

S3 Sink создаёт новый файл при любом из условий:

1. Накоплено flush.size записей.
2. Прошло rotate.interval.ms миллисекунд с последней ротации.
3. Произошла смена временного раздела (при TimeBasedPartitioner).

Tracking consumer offset: sink как обычный consumer

Sink connector использует стандартный Kafka consumer под капотом. Group ID коннектора = имя коннектора. Consumer offset хранится в __consumer_offsets.

Это означает:

• Инструменты мониторинга consumer lag (Kafka UI, Grafana + kafka_exporter) работают для sink коннекторов из коробки.
• kafka-consumer-groups.sh --group my-sink-connector покажет отставание задач.
• При перезапуске задача продолжит с последнего закоммиченного offset.

Exactly-once для sink connectors

По умолчанию sink connector обеспечивает at-least-once: при сбое задача повторно обработает последний незакоммиченный пакет. Внешняя система может получить дублирующиеся записи.

Стратегии достижения идемпотентности:

Схема конвейера с sink connectors

Ключевые выводы

1. Sink connector читает из Kafka как обычный consumer и пишет во внешнюю систему через SinkTask.put().
2. Offset хранится в __consumer_offsets — стандартный механизм Kafka.
3. JDBC Sink поддерживает upsert для идемпотентной записи. auto.create создаёт таблицу автоматически.
4. Elasticsearch Sink — key.ignore=false использует бизнес-ключ как _id для idempotent upsert.
5. S3 Sink буферизует данные до flush.size или rotate.interval.ms. Parquet + TimeBasedPartitioner оптимально для аналитики.
6. AvroConverter с Schema Registry (Модуль 06) обеспечивает автоматическую передачу схемы между source и sink.