Инициализация транзакционного продюсера

«# Вопрос 860: Как обеспечивать exactly-once semantics в Kafka для embedding?

Краткий тезис

Exactly-once semantics (EOS) в Kafka для пайплайна эмбеддингов означает, что каждое сообщение будет обработано ровно один раз — ни пропущено, ни продублировано. Это критично, чтобы не вставлять дублирующиеся векторы в векторную базу данных и не тратить ресурсы на повторное вычисление эмбеддингов. Достигается комбинацией идемпотентного продюсера (idempotence|enable.idempotence=true), транзакционного продюсера/консюмера и идемпотентного консюмера, который перед вычислением эмбеддинга проверяет уникальный идентификатор сообщения (например, в Redis). Для embedding-пайплайна важна именно атомарная связка «прочитать → вычислить эмбеддинг → сохранить → закоммитить оффсет».

1. Термин: Exactly-once semantics (EOS)

Exactly-once semantics — это гарантия доставки, при которой каждое сообщение обрабатывается ровно один раз, даже при сбоях продюсера, брокера или консюмера. В Kafka EOS строится на трёх китах:

• Идемпотентность продюсера — повторная отправка одного и того же сообщения не приводит к дубликатам.
• Транзакции — атомарная запись в несколько разделов и атомарное чтение-запись между консюмером и продюсером.
• Изоляция чтения — консюмер видит только закоммиченные транзакции.

Для пайплайна эмбеддингов EOS необходим, потому что:

• Сохранение дублирующего вектора в векторной БД приведёт к искажению результатов поиска (один и тот же документ будет найден дважды с разными ID).
• Повторное вычисление эмбеддинга через нейросеть — дорогая операция, особенно при больших объёмах.
• Без EOS сложно гарантировать целостность данных между Kafka и внешней системой (векторное хранилище).

2. Проблемы, возникающие без exactly-once

EOS решает все три сценария: сообщение не потеряется и не будет обработано дважды.

3. Компоненты exactly-once в Kafka

3.1 Идемпотентный продюсер

Включается настройками:

producer_config = {
    'bootstrap.servers': 'localhost:9092',
    'enable.idempotence': True,
    'acks': 'all',
    'max.in.flight.requests.per.connection': 5  # можно ≤5 при enable.idempotence
}

• enable.idempotence=true — продюсер присваивает каждому сообщению уникальный producer id (PID) и sequence number. Брокер на стороне отбрасывает дублирующиеся sequence number.
• acks=all — подтверждение от всех in-sync реплик, гарантирует, что сообщение не потеряется при сбое лидера.
• max.in.flight.requests.per.connection — при идемпотентности можно ≤5, чтобы сохранить порядок сообщений (если не нужен строгий порядок, можно больше).

3.2 Транзакционный продюсер и консюмер

Для атомарного связывания чтения из Kafka и записи в Kafka (или во внешнюю систему) используются транзакции.

from confluent_kafka import Producer, Consumer, KafkaException

# Инициализация транзакционного продюсера
txn_producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': 'embedding-pipeline-1',
    'enable.idempotence': True,
})
txn_producer.init_transactions()

Транзакционный консюмер

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'embedding-group',
    'enable.auto.commit': False,
    'isolation.level': 'read_committed',  # не читать незакоммиченные сообщения
})

• transactional.id — уникальный идентификатор, который обеспечивает восстановление после сбоя. При перезапуске продюсер с тем же transactional.id «зачищает» незавершённые транзакции.
• isolation.level=read_committed — консюмер не видит сообщения из открытых или прерванных транзакций, что предотвращает чтение «грязных» данных.

3.3 Идемпотентный консюмер (внешняя дедупликация)

Даже с транзакциями и идемпотентностью продюсера, если консюмер делает внешний side-effect (сохранение эмбеддинга в БД), нужен дополнительный механизм. Kafka не умеет откатывать запись в MongoDB/Pinecone. Поэтому консюмер должен сам обеспечивать идемпотентность.

Подход хранить идентификаторы обработанных сообщений в быстром хранилище (Redis, таблица в БД с уникальным индексом).

import redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def process_message(msg):
    msg_id = msg.key().decode('utf-8')  # уникальный ключ сообщения
    if redis_client.get(f"processed:{msg_id}"):
        return  # уже обработано
    # иначе — вычисляем эмбеддинг
    text = msg.value().decode('utf-8')
    embedding = embed_model.encode(text)
    vector_db.insert(embedding, id=msg_id)  # id уникален
    # помечаем как обработанное
    redis_client.setex(f"processed:{msg_id}", 86400, "1")  # TTL 24 часа
    return True

• TTL — сообщение может переотправиться спустя долгое время. TTL задаётся больше разумного окна повторной обработки (например, 24 часа).
• Уникальный ID — если сообщения не имеют естественного ключа, можно использовать комбинацию (topic, partition, offset).

4. Пример полного цикла exactly-once для embedding

