Конфигурация

Опции коннектора распределены по трём уровням DDL и почти не пересекаются: подключение и безопасность живут на SERVER, всё, что относится к топику и формату - на FOREIGN TABLE, учётные данные - в USER MAPPING.

Приоритет уровней

Пересекаются только sasl_username и sasl_password: значение из USER MAPPING перекрывает серверное. Остальные опции безопасности задаются лишь на сервере, опции таблицы - лишь на таблице.

Опции сервера

Подключение

Опция	Умолчание	Описание
brokers	-	обязательная; список брокеров host1:port1,host2:port2 для начального подключения
connect_timeout_ms	10000	таймаут подключения к брокеру, > 0
offset_fetch_timeout_ms	30000	таймаут офсетных запросов на старте скана: закоммиченные офсеты группы и high watermark партиций. Первый такой запрос включает SSL handshake и поиск координатора группы, на нагруженном кластере он небыстрый. При превышении таймаута запрос падает с ошибкой, тихого отката к start_offset нет

Безопасность

Опция	Умолчание	Описание
security_protocol	PLAINTEXT	PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
sasl_mechanism	-	PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI; обязателен, если протокол содержит SASL
sasl_username / sasl_password	-	учётные данные SASL; для PLAIN и SCRAM-* обязательны, переопределяются в USER MAPPING
ssl_ca_location	-	путь к CA-сертификату (PEM) для проверки сертификата брокера
ssl_cert_location / ssl_key_location	-	клиентский сертификат и приватный ключ для mTLS
ssl_key_password	-	пароль приватного ключа

Файлы сертификатов должны лежать по одинаковому пути на координаторе и всех сегмент-хостах: к брокерам ходит каждый сегмент отдельно. Пароли (sasl_password, ssl_key_password) маскируются в логах и затираются в памяти при освобождении.

CREATE SERVER kafka_secure
  FOREIGN DATA WRAPPER kafka_fdw
  OPTIONS (
    brokers                 'broker1:9093,broker2:9093',
    connect_timeout_ms      '15000',
    offset_fetch_timeout_ms '30000',
    security_protocol       'SASL_SSL',
    sasl_mechanism          'SCRAM-SHA-256',
    ssl_ca_location         '/etc/kafka/certs/ca-cert.pem'
  );

Опции таблицы

Опция	Умолчание	Описание
topic	-	обязательная; имя топика
format	json	json или csv
mpp_execute	-	обязательная, единственное рабочее значение - all segments: читают и пишут только сегменты. Без опции или с 'master' первый SELECT или INSERT упадёт с подсказкой. Ядро Greenplum не передаёт эту опцию коннектору при CREATE, поэтому проверка срабатывает при обращении к таблице
batch_size	10000	размер порции, которую сегмент запрашивает у librdkafka за один вызов, > 0; значения выше 100000 усекаются. На объём результата не влияет
batch_timeout_ms	5000	предельная длительность скана на сегменте, > 0. При end_offset 'latest' скан завершается раньше, как только дочитает до границы
start_offset	earliest	откуда читать партицию без закоммиченного офсета: earliest или latest
end_offset	none	none - читать всё, что приходит, пока не истечёт batch_timeout_ms; latest - остановиться на high watermark, зафиксированном в начале скана
consumer_group	-	Kafka group.id для хранения офсетов; обязателен при включённом offset_storage
offset_storage	none	none - не хранить офсеты; kafka - коммитить в __consumer_offsets; gp - устаревший алиас kafka, ведёт себя идентично
error_action	fail	что делать со структурно битым сообщением: fail или skip
error_limit	0	предел пропусков при skip, считается на каждый сегмент; 0 - без ограничений
csv_delimiter	,	разделитель полей, ровно один символ
csv_quote	"	символ кавычек
csv_escape	"	символ экранирования кавычек внутри поля
csv_header	false	пропускать первую строку сообщения как заголовок

CSV-опции работают только при format 'csv'. Разделитель, кавычка и символ экранирования должны быть попарно различными.

batch_timeout_ms относится к одному сегменту, и это единственное, что ограничивает объём SELECT при end_offset 'none': запрос вернёт всё, что успело прийти за это время. До 1.5 batch_size обрезал выдачу до указанного числа строк на сегмент; если такой лимит нужен, задайте его фильтром WHERE __offset < … или сократите окно таймаута.

