Запуск узлов кластера Corax#

Выполнив вышеуказанные настройки, можно запускать узлы кластера Corax.

Отдельные узлы можно запускать в любое время, то есть запуск отдельного узла сервера Corax не зависит от остальных, как это было с кластером ZooKeeper.

Запуск каждого отдельного сервера Corax выполняется командой:

bin/kafka-server-start.sh config/server.properties

При запуске всех узлов кластера Corax в консолях запуска должны быть следующие записи:

INFO [Kafka Server ID], started (kafka.server.KafkaServer)

Где ID — идентификатор сервера Corax, указанный в ключе broker.id.

Сетевые настройки#

Название	Описание	Тип	Значение по умолчанию
message.max.bytes	Максимальный размер сообщений, принимаемых сервером. Пример: message.max.bytes=1000012	int	1000012
num.network.threads	Количество сетевых потоков, используемых сервером для обработки сетевых запросов. Пример: num.network.threads=3	int	3
queued.max.requests	Лимит запросов. Если лимит превышен, начнется блокировка сетевых запросов. Пример: message.max.bytes=500	int	500
replica.socket.receive.buffer.bytes	Размер (в байтах) буфера сокета для приема сетевых запросов. Пример: replica.socket.receive.buffer.bytes=65536	int	65536
replica.socket.timeout.ms	Тайм-аут (в миллисекундах) сокета для сетевых запросов. Пример: replica.socket.timeout.ms=102400	int	30000
request.timeout.ms	Максимальное количество времени, которое уходит на ожидание клиентом ответа на отправленный запрос. Если в течение этого времени ответ на запрос не получен, клиент отправит повторный запрос. Если лимит повторных запросов превышен, то запросы будут прекращены. Пример: request.timeout.ms=102400	int	30000
socket.receive.buffer.bytes	Размер буфера в байтах на прием данных через сокет. Пример: socket.receive.buffer.bytes=102400	int	102400
socket.request.max. bytes	Максимальное количество байтов на сокет-запрос. Пример: socket.request.max.bytes=104857600	int	104857600
socket.send.buffer.bytes	Размер буфера в байтах на передачу данных через сокет. Пример: socket.send.buffer.bytes =102400	int	102400
connections.max.idle.ms	Тайм-аут неудачных попыток соединения в миллисекундах. Пример: connections.max.idle.ms=60000	long	60000
controller.socket.timeout.ms	Тайм-аут сокета в миллисекундах для Controller-to-Broker каналов. Пример: controller.socket.timeout.ms=30000	int	30000
max.connections.per.ip	Максимальное количество соединений с каждого IP-адреса. Пример: max.connections.per.ip=2147483647	int	2147483647
max.connections.per.ip.overrides	Настройка, позволяющая переопределять заданное максимальное количество соединений max.connections.per.ip, варьируя его для каждого IP-адреса. Пример: max.connections.per.ip.overrides=<host1-ip-address>:50,<host2-ip-address>:20,<host3-ip-address>:45	string	«»

Настройки подключения к ZooKeeper#

Запуск Corax невозможен без подключения к ZooKeeper.

Для подключения Corax-брокеров к кластеру ZooKeeper в файле конфигураций server.properties каждого Corax-брокера в свойство zookeeper.connect необходимо прописать список всех хостов ZooKeeper-брокеров в формате IP:PORT, разделяя их запятыми. Где:

IP — адрес хоста, на котором работает ZooKeeper-брокер;
PORT — порт (clientPort в файле zookeeper.properties) для подключения клиентов ZooKeeper — Corax-брокеров.

Для всех остальных настроек выставятся значения по умолчанию, если они не будут указаны явно в файле конфигураций. Список всех известных настроек ZooKeeper, относящихся к файлу конфигураций server.properties Corax-брокера, и их значений по умолчанию приведен в таблице ниже.

Название	Описание	Тип	Значение по умолчанию
zookeeper.connect	Список адресов ZooKeeper-брокеров в формате `IP:PORT`, где: `IP` — адрес хоста ZooKeeper; `PORT` — порт клиентов ZooKeeper (`clientPort`). Пример: zookeeper.connect=`<host1-ip-address>:2181`, `<host2-ip-address>:2181`,`<host1-ip-address>:2181`	string	«»
zookeeper.connection.timeout.ms	Максимальное время ожидания клиентом соединения с ZooKeeper. Если это значение не выставлено, то используется zookeeper.session.timeout.ms. Пример: zookeeper.connection.timeout.ms=6000	int	null
zookeeper.session.timeout.ms	Тайм-аут сессии ZooKeeper в миллисекундах. Пример: zookeeper.session.timeout.ms=6000	int	6000
zookeeper.set.acl	Принудить клиента использовать безопасные ACL (Access Control Lists — списки контроля доступа). Пример: zookeeper.set.acl=false	boolean	false
zookeeper.sync.time.ms	Продолжительность времени в миллисекундах, через которое происходит синхронизация ZooKeeper-последователя с ZooKeeper-лидером. Пример: zookeeper.sync.time.ms=2000	int	2000