from confluent_kafka import Consumer, Producer, KafkaException
import numpy as np
import redis

# Инициализация
redis_client = redis.Redis(...)
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'embed-group',
    'enable.auto.commit': False,
    'isolation.level': 'read_committed',
})
producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': 'embed-txn-1',
    'enable.idempotence': True,
})
producer.init_transactions()

consumer.subscribe(['raw-texts'])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        continue
    if msg.error():
        continue

    msg_id = msg.key().decode()
    # Проверка дубликата
    if redis_client.get(f"processed:{msg_id}"):
        consumer.commit(asynchronous=False)
        continue

    # Начало транзакции
    producer.begin_transaction()
    try:
        text = msg.value().decode()
        embedding = embed(text)  # ваша функция эмбеддинга
        # сохраняем в векторную БД (атомарно через транзакцию не получится, используем идемпотентность)
        vector_store.insert(embedding, id=msg_id)  # id должен быть уникальным

        # Отправляем мета-сообщение (опционально)
        producer.produce('embedding-done', key=msg_id, value=b'done')

        # Отмечаем в Redis
        redis_client.setex(f"processed:{msg_id}", 86400, "1")

        # Коммит Kafka-транзакции
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata()
        )
        producer.commit_transaction()
        # Коммит consumer offset после успеха транзакции
        consumer.commit(asynchronous=False)
    except Exception as e:
        producer.abort_transaction()
        # offset не коммитим – повторим при ребалансе или перезапуске
        redis_client.delete(f"processed:{msg_id}")  # очищаем маркер
        print(f"Error: {e}")

Важно consumer.commit() вызывается только после успешного завершения транзакции и подтверждения, что эмбеддинг сохранён. Если упасть после commit_transaction, но до consumer.commit, при перезапуске сообщение будет прочитано снова, но Redis-маркер уже будет установлен → обработка не повторится.

5. Особенности и компромиссы

6. Альтернативные подходы без транзакций

• At-least-once + идемпотентный консюмер отключить enable.idempotence и транзакции, просто дедуплицировать через Redis/уникальный ключ в БД. Проще, но могут быть дубликаты при сбое после записи в БД, но до коммита оффсета (сообщение прочитается повторно, Redis не даст обработать снова). Минус: потеря порядка при ребалансе.
• Уникальный индекс в векторной БД если вставить запись с уже существующим ID, БД вернёт ошибку — консюмер может закоммитить оффсет и не повторять. Но это не защищает от потери сообщения при сбое до коммита.

7. Практические рекомендации

• Используйте transactions только если у вас есть строгое требование exactly-once на уровне Kafka. Для большинства embedding-пайплайнов достаточно at-least-once с идемпотентным консюмером.
• Мониторинг числа прерванных транзакций, размера очереди __transaction_state, лага consumer group.
• Redis для дедупликации должен быть устойчивым (репликация, AOF). При потере Redis часть сообщений обработается повторно.
• Генерация уникального ID сообщения ключом может быть хэш текста, UUID, или (topic, partition, offset). Используйте msg.offset() при отсутствии кастомного ключа.

8. Связь с embedding pipeline

EOS в Kafka обеспечивает, что каждый чанк текста будет превращён в вектор ровно один раз, даже при перезапуске пайплайна. Это особенно важно в архитектуре Agentic RAG, где несколько агентов могут подписываться на одну и ту же тему и генерировать эмбеддинги. Без EOS возможна дупликация векторов, что приводит к артефактам в retrieval и снижению качества ответов LLM.

Пет-проект для закрепления

Задача Реализовать консюмер, который читает топик docs (JSON с полями id и text), вычисляет эмбеддинг через sentence-transformers/all-MiniLM-L6-v2 и сохраняет в Qdrant с exactly-once гарантией.

Инструменты

• Kafka (например, localhost:9092 через docker-compose)
• Confluent Kafka Python (или kafka-python)
• Redis для дедупликации
• Qdrant (векторная БД)
• Sentence-Transformers для эмбеддинга

Шаги:

1. Развернуть Kafka, Redis, Qdrant локально (docker-compose).
2. Создать топик docs с 2 партициями.
3. Написать продюсер, который отправляет 100 сообщений с разными id и text.
4. Реализовать консюмер с exactly-once (как в разделе 4).
5. Запустить консюмер, затем повторно запустить те же сообщения (сбросив offset) — убедиться, что в Qdrant не появилось дублирующихся точек с одинаковым ID.
6. Проверить, что при сбое во время insert в Qdrant сообщение не теряется (имитировать ошибку).

Ожидаемый результат

• После первого запуска в Qdrant 100 уникальных векторов.
• После повторного запуска (с перечитыванием тех же сообщений) количество векторов не изменится.
• Логи консюмера показывают, что дубликаты отбрасываются проверкой Redis.

Связь с другими вопросами

Навигация

• Предыдущий: 859
• Следующий: 861
• Индекс: 00. Индекс разборов