Page History

...

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. Для этого следует сформировать ключ, который должен быть уникальным в пределах вашего кластера, т.е. на каждом узле он должен быть одинаков.

...

controller.quorum.voters=1@172.16.135.10:9093,2@10.241.200.123:9093,3@10.241.200.124:9093

числа перед @ это node.id, которые должны быть прописаны одинаково в конфигурациях на всех серверах кластера.

...

netams:
 wnam2:
  cluster:
    # optional, default false. Использовать kafka
    kafka_enabled: true 
    # optional default false, отвечает за синхронизацию данных, между разными кластерами wnam, если true синхронизируются все коллекции, если false только конфигурационные параметры 
    full_sync: true 
    # optional, default 12 количество реплик (серверов) в кластере kafka
    replicas: 32 
    #  optional, default true. Включает репликацию изменений между нодами.
    use_replication: true
    # optional, default true. Пытаться выбрать лидера в kafka при падении брокеров kafka
    unclean_election: true 
    # optional, default empty. Если master, то все задачи выполняются на этом сервере.
    role: master 
    # optional, default 'true'. Использовать локальный кэш. Может быть true или false, отключать кэш следует только в конфигурации когда нет кафки и больше одного сервера wnam
    use_cache: true 
    # optional,experimental, default false. Показывает кнопку для принудительного запуска синхронизации одного узла кластера с другого Периодически проверяет расхождение в количестве записей в таблицах и при необходимости запускает синхронизацию. 
    # По умолчанию количество синхронизируемых записей ограничено в 100000, настраивается параметром sync_records_limit в Настройки | Параметры 
    showauto_sync: true false
    # optional. Объекты которые передаются через кафку на другие ноды. Список через запятую. По умолчанию пустой (все объекты)
    producer.types: 
    # optional. Объекты которые принимаются через кафку с других нод. Список через запятую. По умолчанию пустой (все объекты)
    consumer.types: 
 
spring:
 # применяется если kafka_enabled = true
 kafka:
  bootstrap-servers: 172.16.135.10:9092,10.241.200.123:9092,10.241.200.124:9092

...