Настройки диска#

Название	Описание	Тип	Значение по умолчанию
log.dir	Директория, в которой хранятся данные Corax — логи сообщений. Это свойство является дополнением к свойству `log.dirs`. В случае, если в файле конфигураций не был указан путь хранения данных, Corax разместит логи в /tmp/kafka-logs. Пример: log.dir=/tmp/kafka-logs	string	/tmp/kafka-logs
log.dirs	Список директорий, в которых хранятся данные Corax — логи сообщений. Если не задано, используется значение из `log.dir`. Если планируется использовать Corax под большой нагрузкой, то должно быть выделено достаточно места, так как все сообщения будут сохраняться на диск именно в указанной директории. Если на физическом сервере возможно использование нескольких физических жестких дисков для хранения сообщений Corax, то для увеличения производительности перечислите пути к директориям на этих дисках через запятую. Пример: log.dirs=/var/log/kafka/kafka-logs,/another\disk/kafka/kafka-logs	string	null
log.flush.interval.messages	Интервал, выражаемый количеством сообщений, накапливающихся в логе партиции до момента сохранения этого лога на диск. Если в логе накопилось сообщений меньше, чем задано в этом свойстве, то в случае сбоя (например, перезапуска ОС) эти сообщения будут потеряны на данной машине, поскольку хранились в оперативной памяти и не были сохранены на диск. Пример: log.flush.interval.messages=10000	long	9223372036854775807
log.flush.interval.ms	Максимальное время (в миллисекундах) хранения сообщения любого топика в оперативной памяти до того, как оно будет сохранено на диск. Если свойство не задано, будет использоваться значение из другого свойства — log.flush.scheduler.interval.ms. Пример: log.flush.interval.ms=1000	long	null
log.flush.offset.checkpoint.interval.ms	Частота, с которой происходит обновление персистентной записи, сохраненной в ходе последней сессии записи на диск. Эта частота служит отправной точкой восстановления лога. Пример: log.flush.offset.checkpoint.interval.ms=10	int	60000
log.flush.scheduler.interval.ms	Частота в миллисекундах, с которой данные каждого лога сохраняются на диск. Пример: log.flush.scheduler.interval.ms=10	long	9223372036854775807
log.retention.bytes	Максимальный размер лога перед его удалением. Пример: log.retention.bytes =100	long	-1
log.retention.hours	Время (в часах) хранения лог-файла до его удаления. Пример: log.retention.hours=168	int	168
log.retention.minutes	Время (в минутах) хранения лог-файла до его удаления. Если это свойство не задано, будет использоваться значение из log.retention.hours. Пример: log.retention.minutes=100000	int	null
log.retention.ms	Время (в миллисекундах) хранения лог-файла до его удаления (третье по важности после свойства log.retention.minutes). Если это свойство не задано, будет использоваться значение из log.retention.minutes. Пример: log.retention.ms=6000	long	null
log.roll.hours	Циркуляция логов (в часах) — максимальное количество времени до записи нового сегмента лога. Это свойство имеет более низкий приоритет, чем аналогичное свойство log.roll.ms. Пример: log.roll.hours=10	int	168
log.roll.jitter.hours	Максимальное случайное отклонение (для циркуляции логов) (в часах) для вычитания из значения logRollTimeMillis (в миллисекундах). Вторично по отношению к свойству log.roll.jitter.ms. Пример: log.roll.jitter.hours=1	int	0
log.roll.ms	Циркуляция логов (в миллисекундах) — максимальное количество времени до записи нового сегмента лога. Если свойство не задано, используется значение из log.roll.hours. Пример: log.roll.ms=10	long	null
log.segment.bytes	Максимальный размер одного лог-файла. Пример: log.segment.bytes=250	int	1073741824
log.segment.delete.delay.ms	Количество времени (в миллисекундах) до удаления файла из файловой системы. Пример: log.roll.delete.delay.ms=10	long	60000
log.cleaner.backoff.ms	Количество времени на простаивание до появления логов для удаления. Пример: log.cleaner.backoff.ms=10	long	15000
log.cleaner.dedupe.buffer.size	Общий размер памяти, выделяемой для удаления дубликатов во всех потоках службы «Log Cleaner». Пример: log.cleaner.dedupe.buffer.size=10	long	134217728
log.cleaner.delete.retention.ms	Количество времени сохранения удаленных записей. Пример: log.cleaner.delete.retention.ms=10	long	86400000
log.cleaner.enable	Включение службы «Log Cleaner». Должно иметь значение true, если есть какой-либо топик со свойством cleanup.policy=compact, включая топик с внутренними смещениями. Если значение установлено в false, то эти топики не будут компактными и будут постоянно увеличиваться в размере. Пример: log.cleaner.enable=true	boolean	true
log.cleaner.io.buffer.load.factor	Коэффициент загрузки буфера при удалении дубликатов процессом «Log Cleaner». Например, при коэффициенте, равном 0.9, степень загрузки буфера — 90%. Чем больше значение, тем большее количество логов будет удаляться за один раз, однако вместе с этим будет возрастать и число хеш-коллизий. Пример: log.cleaner.io.buffer.load.factor=0.9	double	0.9
log.cleaner.io.buffer.size	Общий объем памяти, используемый службой «Log Cleaner» на операции ввода/вывода для всех потоков этой службы. Пример: log.cleaner.io.buffer.size=255	int	524288
log.cleaner.io.max.bytes.per.second	Ограничение операций ввода/вывода службы «Log Cleaner» так, чтобы сумма байтов при операциях ввода и вывода была всегда меньше значения, указанного в этом свойстве. Пример: log.cleaner.io.max.bytes.per.second=5.0	double	1.7976931348623157E308
log.cleaner.min.cleanable.ratio	Минимальное соотношение между «грязным» и общим логом для определения пригодности лога для удаления. Пример: log.cleaner.min.cleanable.ratio=0.5	double	0.5
log.cleaner.threads	Число фоновых потоков, используемых для чистки логов службой «Log Cleaner». Пример: log.cleaner.threads=1	int	1
log.cleanup.policy	Политика очистки сегментов логов, находящихся за пределами сохраняющего окна «retention window». Возможны два режима: delete — удаление лога, compact — сжатие лога. Пример: log.cleanup.policy=delete	string	delete
log.index.interval.bytes	Интервал (в байтах), через который прибавляется запись к индексу смещения (offset). Пример: log.index.interval.bytes=4096	int	4096
log.index.interval.bytes	Интервал (в байтах), через который прибавляется запись к индексу смещения (offset). Пример: log.index.interval.bytes=4096	int	4096
log.index.size.max.bytes	Максимальный размер индекса смещения (offset) в байтах. Пример: log.index.size.max.bytes=10485760	int	10485760
log.preallocate	Требуется ли создавать файл при выделении нового сегмента. Для Kafka на Windows рекомендуется установить значение true. Пример: log.preallocate=false	boolean	false
log.retention.check.interval.ms	Частота в миллисекундах, с которой служба «Log Cleaner» проверяет каждый лог на необходимость его удаления. Пример: log.retention.check.interval.ms=300000	long	300000