CREATE FOREIGN TABLE events (
    event_id    int,
    event_type  text,
    payload     jsonb,
    __offset    bigint,
    __partition int
)
  SERVER kafka_secure
  OPTIONS (
    topic          'events',
    format         'json',
    batch_size     '5000',
    end_offset     'latest',
    consumer_group 'etl_pipeline',
    offset_storage 'kafka',
    mpp_execute    'all segments'
  );

Учётные данные

USER MAPPING нужен каждому пользователю, который обращается к внешним таблицам. Без SASL достаточно пустого:

CREATE USER MAPPING FOR CURRENT_USER SERVER kafka_server;

С аутентификацией пароль лучше держать здесь, а не в опциях сервера: серверные опции видны всем, кто может посмотреть каталог, а маппинг - только своему владельцу и суперпользователю.

CREATE USER MAPPING FOR analyst
  SERVER kafka_secure
  OPTIONS (sasl_username 'analyst_user', sasl_password 'secret');

Системные колонки

Колонки с префиксом __ необязательны. Добавьте нужные в определение таблицы - коннектор исключит их из сериализации и заполнит из метаданных сообщения.

Колонка	Тип	При чтении	При записи
__offset	bigint	офсет сообщения в партиции	игнорируется, офсет назначает Kafka
__partition	int	номер партиции	целевая партиция
__timestamp	bigint	отметка времени сообщения, мс от epoch	игнорируется
__key	text	ключ сообщения	ключ сообщения
__topic	text	имя топика	игнорируется

Типы колонок

Значение из JSON или CSV передаётся во входную функцию типа колонки, поэтому годится любой тип Greenplum, чья входная функция примет текст значения. Отсюда несколько правил, о которые чаще всего спотыкаются:

Значение в сообщении	Тип колонки	Комментарий
скаляр, строка, null	int, numeric, text, timestamp, bool и т. п.	обычный случай
вложенный объект {…} или массив […]	jsonb, json, text	попадает сырым JSON-текстом; jsonb даёт индексацию и jsonb_array_length()
вложенный массив	не jsonb[], json[], text[]	у массива PostgreSQL другой синтаксис литерала, входная функция отвергнет JSON-массив
большое целое, например 17937001869924977406676457305	numeric	не влезает ни в int, ни в bigint

Несоответствие типа значению - ошибка преобразования. Она прерывает запрос всегда и error_action 'skip' её не подавляет.

Функции мониторинга

Все читают состояние вживую из Kafka, ничего не кэшируя.

Функция	Что возвращает
kafka.topic_info(server)	топик, партиция, лидер, число реплик и ISR по каждому топику кластера
kafka.consumer_lag(server, group, topic)	закоммиченный офсет, high watermark и lag по партициям
kafka.offsets_info(server, group, topic)	то же плюс last_loaded_offset = committed_offset - 1, то есть офсет последнего сообщения, реально загруженного в Greenplum
kafka.version()	версия библиотеки kafka_fdw.so на хосте, где выполняется вызов

NULL в committed_offset означает, что группа ещё ничего не коммитила для этой партиции. Позиция, помеченная незавершённой транзакцией (см. Использование), показывается как прежняя: функции отдают то, с чего начнёт следующий запрос. Функции kafka._committed() и kafka._watermarks() внутренние, вызывать их напрямую не нужно.

Функции управления офсетами

Функция	Что делает
kafka.set_offset(server, group, topic, partition, offset)	выставляет закоммиченный офсет одной партиции; возвращает значение, прочитанное обратно из Kafka
kafka.reset_offsets(server, group, topic, position)	переводит все партиции топика; position - 'earliest', 'latest' или timestamp вида '2026-10-01 00:00:00+03'; возвращает по строке на партицию

Офсет - номер следующего сообщения для чтения, то же, что committed_offset в kafka.offsets_info(). Он должен попадать в [low_watermark, high_watermark] партиции, иначе функция вернёт ошибку с границами. При сбросе по времени партиции, в которых нет сообщений не раньше указанного момента, встают на high watermark. Пока функция работает, сканов с той же consumer_group быть не должно: их коммит перезапишет установленное значение. Права в Kafka нужны те же, что для обычного коммита, - Read на группу.