Skip to content

Обновление

Порядок важен: сначала библиотека на всех хостах, потом схема каталога, потом переподключение сессий.

1. Файлы расширения

Разложите новые kafka_fdw.so и файлы расширения на координатор и все сегмент-хосты - через Ansible-роль или вручную, так же, как при первой установке (см. Установка):

ansible-playbook -i inventory playbook.yml --tags install

Останавливать Greenplum не нужно. Библиотека на всех хостах должна быть одной и той же версии: устаревшая .so хотя бы на одном сегменте будет давать ошибки старых версий на тех запросах, где ему достанется партиция. Сверить можно контрольными суммами, а начиная с 1.5 - из SQL после переподключения (шаг 3):

for h in gp-mstr gp-seg1 gp-seg2; do ssh $h md5sum $GPHOME/lib/postgresql/kafka_fdw.so; done

2. Схема каталога

В каждой базе, где установлено расширение:

SELECT extversion FROM pg_extension WHERE extname = 'kafka_fdw';   -- текущая версия
ALTER EXTENSION kafka_fdw UPDATE;

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

3. Переподключение сессий

Открытые соединения продолжают работать со старой библиотекой до переподключения - это самая частая причина «поставили новую версию, а баг остался». Переоткройте сессии приложений и пулов после обновления.

Проверка, что всё встало:

SELECT extversion FROM pg_extension WHERE extname = 'kafka_fdw';          -- 1.5

-- библиотека на координаторе и на каждом сегменте, везде должна быть 1.5.0
SELECT 'master' AS host, kafka.version()
UNION ALL
SELECT 'seg' || gp_segment_id, kafka.version() FROM gp_dist_random('gp_id')
ORDER BY 1;

SELECT * FROM kafka.topic_info('kafka_server') LIMIT 5;

Обновление 1.4.x -> 1.5

Нужны все три шага: в 1.5 появились новые SQL-функции, поэтому ALTER EXTENSION kafka_fdw UPDATE обязателен. Внешние таблицы, серверы и офсеты в Kafka сохраняются, загрузки продолжатся с прежних позиций.

После обновления проверьте внешние таблицы на опцию mpp_execute. Таблицы без неё или со значением 'master' раньше молча возвращали ноль строк, теперь они падают при обращении:

SELECT c.relname, ft.ftoptions
  FROM pg_foreign_table ft JOIN pg_class c ON c.oid = ft.ftrelid
 WHERE NOT (ft.ftoptions::text LIKE '%mpp_execute=all segments%');

ALTER FOREIGN TABLE <table> OPTIONS (ADD mpp_execute 'all segments');

Что ещё поменялось в поведении и может задеть существующие загрузки:

  • batch_size больше не ограничивает число строк в SELECT. При end_offset 'none' запрос вернёт всё, что пришло за batch_timeout_ms.
  • INSERT падает, если брокеры не подтвердили все сообщения за 60 секунд; раньше он завершался успешно с WARNING.
  • BEGIN; SELECT ... FROM <внешняя>; ROLLBACK; не продвигает офсеты; в явной транзакции они фиксируются по COMMIT.
  • Координатор подключается к Kafka при фиксации транзакций с загрузкой. Брокеры должны быть доступны с хоста координатора, для SSL - с теми же сертификатами, что на сегментах.
  • В логе и выводе psql появились WARNING про retention (below the low watermark) и перебои брокеров (transient broker error). Это не ошибки загрузки, но повод посмотреть на lag и состояние кластера Kafka.

Обновление 1.3 -> 1.4.1

Все изменения находятся в библиотеке, SQL-объекты не менялись. Достаточно шагов 1 и 3: разложить новую .so и переподключить сессии. ALTER EXTENSION выполнить не помешает, но версия расширения останется той же - 1.4.

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

Обновление с версий 1.2 и старше

До 1.3 офсеты хранились в таблице kafka.offsets внутри Greenplum. Скрипт обновления её удаляет, содержимое не мигрируется.

После обновления первый SELECT по каждой партиции начнёт с start_offset таблицы (по умолчанию earliest), если для её consumer_group в Kafka ещё нет закоммиченного офсета. На топике с длинной историей это означает повторную загрузку всего, что там лежит.

Если перечитывание недопустимо, перед обновлением выпишите нужные позиции из kafka.offsets и после обновления проставьте их в Kafka. С 1.5 это делается из SQL; значение - committed_offset из старой таблицы плюс один, потому что в ней хранился последний загруженный офсет, а Kafka хранит следующий для чтения:

SELECT kafka.set_offset('kafka_server', 'etl_pipeline', 'events', 0, 5001);

То же через Kafka CLI:

kafka-consumer-groups.sh --bootstrap-server broker1:9092 \
  --group etl_pipeline --topic events:0 \
  --reset-offsets --to-offset 5001 --execute

Опцию offset_storage 'gp' в существующих таблицах менять не обязательно: с 1.3 это алиас kafka, ведёт себя идентично. В новых таблицах пишите kafka.