Настройки топиков#

Название	Описание	Тип	Значение по умолчанию
cleanup.policy	Определяет политику очистки логов: — delete: удаление старых сообщений после истечения срока хранения. — compact: сохранение последних версий ключей	log.cleanup.policy	delete
delete.retention.ms	Время жизни удаленных записей после удаления ключа (только при политике compact)	log.cleaner.delete.retention.ms	86400000 (24 часа)
flush.messages	Количество сообщений между принудительными сбросами буферов на диск. Если установлено значение None, используется интервал по таймеру (`flush.ms`)	log.flush.interval.messages	None
flush.ms	Интервал времени между принудительными сбросами буферов на диск. Используется совместно с `flush.messages`	log.flush.interval.ms	None
index.interval.bytes	Частота записи индексов в файл журнала. Чем меньше значение, тем быстрее индексируется журнал, но больше расходуется память	log.index.interval.bytes	4096
max.message.bytes	Максимально допустимый размер одного сообщения в байтах. Сообщения большего размера отвергаются брокером	message.max.bytes	1000000
min.cleanable.dirty.ratio	Минимальная доля грязных разделов перед запуском процесса очистки. Грязный сегмент содержит устаревшие данные	log.cleaner.min.cleanable.ratio	0.5
min.insync.replicas	Минимальное количество реплик, которое должно подтвердить запись, чтобы она считалась успешной	min.insync.replicas	1
retention.bytes	Максимальный объем хранимых данных сегмента лога. После достижения лимита начинается очистка логов	log.retention.bytes	None
retention.ms	Максимальное время хранения логов в миллисекундах. Применимо только при политике `delete`	log.retention.ms	604800000 (7 дней)
segment.bytes	Размер файла сегмента лога в байтах. Когда этот лимит достигнут, создается новый сегмент	log.segment.bytes	1073741824 (1 ГБ)
segment.index.bytes	Максимальный размер индекса сегмента лога. Индекс позволяет быстро находить позицию конкретного сообщения по смещению (offset)	log.index.size.max.bytes	10485760 (10 МБ)
segment.ms	Продолжительность существования сегмента лога в миллисекундах. По истечении данного периода создается новый сегмент	log.segment.ms	604800000 (7 дней)
segment.jitter.ms	Случайное отклонение от интервала сегментации, используемое для предотвращения одновременной сегментации всех разделов	log.roll.jitter.ms	0