Page History

...

В отличие от кластера СУБД MongoDB в кластере Kafka сохраняется работоспособность при временной изоляции узла. Лог накапливающихся изменений буферизуется и накатывается на остальные узлы кластера при возобновлении связи. Репликация данных обеспечивается отправкой сервисом системы WNAM сообщений об измененных объектах в кластере Kafka. Репликация является двухсторонней. Поддерживается возможность определения и разрешения конфликтов.

Каждая база данных имеет свой уникальный ID. Он используется при обмене данных, для определения источника и приемника данных.  ID прописывается при первом старте WNAM  в коллекцию wnam_cluster_id и в дальнейшем не меняется. Для сброса ID нужно удалить эту коллекцию ( в консоли mongo команда db.wnam_cluster_id.drop() ) и перезапустить все экземпляры wnam привязанные к этой базе данных. Также WNAM может сам определить если запущены два разных узла mongo с одним ID. В этом случае на странице "Кластеры WNAM" появляется значок с хинтом "Изменить кластер ID. Потребуется рестарт" при нажатии на который WNAM сам заменит ID  и произведет перезагрузку.

Если система WNAM установлена из готового образа (OVF), то брокер Kafka уже установлен в /opt/kafka и предварительно частично настроен. Остаётся только настроить IP-адреса узлов.

...

process.roles=broker,controller
# укажите идентификатор этого узла. у каждого из узлов кластера он свой, используйте числа 1,2,3 и т.д.
node.id=1
# укажите все узлы, с их уникальными идентификаторами
controller.quorum.voters=1@172.16.135.10:9093,2@10.241.200.123:9093,3@10.241.200.124:9093
# укажите адрес этого узла
listeners=PLAINTEXT://172.16.135.10:9092,CONTROLLER://172.16.135.10:9093
inter.broker.listener.name=PLAINTEXT
# укажите адрес этого узла
advertised.listeners=PLAINTEXT://172.16.135.10:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka/kraft-combined-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
# укажите число серверов в кластере
offsets.topic.replication.factor=3
# укажите число серверов в кластере
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=1

#
# Параметры влияющие на потребление дискового пространства, а также на объем и время хранимых данных 

# Количество часов хранения файла журнала перед его удалением (в часах), по умолчанию 168

log.retention.hours=72

# Политика хранения журналов, основанная на размере. Сегменты удаляются из журнала до тех пор, пока оставшиеся сегменты не опустятся ниже log.retention.bytes.
# Функционирует независимо от log.retention.hours. Значение по умолчанию -1

log.retention.bytes=1073741824

# Максимальный размер файла сегмента журнала. Когда этот размер будет достигнут, будет создан новый сегмент журнала. По умолчанию 1073741824

log.segment.bytes=268435456

# Интервал, с которым проверяются сегменты журнала на предмет возможности их удаления в соответствии с политиками хранения
log.retention.check.interval.ms=300000

# При первоначальной регистрации в кворуме контроллера указывается количество миллисекунд, которое необходимо подождать, прежде чем объявить о сбое и выйти из процесса посредника.
initial.broker.registration.timeout.ms = 60000

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

...