Сценарии администрирования#

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

Разрешения на выполнение операций полностью соответствуют разрешениям Apache Kafka.

Конфигурирование Zookeeper#

Конфигурирование Брокера Zookeeper осуществляется в файле config/zookeeper.properties.

Допускается создание файлов конфигурации с различными именами, например:

config/zookeeper-1-node-cluster.properties
config/zookeeper-3-node-cluster.properties
config/zookeeper-3-node-cluster-on-1-machine.properties
config/zookeeper-5-node-cluster.properties

При запуске Zookeeper достаточно указать нужный конфигурационный файл.

Конфигурирование узла Zookeeper#

В данном подразделе описывается конфигурирование Zookeeper для запуска в качестве standalone-приложения на одной машине. Этот вид конфигурации не обеспечивает отказоустойчивости и не рекомендован для использования в промышленной эксплуатации.

Конфигурирование узла Apache Zookeeper осуществляется в файле zookeeper.properties.

Минимальная конфигурация#

Пример файла конфигураций zookeeper.properties для одного Брокера:

dataDir=/path/to/zookeeper-data
clientPort=2181
tickTime=2000

Шаги создания конфигурационного файла:

1. Для переменной dataDir укажите путь к директории, в которую Zookeeper будет сохранять файлы с данными, необходимыми ему для работы.

   Пример

   dataDir=/var/log/kafka_2.11-0.9.0.0/zookeeper-data

2. Для переменной clientPort укажите номер порта, к которому будут подключаться клиенты.

   Пример

   clientPort=2181

3. Для переменной tickTime укажите единицу измерения времени Zookeeper — это время в миллисекундах, которое уходит на один «тик» или «такт» Zookeeper. tickTime неявно используется в настройках для различных временных параметров Zookeeper. Например, для временной настройки initLimit=10 — это означает, что initLimit будет выполняться за 10 * tickTime миллисекунд.

   Пример

   tickTime=2000

Расширенная конфигурация#

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

• maxClientCnxns — максимально допустимое количество соединений (на уровне сокетов), устанавливаемых клиентом (IP-адрес) с одним из серверов кластера Zookeeper.

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

  Данная настройка позволяет предотвратить некоторые виды DoS-атак.

  Пример

  maxClientCnxns=0

• dataLogDir — позволяет выделить отдельную директорию для хранения транзакционного лога Zookeeper. По умолчанию все транзакционные логи Zookeeper (файлы log.) хранятся вместе с файлами snapshot., т. н. «снепшотами», в одной общей директории dataDir.

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

  Пример

  dataLogDir=/another/path/to/zookeeper-logs

  Данные Zookeeper — это персистентные (неизменяемые) копии Zookeeper-узлов (znodes), хранимые соответствующим ансамблем на диске в виде файлов двух видов:

  • Транзакционные логи (transaction log) — это файлы с префиксом log., в которые в последовательном порядке записываются все изменения, происходящие с Zookeeper-узлами.

  • Снепшоты (snapshot) — это файлы с префиксом snapshot. Когда файл транзакционного лога достигает определенного размера, создается копия текущего состояния всех Zookeeper-узлов в виде снепшота. Этот снепшот заменяет все предыдущие логи.

  Примечание

  Устаревшие версии файлов транзакционных логов и снепшотов не удаляются с диска сервером Zookeeper — эта задача полностью ложится на администратора Zookeeper.

• globalOutstandingLimit — ограничение времени, позволяющее регулировать количество обращений клиентов к серверу Zookeeper. Поскольку клиенты отправляют запросы быстрее, чем сервер успевает их обработать, сервер может не справиться и израсходовать всю доступную память. По умолчанию число запросов, находящихся в обработке в некоторый момент времени, не превышает 1000.

  Пример

  globalOutstandingLimit=1000

• preAllocSize — размер блока лог-файлов транзакций. По умолчанию равен 64 Мбайт. Использование транзакционного лога минимизирует количество обращений к диску. Если снепшоты Zookeeper создаются слишком часто, размер логов транзакций может не достигать 64 Мбайт. В таких ситуациях включение этого параметра позволяет оптимизировать расход объема дискового пространства.

• snapCount — число транзакций, которые будут записаны в лог транзакций после создания очередного «снепшота» («snapshot») — файла данных Zookeeper. По умолчанию — 10000.

  Пример

  snapCount =10000

• traceFile — файл трассировки в формате traceFile.year.month.day, в который будут записываться все запросы клиентов. Полезен в целях отладки. Побочный эффект — некоторый спад производительности.

Текущая используемая конфигурация#

Файл конфигураций zookeeper.properties для Брокера следующий:

dataDir=/tmp/kafka/zookeeper
clientPort=2185
maxClientCnxns=0

Конфигурирование кластера Zookeeper#

Конфигурирование кластера Apache ZooKeeper осуществляется в файле zookeeper.properties.

Минимальная конфигурация кластера#

Пример минимального файла конфигураций zookeeper.properties для кластера из трех серверов:

server.1=zookeeper-1-ip:2888:3888
server.2=zookeeper-2-ip:2888:3888
server.3=zookeeper-3-ip:2888:3888
dataDir=/path/to/zookeeper-data
clientPort=2181
tickTime=2000
initLimit=5
syncLimit=2

Шаги создания конфигурационного файла:

1. Добавьте в файл переменные server.INDEX=IP:PORT1:PORT2, где:

   • INDEX — индекс (myid) узла Zookeeper в кластере;

   • IP — адрес данного сервера;

   • PORT1 — порт для подключения серверов-последователей в кластере к лидеру;

   • PORT2 — порт для подключения остальных сервисов Zookeeper в момент выбора сервиса-лидера в кластере.

   Если необходимо запустить кластер Zookeeper на нескольких серверах, то в конфигурационном файле укажите адреса всех серверов.

   Пример

   server.1=xxx.xxx.1.1:2889:3889
   server.2=xxx.xxx.1.2:2889:3889
   

2. Для переменной dataDir укажите путь к директории, в которую Zookeeper будет сохранять файлы с данными, необходимыми ему для работы.

   Пример

   dataDir=/var/log/kafka_2.11-0.9.0.0/zookeeper-data

3. Для переменной clientPort укажите номер порта, к которому будут подключаться клиенты.

   Пример

   clientPort=2181

4. Для переменной tickTime укажите единицу измерения времени Zookeeper — это время в миллисекундах, которое уходит на один «тик» или «такт» Zookeeper. tickTime неявно используется в настройках для различных временных параметров Zookeeper. Например, для временной настройки initLimit=10 — это означает, что initLimit будет выполняться за 10 * tickTime миллисекунд.

   По умолчанию tickTime=2000 миллисекунд.

   Пример

   tickTime=2000

5. Добавьте в файл переменную initLimit — максимальное допустимое количество «тиков» (tickTime), которое может пройти, прежде чем узел кластера синхронизируется при запуске. При превышении будет сгенерирована ошибка.

   Пример

   initLimit=10

   Т. е. максимальное допустимое время на синхронизацию узла кластера при запуске будет равно (tickTime * initLimit) миллисекунд.

   Примечание

   Это обязательное свойство, без которого узел Zookeeper не будет запущен.

6. Добавьте в файл переменную syncLimit — максимальное допустимое количество «тиков» (tickTime) между отправкой запроса и получением ответа, прежде чем будет сгенерировано исключение.

   Пример

   syncLimit=5

   Примечание

   Это обязательное свойство, без которого узел Zookeeper не будет запущен.

На этом минимальное конфигурирование кластера Zookeeper завершается.

Расширенная конфигурация кластера#

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

• electionAlg — тип алгоритма для выбора лидера. Возможные значения:

  • 0 — версия алгоритма для выбора лидера — это оригинальная версия на основе UDP (User Datagram Protocol);

  • 1 — версия алгоритма для быстрого выбора лидера на основе UDP, неавторизованный;

  • 2 — версия алгоритма для быстрого выбора лидера на основе версии 1, но авторизованный;

  • 3 — версия алгоритма для быстрого выбора лидера на основе TCP.

  Версия 3 выбирается по умолчанию, версии 0, 1 и 2 носят статус устаревших и будут удалены в следующем релизе.

  Пример

  electionAlg=3

• leaderServes — опция разрешает клиентам подключаться к серверу-лидеру. Значение по умолчанию — yes. Одна из основных задач сервера-лидера — следить за обновлениями и координировать. Отключением этой опции (установкой значения no) можно добиться некоторого прироста производительности при координации обновлений.

  Пример

  leaderServes=yes

• Свойства, используемые совместно:

  • group.x=nnnn[:nnnn] — позволяет построить кворум с иерархической структурой, где x — это идентификатор кворума, а в качестве значения указывается список идентификаторов серверов, разделенных двоеточием.

    Примечание

    Группы не должны пересекаться, т. е. объединение всех групп должно давать полный список всех серверов.

  • weight.x=nnnnn — при использовании вместе с group.x=nnnn[:nnnn] позволяет назначить «вес» каждому серверу, который будет учитываться при выборе лидера. По умолчанию вес равен 1.

  Пример

  group.1=1:2:3
  group.2=4:5:6
  group.3=7:8:9
  weight.1=1
  weight.2=1
  weight.3=1
  weight.4=1
  weight.5=1
  weight.6=1
  weight.7=1
  weight.8=1
  weight.9=1
  

Следует учитывать, что запуск всех узлов кластера должен происходить за промежуток времени, не превышающий 10 тиков (heartbeat), заданных в строке initLimit=10. Поэтому лучше создать скрипт последовательного запуска всех узлов кластера с одной машины.

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

bin/zookeeper-server-start.sh -daemon
config/zookeeper.properties

Или с параллельным выводом файла логов в консоль:

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties && tail --f logs/zookeeper.out

Запуск всех узлов Zookeeper с одного сервера выполняется специальным скриптом:

bin/zookeeper-cluster-start.sh
config/zookeeper-cluster.properties

Где config/zookeeper-cluster.properties — файл конфигураций кластера с IP-адресами всех серверов, с которых производится запуск, а также пользователей, путей к дистрибутиву Corax и имен файлов конфигураций Zookeeper-брокеров. Формат файла приведен в таблице ниже.

Проверка работоспособности#

При запуске всех узлов кластера Zookeeper в файлах logs/zookeeper.out должны быть следующие записи:

• В файлах logs/zookeeper.out серверов-последователей:

  INFO FOLLOWING - LEADER ELECTION TOOK - 847
  (org.apache.zookeeper.server.quorum.Learner)
  

• В файле logs/zookeeper.out сервера-лидера:

  INFO LEADING - LEADER ELECTION TOOK - 408
  (org.apache.zookeeper.server.quorum.Leader)
  

Также в консоли сервера-лидера должна быть запись, информирующая о создании кворума в кластере с перечислением ID узлов:

INFO Have quorum of supporters, sids: [ 1,2 ]; starting up and setting
last processed zxid: 0x100000000
(org.apache.zookeeper.server.quorum.Leader)

Тестовый запуск кластера Zookeeper (2 узла)#

Тестовый запуск кластера Zookeeper из двух узлов производится на двух удаленных серверах:

Server-1: yyy.yyy.93.87
Server-2: yyy.yyy.93.88

Дополнительная информация:

• версия дистрибутива Corax: kafka\2.11-0.9.0.0;

• версия Zookeeper: zookeeper-3.4.6;

• имя пользователя: mon99usr;

• \$KAFKA_HOME: /home/mon99usr/kafka_2.11-0.9.0.0, где \$KAFKA_HOME — корневая директория дистрибутива Corax.

Для тестового запуска кластера Zookeeper выполните следующие действия:

1. На каждом из серверов создайте директорию, где будут храниться данные Zookeeper:

   mkdir --p /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/
   

2. На каждом из серверов создайте файл myid в директории /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/, созданной на предыдущем шаге (1). Запишите в файл соответствующий уникальный идентификатор узла Zookeeper:

   • на сервере Server-1 в файл myid запишите 1:

     echo 1 > /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     cat /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     1
     

   • на сервере Server-2 в файл myid запишите 2:

     echo 2 > /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     cat /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     2
     

3. Внесите следующие настройки в файл конфигурации \$KAFKA_HOME/config/zookeeper.config для обоих серверов:

   server.1=yyy.yyy.93.87:2888:3888
   server.2=yyy.yyy.93.88:2888:3888
   clientPort=3181
   dataDir=/var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/
   maxClientCnxns=60
   tickTime=2000
   initLimit=10
   syncLimit=5
   

   Где:

   • dataDir — директория для хранения данных Zookeeper, которая была создана на шаге (1).

   • server.1 — адрес Server-1; число 1 в server.1 — это myid сервера Server-1;

   • server.2 — адрес Server-2; число 2 в server.2 — это myid сервера Server-2.

4. Запустите кластер Zookeeper из двух узлов:

   • Выполните запуск узла Zookeeper на Server-1:

     bin/zookeeper-server-start.sh --daemon config/zookeeper.config
     && tail --f logs/zookeeper.out
     
     [mon99usr\@SBT-IPO-203 kafka_2.11-0.9.0.0] bin/zookeeper-server-start.sh config/zookeeper.properties
     
     [YYYY-MM-DD 15:25:19,172] INFO Reading configuration from: config/zookeeper.properties
     
     [YYYY-MM-DD 15:25:19,174] WARN No server failure will be tolerated. You need at least 3 servers
     
     [YYYY-MM-DD 15:25:19,174] INFO Defaulting to majority quorums
     
     [YYYY-MM-DD 15:25:19,179] INFO autopurge.snapRetainCount set to 3
     
     [YYYY-MM-DD 15:25:19,179] INFO autopurge.purgeInterval set to 0
     
     [YYYY-MM-DD 15:25:19,179] INFO Purge task is not scheduled.
     
     [YYYY-MM-DD 15:25:19,198] INFO Starting quorum peer
     
     [YYYY-MM-DD 15:25:19,210] INFO binding to port 0.0.0.0/0.0.0.0:3181
     
     [YYYY-MM-DD 15:25:19,234] INFO tickTime set to 2000
     
     [YYYY-MM-DD 15:25:19,234] INFO minSessionTimeout set to -1
     
     [YYYY-MM-DD 15:25:19,234] INFO maxSessionTimeout set to -1
     
     [YYYY-MM-DD 15:25:19,234] INFO initLimit set to 10
     
     [YYYY-MM-DD 15:25:19,247] INFO Reading snapshot /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.700000001
     
     [YYYY-MM-DD 15:25:19,260] INFO My election bind port: yyy.yyy.93.87:3888
     
     [YYYY-MM-DD 15:25:19,270] INFO New election. My id = 1, proposed zxid=0x700000001
     
     [YYYY-MM-DD 15:25:19,272] INFO Notification: 1 (message format version), 1 (n.leader), 0x700000001 (n.zxid), 0x1 (n.round), LOOKING (n.state), 1 (n.sid), 0x8 (n.peerEpoch) LOOKING (my state) (org.apache.zookeeper.server.quorum.FastLeaderElection)
     
     [YYYY-MM-DD 15:25:19,284] WARN Cannot open channel to 2 at election address /yyy.yyy.93.88:3888
     
     java.net.ConnectException: Connection refused
     
     at java.net.PlainSocketImpl.socketConnect(Native Method)
     
     ...
     
     [YYYY-MM-DD 15:25:19,892] INFO Notification time out: 800 (org.apache.zookeeper.server.quorum.FastLeaderElection)
     
     [YYYY-MM-DD 15:25:20,172] INFO Received connection request /yyy.yyy.93.88:45491 (org.apache.zookeeper.server.quorum.QuorumCnxManager)
     
     [YYYY-MM-DD 15:25:20,379] INFO FOLLOWING
     
     [YYYY-MM-DD 15:25:20,384] INFO TCP NoDelay set to: true
     
     [YYYY-MM-DD 15:25:20,390] INFO Server environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT
     
     [YYYY-MM-DD 15:25:20,390] INFO Server environment:host.name=SBT-IPO-203.youre.adress.ru
     
     [YYYY-MM-DD 15:25:20,390] INFO Server environment:java.version=1.8.0_60
     
     [YYYY-MM-DD 15:25:20,390] INFO Server environment:java.home=/usr/java/jre1.8.0_60
     
     [YYYY-MM-DD 15:25:20,390] INFO Server environment:java.class.path=:/home/mon99usr/kafka_2.11-0.9.0.0/bin/..
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:java.io.tmpdir=/tmp
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:java.compiler=\<NA\>
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:os.name=Linux
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:os.arch=amd64
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:os.version=3.10.0-123.9.3.el7.x86_64
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:user.name=mon99usr
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:user.home=/home/mon99usr
     
     [YYYY-MM-DD 15:25:20,391] INFO Server environment:user.dir=/home/mon99usr/kafka_2.11-0.9.0.0
     
     [YYYY-MM-DD 15:25:20,393] INFO Created server with tickTime 2000 minSessionTimeout 4000 maxSessionTimeout 40000 datadir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2 snapdir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2
     
     [YYYY-MM-DD 15:25:20,394] INFO FOLLOWING - LEADER ELECTION TOOK - 1124 (org.apache.zookeeper.server.quorum.Learner)
     
     [YYYY-MM-DD 15:25:20,421] INFO Getting a diff from the leader 0x700000001 (org.apache.zookeeper.server.quorum.Learner)
     
     [YYYY-MM-DD 15:25:20,425] INFO Snapshotting: 0x700000001 to /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.700000001
     

     Запуск выполнен успешно.

   • Выполните запуск узла Zookeeper на Server-2:

     bin/zookeeper-server-start.sh --daemon config/zookeeper.config
     && tail --f logs/zookeeper.out
     
     [mon99usr\@SBT-IPO-204 kafka_2.11-0.9.0.0]\$
     bin/zookeeper-server-start.sh config/zookeeper.properties
     
     [YYYY-MM-DD 15:25:19,874] INFO Reading configuration from:
     config/zookeeper.properties
     
     [YYYY-MM-DD 15:25:19,878] WARN No server failure will be tolerated. You need at least 3 servers.
     
     [YYYY-MM-DD 15:25:19,878] INFO Defaulting to majority quorums
     
     [YYYY-MM-DD 15:25:19,889] INFO autopurge.snapRetainCount set to 3
     
     [YYYY-MM-DD 15:25:19,889] INFO autopurge.purgeInterval set to 0
     
     [YYYY-MM-DD 15:25:19,889] INFO Purge task is not scheduled.
     
     [YYYY-MM-DD 15:25:19,941] INFO Starting quorum peer
     
     [YYYY-MM-DD 15:25:19,971] INFO binding to port 0.0.0.0/0.0.0.0:3181
     
     [YYYY-MM-DD 15:25:20,037] INFO tickTime set to 2000
     
     [YYYY-MM-DD 15:25:20,037] INFO minSessionTimeout set to -1
     
     [YYYY-MM-DD 15:25:20,037] INFO maxSessionTimeout set to -1
     
     [YYYY-MM-DD 15:25:20,038] INFO initLimit set to 10
     
     [YYYY-MM-DD 15:25:20,128] INFO My election bind port: /yyy.yyy.93.88:3888
     
     [YYYY-MM-DD 15:25:20,152] INFO LOOKING
     
     [YYYY-MM-DD 15:25:20,155] INFO New election. My id = 2, proposed zxid=0x700000001
     
     [YYYY-MM-DD 15:25:20,176] INFO Notification: 1 (message format version), 2 (n.leader), 0x700000001 (n.zxid), 0x1 (n.round), LOOKING (n.state), 2 (n.sid), 0x8 (n.peerEpoch) LOOKING (my state) (org.apache.zookeeper.server.quorum.FastLeaderElection)
     
     ...
     
     [YYYY-MM-DD 15:25:20,381] INFO LEADING
     
     [YYYY-MM-DD 15:25:20,386] INFO TCP NoDelay set to: true
     
     [YYYY-MM-DD 15:25:20,395] INFO Server environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT
     
     [YYYY-MM-DD 15:25:20,395] INFO Server environment:host.name=SBT-IPO-204.youre.adress.ru
     
     [YYYY-MM-DD 15:25:20,395] INFO Server environment:java.version=1.8.0_66
     
     [YYYY-MM-DD 15:25:20,395] INFO Server environment:java.home=/usr/java/jre1.8.0_66
     
     [YYYY-MM-DD 15:25:20,395] INFO Server environment:java.class.path=:/home/mon99usr/kafka_2.11-0.9.0.0/bin/../libs/scala- (org.apache.zookeeper.server.ZooKeeperServer)
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:java.io.tmpdir=/tmp
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:java.compiler=\<NA\>
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:os.name=Linux
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:os.arch=amd64
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:os.version=3.10.0-123.9.3.el7.x86_64
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:user.name=mon99usr
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:user.home=/home/mon99usr
     
     [YYYY-MM-DD 15:25:20,396] INFO Server environment:user.dir=/home/mon99usr/kafka_2.11-0.9.0.0
     
     [YYYY-MM-DD 15:25:20,397] INFO Created server with tickTime 2000 minSessionTimeout 4000 maxSessionTimeout 40000 datadir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2 snapdir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2 (org.apache.zookeeper.server.ZooKeeperServer)
     
     [YYYY-MM-DD 15:25:20,398] INFO LEADING - LEADER ELECTION TOOK - 243 (org.apache.zookeeper.server.quorum.Leader)
     
     [YYYY-MM-DD 15:25:20,410] INFO Follower sid: 1 : info : org.apache.zookeeper.server.quorum.QuorumPeer\$QuorumServer\@64c4b35a
     
     [YYYY-MM-DD 15:25:20,419] INFO Synchronizing with Follower sid: 1 maxCommittedLog=0x700000001 minCommittedLog=0x300000001 peerLastZxid=0x700000001
     
     [YYYY-MM-DD 15:25:20,419] INFO Sending DIFF
     
     [YYYY-MM-DD 15:25:20,431] INFO Received NEWLEADER-ACK message from 1
     
     [YYYY-MM-DD 15:25:20,433] INFO Have quorum of supporters, sids: [ 1,2 ]; starting up and setting last processed zxid: 0x900000000 (org.apache.zookeeper.server.quorum.Leader)
     

     Запуск выполнен успешно.

Пояснения к логам Zookeeper на сервере Server-1:

• WARN No server failure will be tolerated. You need at least 3 servers

  В случае отказа одного из серверов Zookeeper сервис Zookeeper не сможет продолжить свою работу, т. к. нарушено правило кворума о строгом большинстве доступных узлов. Т. е. число работоспособных узлов в Zookeeper-кластере всегда должно быть больше числа неработоспособных узлов. В данном случае узла два и отказ одного из узлов приведет к неработоспособности сервиса.

• WARN Cannot open channel to 2 at election address /yyy.yyy.93.88:3888 java.net.ConnectException: Connection refused

  Отказ при попытке соединения с Server-2, т. к. Server-2 в данный момент еще не был запущен.

• INFO Notification time out: 800 (org.apache.zookeeper.server.quorum.FastLeaderElection)

  Тайм-аут при попытке выбрать лидера.

• INFO Received connection request /yyy.yyy.93.88:45491

  Получен ответ на запрос о соединении с сервером Server-2.

• INFO Created server with tickTime 2000 minSessionTimeout 4000 maxSessionTimeout 40000 datadir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2 snapdir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2

  Описание конфигурации созданного сервера.

• INFO Snapshotting: 0x700000001 to /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.700000001

  Сохранение образа текущего состояния Zookeeper.

• INFO FOLLOWING

  После успешного соединения Server-1 с Server-2:

  • Server-1 становится сервером-последователем;

  • Server-2 становится сервером-лидером.

• FATAL Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer) kafka.common.InconsistentBrokerIdException: Configured brokerId 2 doesn\'t match stored brokerId 1 in meta.properties

  Данная ошибка запуска Zookeeper на сервере Server-2 происходит по причине конфликта в служебный файл Zookeeper meta.properties записался неверный идентификатор брокера broker.id=1, в то время как идентификатор myid сервера Server-2 указан, как равный двум. Решением проблемы будет удаление файла meta.properties, тогда при новом запуске Zookeeper этот файл будет сгенерирован автоматически и с корректными настройками.

Пояснения к логам Zookeeper на сервере Server-2:

• INFO LOOKING

  INFO New election. My id = 2, proposed zxid=0x700000001...

  INFO LEADING -- LEADING ELECTION TOOK -- 250

  Поиск сервера-лидера (т. к. сервер-лидер еще не выбран), голосование на выбор сервера-лидера (каждый из серверов предлагает себя), сервером-лидером выбран Server-2, а Server-1 автоматически становится сервером-последователем.

• WARN Cannot open channel to 2 at election address /yyy.yyy.93.88:3888

  (org.apache.zookeeper.server.quorum.QuorumCnxManager)

  java.net.ConnectException: Connection refused

  Попытка подключиться к отключенному серверу для выбора лидера. Сообщение возникает, если отключить Server-1 или Server-2 во время работы обоих серверов.

Тестовый запуск кластера Zookeeper (3 узла)#

Тестовый запуск кластера Zookeeper из трех узлов и кластера Corax из трех узлов будет производиться на трех удаленных серверах:

Server-1: yyy.yyy.93.87 [ 1 Брокер Zookeeper, 1 Брокер Kafka ]
Server-2: yyy.yyy.93.88 [ 1 Брокер Zookeeper, 1 Брокер Kafka ]
Server-3: yyy.yyy.97.221 [ 1 Брокер Zookeeper, 1 Брокер Kafka ]

Дополнительная информация:

• версия дистрибутива Corax: kafka\2.11-0.9.0.0;

• версия Zookeeper: zookeeper-3.4.6;

• имя пользователя: mon99usr;

• \$KAFKA_HOME: /home/mon99usr/kafka_2.11-0.9.0.0, где \$KAFKA_HOME — домашняя корневая директория дистрибутива Corax.

Конфигурация кластера Zookeeper из трех узлов

Каждый узел располагается на отдельном сервере.

Для конфигурации кластера Zookeeper выполните следующие действия:

1. На каждом из серверов создайте директорию, где будут храниться данные Zookeeper:

   mkdir --p /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/
   

2. На каждом из серверов создайте файл myid в директории /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/, созданной на предыдущем шаге (1). Это уникальный идентификатор каждого узла Zookeeper, запишите его в файл:

   • на сервере Server-1 в файл myid запишите 1:

     echo 1 > /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     cat /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     1
     

   • на сервере Server-2 в файл myid запишите 2:

     echo 2 > /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     cat /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     2
     

   • на сервере Server-3 в файл myid запишите 3:

     echo 3 > /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     cat /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/myid
     3
     

3. Внесите следующие настройки в файл конфигурации. \$KAFKA_HOME/config/zookeeper.config для трех серверов одинаков:

   server.1=yyy.yyy.93.87:2888:3888
   server.2=yyy.yyy.93.88:2888:3888
   server.3=yyy.yyy.97.221:2888:3888
   clientPort=3181
   dataDir=/var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/
   tickTime=2000
   initLimit=10
   syncLimit=5
   maxClientCnxns=60
   

   Где:

   • dataDir — директория для хранения данных Zookeeper, которая была создана на шаге (1);

   • server.1 — адрес Server-1; число 1 в server.1 — это myid сервера Server-1;

   • server.2 — адрес Server-2; число 2 в server.2 — это myid сервера Server-2;

   • server.3 — адрес Server-3; число 3 в Server.3 — это myid сервера Server-3.

Запуск кластера Zookeeper из трех узлов

1. Выполните запуск узла Zookeeper на сервере Server-1 в режиме демона с выводом логов:

   bin/zookeeper-server-start.sh -daemon
   config/zookeeper.properties && tail -f logs/zookeeper.out
   [mon99usr\@SBT-IPO-203 kafka_2.11-0.9.0.0]\$
   bin/zookeeper-server-start.sh -daemon config/zookeeper.properties &&
   tail -f logs/zookeeper.out
   
   [YYYY-MM-DD 12:56:43,375] INFO Reading configuration from: config/zookeeper.properties
   
   [YYYY-MM-DD 12:56:43,377] INFO Defaulting to majority quorums
   
   [YYYY-MM-DD 12:56:43,382] INFO autopurge.snapRetainCount set to 3
   
   [YYYY-MM-DD 12:56:43,382] INFO autopurge.purgeInterval set to 0
   
   [YYYY-MM-DD 12:56:43,382] INFO Purge task is not scheduled.
   
   [YYYY-MM-DD 12:56:43,411] INFO Starting quorum peer
   
   [YYYY-MM-DD 12:56:43,424] INFO binding to port 0.0.0.0/0.0.0.0:3181
   
   [YYYY-MM-DD 12:56:43,447] INFO tickTime set to 2000
   
   [YYYY-MM-DD 12:56:43,448] INFO minSessionTimeout set to -1
   
   [YYYY-MM-DD 12:56:43,448] INFO maxSessionTimeout set to -1
   
   [YYYY-MM-DD 12:56:43,448] INFO initLimit set to 10
   
   [YYYY-MM-DD 12:56:43,467] INFO Reading snapshot /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.30000000ac
   
   [YYYY-MM-DD 12:56:43,507] INFO My election bind port: /yyy.yyy.93.87:3888
   
   [YYYY-MM-DD 12:56:43,518] INFO LOOKING
   
   [YYYY-MM-DD 12:56:43,520] INFO New election. My id = 1, proposed zxid=0x310000003d
   
   [YYYY-MM-DD 12:56:43,522] INFO Notification: 1 (message format version), 1 (n.leader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 1 (n.sid), 0x31 (n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:56:43,530] WARN Cannot open channel to 2 at election address /yyy.yyy.93.88:3888
   
   java.net.ConnectException: Connection refused
   
   [YYYY-MM-DD 12:56:43,534] WARN Cannot open channel to 3 at election address /yyy.yyy.97.221:3888
   
   java.net.ConnectException: Connection refused
   
   [YYYY-MM-DD 12:56:56,388] INFO Received connection request /yyy.yyy.93.88:11287 (org.apache.zookeeper.server.quorum.QuorumCnxManager)
   
   [YYYY-MM-DD 12:56:56,595] INFO FOLLOWING (org.apache.zookeeper.server.quorum.QuorumPeer)
   
   [YYYY-MM-DD 12:56:56,600] INFO TCP NoDelay set to: true
   
   [YYYY-MM-DD 12:56:56,606] INFO Server environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:host.name=SBT-IPO-203.youre.adress.ru
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:java.version=1.8.0_60
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:java.home=/usr/java/jre1.8.0_60
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:java.class.path=:/home/mon99usr/kafka_2.11-0.9.0.0/bin/../....
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:java.library.path=/usr/java/packages/...
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:java.io.tmpdir=/tmp
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:java.compiler=\<NA\>
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:os.name=Linux
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:os.arch=amd64
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:os.version=3.10.0-123.9.3.el7.x86_64
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:user.name=mon99usr
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:user.home=/home/mon99usr
   
   [YYYY-MM-DD 12:56:56,607] INFO Server environment:user.dir=/home/mon99usr/kafka_2.11-0.9.0.0
   
   [YYYY-MM-DD 12:56:56,609] INFO Created server with tickTime 2000 minSessionTimeout 4000 maxSessionTimeout 40000 Datadir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2 snapdir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2
   
   [YYYY-MM-DD 12:56:56,609] INFO FOLLOWING - LEADER ELECTION TOOK - 13090 (org.apache.zookeeper.server.quorum.Learner)
   
   [YYYY-MM-DD 12:56:56,616] WARN Unexpected exception, tries=0, connecting to /yyy.yyy.93.88:2888
   
   java.net.ConnectException: Connection refused
   
   ...
   
   [YYYY-MM-DD 12:56:57,640] INFO Getting a diff from the leader 0x310000003d
   
   [YYYY-MM-DD 12:56:57,645] INFO Snapshotting: 0x310000003d to /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.310000003d
   
   [YYYY-MM-DD 12:57:04,004] WARN Got zxid 0x3200000001 expected 0x1
   
   [YYYY-MM-DD 12:57:04,004] INFO Creating new log file: log.3200000001
   
   [YYYY-MM-DD 12:57:06,472] INFO Received connection request /yyy.yyy.97.221:47171
   
   [YYYY-MM-DD 12:57:06,475] INFO Notification: 1 (message format version), 3 (n.leader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 3 (n.sid), 0x31 (n.peerEpoch) FOLLOWING (my state)
   

   Где tail -f logs/zookeeper.out — вывод текущего лога Zookeeper-брокера из файла в реальном времени.

   Запуск Zookeeper-брокера на Server-1 выполнен успешно.

   Краткий анализ логов

   • Поиск других Брокеров с целью выбора лидера, отказ соединения с Брокером на сервере yyy.yyy.93.88 и с Брокером на сервере yyy.yyy.97.221, т. к. Брокеры на этих серверах еще не запущены:

     INFO LOOKING
     
     WARN Cannot open channel to 2 at election address /yyy.yyy.93.88:3888
     
     java.net.ConnectException: Connection refused
     
     WARN Cannot open channel to 3 at election address /yyy.yyy.97.221:3888
     
     java.net.ConnectException: Connection refused
     

   • Принят запрос на соединение от Брокера на сервере Server-2 по адресу yyy.yyy.93.88 в связи со запуском этого Брокера:

     INFO Received connection request /yyy.yyy.93.88:11287
     

     Как правило, после этого сообщения новый Брокер включается в кластер.

   • Выбран сервер-лидер — это только что запущенный Брокер на сервере Server-2. Брокер на текущем сервере Server-1 выбран в качестве сервера-последователя (FOLLOWER):

     INFO FOLLOWING
     

   • Синхронизация Брокера-последователя на сервере Server-1 с Брокером-лидером на сервере Server-2: получение от сервера-лидера Server-2 текущего состояния системы (diff). В результате на сервере Server-1 будет создан и сохранен соответствующий образ (snapshot) только что полученных данных:

     INFO Getting a diff from the leader 0x310000003d
     
     INFO Snapshotting: 0x310000003d to /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.310000003d
     

   • Принят запрос на соединение от Брокера на сервере Server-3 по адресу yyy.yyy.97.221 в связи с запуском этого Брокера:

     INFO Received connection request /yyy.yyy.97.221:11287
     

     С этого момента новый Брокер включается в кластер.

   • После сообщения INFO Notification: 1... вывод логов прекращается в связи с отсутствием дальнейших событий.

2. Выполните запуск узла Zookeeper на сервере Server-2 в режиме демона с выводом логов:

   bin/zookeeper-server-start.sh -daemon
   config/zookeeper.properties && tail -f logs/zookeeper.out
   [mon99usr\@SBT-IPO-204 kafka_2.11-0.9.0.0]\$
   bin/zookeeper-server-start.sh -daemon config/zookeeper.properties &&
   tail -f logs/zookeeper.out
   
   [YYYY-MM-DD 12:56:56,055] INFO Reading configuration from:
   config/zookeeper.properties
   
   [YYYY-MM-DD 12:56:56,058] INFO Defaulting to majority quorums
   
   [YYYY-MM-DD 12:56:56,067] INFO autopurge.snapRetainCount set to 3
   
   [YYYY-MM-DD 12:56:56,067] INFO autopurge.purgeInterval set to 0
   
   [YYYY-MM-DD 12:56:56,067] INFO Purge task is not scheduled.
   
   [YYYY-MM-DD 12:56:56,116] INFO Starting quorum peer
   
   [YYYY-MM-DD 12:56:56,135] INFO binding to port 0.0.0.0/0.0.0.0:3181
   
   [YYYY-MM-DD 12:56:56,184] INFO tickTime set to 2000
   
   [YYYY-MM-DD 12:56:56,185] INFO minSessionTimeout set to -1
   
   [YYYY-MM-DD 12:56:56,185] INFO maxSessionTimeout set to -1
   
   [YYYY-MM-DD 12:56:56,185] INFO initLimit set to 10
   
   [YYYY-MM-DD 12:56:56,211] INFO Reading snapshot
   /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.2a000000a9
   
   [YYYY-MM-DD 12:56:56,375] INFO My election bind port:
   /yyy.yyy.93.88:3888
   
   [YYYY-MM-DD 12:56:56,378] INFO LOOKING
   (org.apache.zookeeper.server.quorum.QuorumPeer)
   
   [YYYY-MM-DD 12:56:56,380] INFO New election. My id = 2, proposed zxid=0x310000003d
   
   [YYYY-MM-DD 12:56:56,394] INFO Notification: 1 (message format
   version), 2 (n.leader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING
   (n.state), 2 (n.sid), 0x31 (n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:56:56,396] WARN Cannot open channel to 3 at election address /yyy.yyy.97.221:3888
   
   java.net.ConnectException: Connection refused
   
   at java.net.PlainSocketImpl.socketConnect(Native Method)
   
   ...
   
   [YYYY-MM-DD 12:56:56,398] INFO Notification: 1 (message format
   version), 1 (n.leader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 1 (n.sid), 0x31 (n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:56:56,399] INFO Notification: 1 (message format
   version), 2 (n.leader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 1 (n.sid), 0x31 (n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:56:56,601] INFO LEADING
   
   [YYYY-MM-DD 12:56:56,605] INFO TCP NoDelay set to: true
   
   [YYYY-MM-DD 12:56:56,615] INFO Server environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT
   
   [YYYY-MM-DD 12:56:56,615] INFO Server environment:host.name=SBT-IPO-204.youre.adress.ru
   
   [YYYY-MM-DD 12:56:56,615] INFO Server environment:java.version=1.8.0_66
   
   [YYYY-MM-DD 12:56:56,615] INFO Server environment:java.home=/usr/java/jre1.8.0_66
   
   [YYYY-MM-DD 12:56:56,615] INFO Server environment:java.class.path=:/home/mon99usr/kafka_2.11-0.9.0.0/bin/....
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:java.library.path=/usr/java/...
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:java.io.tmpdir=/tmp
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:java.compiler=\<NA\>
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:os.name=Linux
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:os.arch=amd64
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:os.version=3.10.0-123.9.3.el7.x86_64
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:user.name=mon99usr
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:user.home=/home/mon99usr
   
   [YYYY-MM-DD 12:56:56,616] INFO Server environment:user.dir=/home/mon99usr/kafka_2.11-0.9.0.0
   
   [YYYY-MM-DD 12:56:56,618] INFO Created server with tickTime 2000 minSessionTimeout 4000 maxSessionTimeout 40000
   
   datadir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2 snapdir /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2
   
   [YYYY-MM-DD 12:56:56,619] INFO LEADING - LEADER ELECTION TOOK - 239
   
   [YYYY-MM-DD 12:56:57,627] INFO Follower sid: 1 : info :
   org.apache.zookeeper.server.quorum.QuorumPeer\$...
   
   [YYYY-MM-DD 12:56:57,638] INFO Synchronizing with Follower sid: 1
   maxCommittedLog=0x310000003d minCommittedLog=0x2e0000006f
   peerLastZxid=0x310000003d
   
   [YYYY-MM-DD 12:56:57,638] INFO Sending DIFF
   
   [YYYY-MM-DD 12:56:57,666] INFO Received NEWLEADER-ACK message from 1
   
   [YYYY-MM-DD 12:56:57,669] INFO Have quorum of supporters, sids: [ 1,2]; starting up and setting last processed zxid: ..
   
   [YYYY-MM-DD 12:57:04,000] INFO Expiring session 0x253bc816e870000, timeout of 6000ms exceeded
   
   [YYYY-MM-DD 12:57:04,001] INFO Processed session termination for sessionid: 0x253bc816e870000
   
   [YYYY-MM-DD 12:57:04,002] INFO Creating new log file: log.3200000001
   
   [YYYY-MM-DD 12:57:06,475] INFO Received connection request
   /yyy.yyy.97.221:34164
   
   [YYYY-MM-DD 12:57:06,477] INFO Notification: 1 (message format
   version), 3 (n.leader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 3 (n.sid), 0x31 (n.peerEpoch) LEADING (my state)
   
   [YYYY-MM-DD 12:57:06,513] INFO Follower sid: 3 : info :
   org.apache.zookeeper.server.quorum.QuorumPeer\$QuorumServer@...
   
   [YYYY-MM-DD 12:57:06,517] INFO Synchronizing with Follower sid: 3
   maxCommittedLog=0x3200000001 minCommittedLog=0x2e00000070
   peerLastZxid=0x310000003d
   
   [YYYY-MM-DD 12:57:06,518] INFO Sending DIFF
   
   [YYYY-MM-DD 12:57:06,540] INFO Received NEWLEADER-ACK message from 3
   

   Запуск брокера на сервере Server-2 выполнен успешно.

   Краткий анализ логов

   • Поиск других Брокеров с целью выбора лидера, отказ соединения с Брокером на сервере yyy.yyy.97.221, т. к. Брокер на этом сервере еще не запущен:

     INFO LOOKING
     
     WARN Cannot open channel to 3 at election address /yyy.yyy.97.221:3888
     
     java.net.ConnectException: Connection refused
     

   • Лидером выбран Брокер на сервере Server-2 (на выбор лидера ушло 239 миллисекунд). От Брокера на сервере Server-1 получено соответствующее подтверждение. Выведено сообщение, что создан новый кворум [1,2] из двух источников-брокеров с идентификаторами myid = 1 и 2 соответственно.

     INFO LEADING
     
     INFO LEADING - LEADER ELECTION TOOK -- 239
     
     INFO Have quorum of supporters, sids: [ 1,2 ];
     

   • Синхронизация Брокера-последователя на сервере Server-1 с Брокером-лидером на сервере Server-2. Брокер-лидер отправляет текущее состояние системы (DIFF) и тем самым запускает синхронизацию. После обработки DIFF Брокером-последователем (сохранения DIFF в виде образа (snapshot) текущего состояния системы) от Брокера-последователя поступает соответствующее подтверждение о получении Received NEWLEADER-ACK:

     INFO Follower sid: 1 : info : org.apache.zookeeper.server.quorum.Q...
     
     INFO Synchronizing with Follower sid: 1
     
     INFO Sending DIFF
     
     INFO Received NEWLEADER-ACK message from 1
     

   • Принят запрос на соединение от нового Брокера на сервере Server-3 по адресу yyy.yyy.97.221 в связи со запуском этого Брокера:

     INFO Received connection request /yyy.yyy.97.221:34164
     

   • Новый Брокер с идентификатором myid=3 принят в кворум в качестве Брокера-последователя:

     INFO Follower sid: 3
     

   • Синхронизация Брокера-последователя на сервере Server-3 с Брокером-лидером на сервере Server-2. Брокеру-последователю отправлено текущее состояние системы (DIFF), успешно обработано Брокером-последователем, в ответ выслано соответствующее подтверждение:

     INFO Synchronizing with Follower sid: 3 maxCommittedLog=0x3200000001 minCommittedLog=0x2e00000070 peerLastZxid=0x310000003d
     
     [YYYY-MM-DD 12:57:06,518] INFO Sending DIFF
     
     [YYYY-MM-DD 12:57:06,540] INFO Received NEWLEADER-ACK message from 3
     

3. Выполните запуск узла Zookeeper на сервере Server-3 в режиме демона с выводом логов:

   > bin/zookeeper-server-start.sh -daemon
   
   config/zookeeper.properties && tail -f logs/zookeeper.out
   
   [mon99usr\@SBT-IPO-208 kafka_2.11-0.9.0.0]\$
   bin/zookeeper-server-start.sh -daemon config/zookeeper.properties && tail -f logs/zookeeper.out
   
   [YYYY-MM-DD 12:57:07,273] INFO Reading configuration from:
   config/zookeeper.properties
   
   [YYYY-MM-DD 12:57:07,276] INFO Defaulting to majority quorums
   
   [YYYY-MM-DD 12:57:07,282] INFO autopurge.snapRetainCount set to 3
   
   [YYYY-MM-DD 12:57:07,282] INFO autopurge.purgeInterval set to 0
   
   [YYYY-MM-DD 12:57:07,282] INFO Purge task is not scheduled.
   
   [YYYY-MM-DD 12:57:07,303] INFO Starting quorum peer
   
   [YYYY-MM-DD 12:57:07,321] INFO binding to port 0.0.0.0/0.0.0.0:3181
   
   [YYYY-MM-DD 12:57:07,357] INFO tickTime set to 2000
   
   [YYYY-MM-DD 12:57:07,357] INFO minSessionTimeout set to -1
   
   [YYYY-MM-DD 12:57:07,357] INFO maxSessionTimeout set to -1
   
   [YYYY-MM-DD 12:57:07,357] INFO initLimit set to 10
   
   [YYYY-MM-DD 12:57:07,378] INFO Reading snapshot
   /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2/snapshot.30000000ac
   
   [YYYY-MM-DD 12:57:07,430] INFO My election bind port:
   /yyy.yyy.97.221:3888
   
   [YYYY-MM-DD 12:57:07,444] INFO LOOKING
   
   [YYYY-MM-DD 12:57:07,445] INFO New election. My id = 3, proposed zxid=0x31000003d
   
   [YYYY-MM-DD 12:57:07,459] INFO Notification: 1 (message format
   version), 3 (n.lader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 3 (n.sid), 0x31(n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:57:07,460] INFO Notification: 1 (message format
   version), 2 (n.lader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 1 (n.sid), 0x31(n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:57:07,460] INFO Notification: 1 (message format
   version), 2 (n.lader), 0x310000003d (n.zxid), 0x1 (n.round), FOLLOWING (n.state), 1 (n.sid), 0x2 (n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:57:07,461] INFO Notification: 1 (message format
   version), 2 (n.lader), 0x310000003d (n.zxid), 0x1 (n.round), LOOKING (n.state), 2 (n.sid), 0x31(n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:57:07,462] INFO Notification: 1 (message format
   version), 2 (n.lader), 0x310000003d (n.zxid), 0x1 (n.round), LEADING (n.state), 2 (n.sid), 0x32(n.peerEpoch) LOOKING (my state)
   
   [YYYY-MM-DD 12:57:07,462] INFO FOLLOWING (org.apache.zookeeper.server.quorum.QuorumPeer)
   
   [YYYY-MM-DD 12:57:07,469] INFO TCP NoDelay set to: true
   
   [YYYY-MM-DD 12:57:07,478] INFO Server
   environment:zookeeper.version=3.4.6-156995, built on 02/20/2014 09:09 GMT
   
   [YYYY-MM-DD 12:57:07,478] INFO Server environment:host.name=SBT-IPO-208.youre.adress.ru
   
   [YYYY-MM-DD 12:57:07,487] INFO Server environment:java.version=1.8.0_71
   
   [YYYY-MM-DD 12:57:07,487] INFO Server
   environment:java.home=/usr/java/jre1.8.0_1
   
   [YYYY-MM-DD 12:57:07,487] INFO Server
   environment:java.class.path=:/home/mon99ur/...
   
   [YYYY-MM-DD 12:57:07,488] INFO Server
   environment:java.library.path=/usr/java/packages/...
   
   [YYYY-MM-DD 12:57:07,488] INFO Server environment:java.io.tmpdir=/tmp
   
   [YYYY-MM-DD 12:57:07,488] INFO Server environment:java.compiler=\<NA\>
   
   [YYYY-MM-DD 12:57:07,488] INFO Server environment:os.name=Linux
   
   [YYYY-MM-DD 12:57:07,488] INFO Server environment:os.arch=amd64
   
   [YYYY-MM-DD 12:57:07,488] INFO Server
   environment:os.version=3.10.0-123.9.3.el7x86_64
   
   [YYYY-MM-DD 12:57:07,488] INFO Server environment:user.name=mon99usr
   
   [YYYY-MM-DD 12:57:07,488] INFO Server
   environment:user.home=/home/mon99usr
   
   [YYYY-MM-DD 12:57:07,488] INFO Server
   environment:user.dir=/home/mon99usr/kafka2.11-0.9.0.0
   
   [YYYY-MM-DD 12:57:07,490] INFO Created server with tickTime 2000
   minSessionTimeout 4000 maxSessionTimeout 40000 datadir
   
   /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2 snapdir
   /var/tmp/kafka_2.11-0.9.0.0/zookeeper-data/version-2
   
   [YYYY-MM-DD 12:57:07,491] INFO FOLLOWING - LEADER ELECTION TOOK - 46
   
   [YYYY-MM-DD 12:57:07,502] INFO Getting a diff from the leader
   0x3200000001
   
   [YYYY-MM-DD 12:57:07,506] WARN Got zxid 0x3200000001 expected 0x1
   
   [YYYY-MM-DD 12:57:07,508] INFO Snapshotting: 0x3200000001 to
   /var/tmp/kafka_2.1-0.9.0.0/zookeeper-data/version-2/snapshot.3200000001
   

   Запуск брокера на сервере Server-3 выполнен успешно.

   Краткий анализ логов

   • Поиск других Брокеров с целью выбора лидера, инициация алгоритма выбора лидера, предложение своей кандидатуры в качестве лидера аналогичны инициализации двух предыдущих Брокеров:

     INFO LOOKING
     
     INFO New election. My id = 3, proposed zxid=0x31000003d
     

   • Брокер на сервере Server-3 выбран в качестве Брокера-последователя:

     INFO FOLLOWING
     
     INFO FOLLOWING - LEADER ELECTION TOOK -- 46
     

   • Синхронизация с Брокером на сервере Server-2. Получение состояния системы и сохранение образа:

     INFO Getting a diff from the leader 0x3200000001
     
     INFO Snapshotting: 0x3200000001 to /var/tmp/kafka_2.1-0.9.0.0/zookeeper-data/version-2/snapshot.3200000001
     

Конфигурация кластера Corax из трех узлов

Каждый узел расположен на отдельном сервере.

В целях тестирования доступности сервиса Zookeeper с различным количеством работающих узлов сконфигурируйте кластер Corax:

1. На каждом из серверов создайте директорию, где будут храниться логи Corax:

   mkdir --p /var/tmp/kafka_2.11-0.9.0.0/kafka-logs/
   

2. Создайте файл конфигурации \$KAFKA_HOME/config/server.config для каждого Corax-брокера.

   • Для сервера Server-1:

     broker.id=1
     
     host.name=yyy.yyy.93.87
     
     log.dirs=/var/tmp/kafka_2.11-0.9.0.0/kafka-logs
     
     zookeeper.connect=yyy.yyy.93.87:3181,yyy.yyy.93.88:3181,yyy.yyy.97.221:3181
     

   • Для сервера Server-2:

     broker.id=2
     
     host.name=yyy.yyy.93.88
     
     log.dirs=/var/tmp/kafka_2.11-0.9.0.0/kafka-logs
     
     zookeeper.connect=yyy.yyy.93.87:3181,yyy.yyy.93.88:3181,yyy.yyy.97.221:3181
     

   • Для сервера Server-3:

     broker.id=3
     
     host.name=yyy.yyy.97.221
     
     log.dirs=/var/tmp/kafka_2.11-0.9.0.0/kafka-logs
     
     zookeeper.connect=yyy.yyy.93.87:3181,yyy.yyy.93.88:3181,yyy.yyy.97.221:3181
     

   Где:

   • log.dirs — директория, где хранятся логи Corax, которая была создана на шаге (1);

   • host.name — IP-адрес сервера, на котором находится Corax-брокер;

   • broker.id — идентификатор Corax-брокера, обязательно должен быть уникальным в рамках кластера Corax;

   • zookeeper.connect — список адресов Zookeeper-брокеров в формате IP:PORT, где:

     • IP — адрес сервера, на котором выполняется Zookeeper-брокер;

     • PORT — порт, к которому подключаются клиенты Zookeeper — его clientPort.

   Примечание

   Свойство port не указано в файлах, т. к. нет необходимости прописывать его явно: port=9092 используется Corax по умолчанию.

   При необходимости значение порта можно прописать явно или изменить на другое.

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

1. Выполните запуск узла Corax на сервере Server-1 в режиме демона с выводом логов:

   bin/kafka-server-start.sh -daemon
   
   config/server.properties && tail -f logs/server.log
   
   [mon99usr\@SBT-IPO-203 kafka_2.11-0.9.0.0]\$ bin/kafka-server-start.sh -daemon config/server.properties && tail -f logs/server.log
   
   [YYYY-MM-DD 17:16:50,514] INFO [Group Metadata Manager on Broker 1]: Removed 0 expired offsets in 11 milliseconds.
   
   [YYYY-MM-DD 17:16:50,531] INFO [ThrottledRequestReaper-Produce], Starting (kafka.server.ClientQuotaManager\$ThrottledRequestReaper)
   
   [YYYY-MM-DD 17:16:50,532] INFO [ThrottledRequestReaper-Fetch], Starting (kafka.server.ClientQuotaManager\$ThrottledRequestReaper)
   
   [YYYY-MM-DD 17:16:50,537] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader\$)
   
   [YYYY-MM-DD 17:16:50,549] INFO Creating /brokers/ids/1 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:16:50,555] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:16:50,556] INFO Registered broker 1 at path /brokers/ids/1 with addresses: PLAINTEXT -> EndPoint(yyy.yyy.93.87,9092,PLAINTEXT) (kafka.utils.ZkUtils)
   
   [YYYY-MM-DD 17:16:50,571] INFO [Kafka Server 1], started (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:17:06,309] INFO KafkaConfig values:
   
   advertised.host.name = null
   
   metric.reporters = []
   
   quota.producer.default = 9223372036854775807
   
   offsets.topic.num.partitions = 50
   
   log.flush.interval.messages = 9223372036854775807
   
   auto.create.topics.enable = true
   
   controller.socket.timeout.ms = 30000
   
   log.flush.interval.ms = null
   
   principal.builder.class = class org.apache.kafka.common.security.auth.DefaultPrincipalBuilder
   
   replica.socket.receive.buffer.bytes = 65536
   
   min.insync.replicas = 1
   
   replica.fetch.wait.max.ms = 500
   
   num.recovery.threads.per.data.dir = 1
   
   ssl.keystore.type = JKS
   
   default.replication.factor = 1
   
   ssl.truststore.password = null
   
   log.preallocate = false
   
   sasl.kerberos.principal.to.local.rules = [DEFAULT]
   
   fetch.purgatory.purge.interval.requests = 1000
   
   ssl.endpoint.identification.algorithm = null
   
   replica.socket.timeout.ms = 30000
   
   message.max.bytes = 1000012
   
   num.io.threads = 8
   
   offsets.commit.required.acks = -1
   
   log.flush.offset.checkpoint.interval.ms = 60000
   
   delete.topic.enable = false
   
   quota.window.size.seconds = 1
   
   ssl.truststore.type = JKS
   
   offsets.commit.timeout.ms = 5000
   
   quota.window.num = 11
   
   zookeeper.connect = yyy.yyy.93.87:3181,yyy.yyy.93.88:3181,yyy.yyy.97.221:3181
   
   authorizer.class.name =
   
   num.replica.fetchers = 1
   
   log.retention.ms = null
   
   log.roll.jitter.hours = 0
   
   log.cleaner.enable = false
   
   offsets.load.buffer.size = 5242880
   
   log.cleaner.delete.retention.ms = 86400000
   
   ssl.client.auth = none
   
   controlled.shutdown.max.retries = 3
   
   queued.max.requests = 500
   
   offsets.topic.replication.factor = 3
   
   log.cleaner.threads = 1
   
   sasl.kerberos.service.name = null
   
   sasl.kerberos.ticket.renew.jitter = 0.05
   
   socket.request.max.bytes = 104857600
   
   ssl.trustmanager.algorithm = PKIX
   
   zookeeper.session.timeout.ms = 6000
   
   log.retention.bytes = -1
   
   sasl.kerberos.min.time.before.relogin = 60000
   
   zookeeper.set.acl = false
   
   connections.max.idle.ms = 600000
   
   offsets.retention.minutes = 1440
   
   replica.fetch.backoff.ms = 1000
   
   inter.broker.protocol.version = 0.9.0.X
   
   log.retention.hours = 168
   
   num.partitions = 1
   
   listeners = null
   
   ssl.provider = null
   
   ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
   
   log.roll.ms = null
   
   log.flush.scheduler.interval.ms = 9223372036854775807
   
   ssl.cipher.suites = null
   
   log.index.size.max.bytes = 10485760
   
   ssl.keymanager.algorithm = SunX509
   
   security.inter.broker.protocol = PLAINTEXT
   
   replica.fetch.max.bytes = 1048576
   
   advertised.port = null
   
   log.cleaner.dedupe.buffer.size = 524288000
   
   replica.high.watermark.checkpoint.interval.ms = 5000
   
   log.cleaner.io.buffer.size = 524288
   
   sasl.kerberos.ticket.renew.window.factor = 0.8
   
   zookeeper.connection.timeout.ms = null
   
   controlled.shutdown.retry.backoff.ms = 5000
   
   log.roll.hours = 168
   
   log.cleanup.policy = delete
   
   host.name = yyy.yyy.93.87
   
   log.roll.jitter.ms = null
   
   max.connections.per.ip = 2147483647
   
   offsets.topic.segment.bytes = 104857600
   
   background.threads = 10
   
   quota.consumer.default = 9223372036854775807
   
   request.timeout.ms = 30000
   
   log.index.interval.bytes = 4096
   
   log.dir = /tmp/kafka-logs
   
   log.segment.bytes = 1073741824
   
   log.cleaner.backoff.ms = 15000
   
   offset.metadata.max.bytes = 4096
   
   ssl.truststore.location = null
   
   group.max.session.timeout.ms = 30000
   
   ssl.keystore.password = null
   
   zookeeper.sync.time.ms = 2000
   
   port = 9092
   
   log.retention.minutes = null
   
   log.segment.delete.delay.ms = 60000
   
   log.dirs = /var/tmp/kafka_2.11-0.9.0.0/kafka-logs
   
   controlled.shutdown.enable = true
   
   compression.type = producer
   
   max.connections.per.ip.overrides =
   
   sasl.kerberos.kinit.cmd = /usr/bin/kinit
   
   log.cleaner.io.max.bytes.per.second = 1.7976931348623157E308
   
   auto.leader.rebalance.enable = true
   
   leader.imbalance.check.interval.seconds = 300
   
   log.cleaner.min.cleanable.ratio = 0.5
   
   replica.lag.time.max.ms = 10000
   
   num.network.threads = 3
   
   ssl.key.password = null
   
   reserved.broker.max.id = 1000
   
   metrics.num.samples = 2
   
   socket.send.buffer.bytes = 102400
   
   ssl.protocol = TLS
   
   socket.receive.buffer.bytes = 102400
   
   ssl.keystore.location = null
   
   replica.fetch.min.bytes = 1
   
   unclean.leader.election.enable = true
   
   group.min.session.timeout.ms = 6000
   
   log.cleaner.io.buffer.load.factor = 0.9
   
   offsets.retention.check.interval.ms = 600000
   
   producer.purgatory.purge.interval.requests = 1000
   
   metrics.sample.window.ms = 30000
   
   broker.id = 1
   
   offsets.topic.compression.codec = 0
   
   log.retention.check.interval.ms = 300000
   
   advertised.listeners = null
   
   leader.imbalance.per.broker.percentage = 10
   
   (kafka.server.KafkaConfig)
   
   [YYYY-MM-DD 17:17:06,377] INFO starting (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:17:06,382] INFO Connecting to zookeeper on yyy.yyy.93.87:3181,yyy.yyy.93.88:3181,yyy.yyy.97.221:3181 (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:17:06,748] INFO Loading logs. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:17:06,986] WARN Found an corrupted index file, /var/tmp/kafka_2.11-0.9.0.0/kafka-logs/NUMBERS-0/00000000000000000000.index, deleting and rebuilding index... (kafka.log.Log)
   
   [YYYY-MM-DD 17:17:06,987] INFO Recovering unflushed segment 0 in log NUMBERS-0. (kafka.log.Log)
   
   [YYYY-MM-DD 17:17:06,988] INFO Completed load of log NUMBERS-0 with log end offset 13 (kafka.log.Log)
   
   [YYYY-MM-DD 17:17:06,990] WARN Found an corrupted index file, /var/tmp/kafka_2.11-0.9.0.0/kafka-logs/NUMBERS-1/00000000000000000000.index, deleting and rebuilding index... (kafka.log.Log)
   
   [YYYY-MM-DD 17:17:06,991] INFO Recovering unflushed segment 0 in log NUMBERS-1. (kafka.log.Log)
   
   [YYYY-MM-DD 17:17:06,992] INFO Completed load of log NUMBERS-1 with log end offset 14 (kafka.log.Log)
   
   [YYYY-MM-DD 17:17:06,994] INFO Logs loading complete. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:17:06,995] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:17:06,996] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:17:07,040] INFO Awaiting socket connections on SBT-IPO-203.youre.adress.ru:9092. (kafka.network.Acceptor)
   
   [YYYY-MM-DD 17:17:07,043] INFO [Socket Server on Broker 1], Started 1 acceptor threads (kafka.network.SocketServer)
   
   [YYYY-MM-DD 17:17:07,062] INFO [ExpirationReaper-1], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:17:07,063] INFO [ExpirationReaper-1], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:17:07,124] INFO Creating /controller (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:17:07,142] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:17:07,142] INFO 1 successfully elected as leader (kafka.server.ZookeeperLeaderElector)
   
   [YYYY-MM-DD 17:17:09,020] INFO New leader is 1 (kafka.server.ZookeeperLeaderElector\$LeaderChangeListener)
   
   [YYYY-MM-DD 17:17:09,051] INFO [GroupCoordinator 1]: Starting up. (kafka.coordinator.GroupCoordinator)
   
   [YYYY-MM-DD 17:17:09,053] INFO [ExpirationReaper-1], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:17:09,053] INFO [GroupCoordinator 1]: Startup complete. (kafka.coordinator.GroupCoordinator)
   
   [YYYY-MM-DD 17:17:09,057] INFO [ExpirationReaper-1], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:17:09,060] INFO [Group Metadata Manager on Broker 1]: Removed 0 expired offsets in 9 milliseconds. (kafka.coordinator.GroupMetadataManager)
   
   [YYYY-MM-DD 17:17:09,075] INFO [ThrottledRequestReaper-Produce], Starting (kafka.server.ClientQuotaManager\$ThrottledRequestReaper)
   
   [YYYY-MM-DD 17:17:09,084] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader\$)
   
   [YYYY-MM-DD 17:17:09,128] INFO Creating /brokers/ids/1 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:17:09,138] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:17:09,139] INFO Registered broker 1 at path /brokers/ids/1 with addresses: PLAINTEXT -\> EndPoint(yyy.yyy.93.87,9092,PLAINTEXT) (kafka.utils.ZkUtils)
   
   [YYYY-MM-DD 17:17:09,151] INFO [Kafka Server 1], started (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:17:09,409] INFO [ReplicaFetcherManager on broker 1] Removed fetcher for partitions [NUMBERS,0],[NUMBERS,1] (kafka.server.ReplicaFetcherManager)
   
   [YYYY-MM-DD 17:17:10,584] INFO [ReplicaFetcherManager on broker 1] Removed fetcher for partitions[NUMBERS,0] (kafka.server.ReplicaFetcherManager)
   

   Где tail -f logs/server.log — вывод текущего лога Corax-брокера из файла в реальном времени.

   Запуск Corax-брокера на сервере Server-1 выполнен успешно.

2. Выполните запуск узла Zookeeper на сервере Server-2 в режиме демона с выводом логов:

   bin/kafka-server-start.sh -daemon
   
   config/server.properties && tail -f logs/server.log
   
   [mon99usr\@SBT-IPO-204 kafka_2.11-0.9.0.0]\$ bin/kafka-server-start.sh -daemon config/server.properties && tail -f logs/server.log
   
   [YYYY-MM-DD 17:35:35,726] INFO [Group Metadata Manager on Broker 2]: Loading offsets and group metadata from [consumer_offsets,36]
   
   [YYYY-MM-DD 17:35:35,728] INFO [Group Metadata Manager on Broker 2]: Finished loading offsets from [consumer_offsets,36] in 2 milliseconds.
   
   [YYYY-MM-DD 17:36:11,334] INFO KafkaConfig values:
   
   ....
   
   (kafka.server.KafkaConfig)
   
   [YYYY-MM-DD 17:36:11,441] INFO starting (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:36:11,450] INFO Connecting to zookeeper on yyy.yyy.93.87:3181,yyy.yyy.93.88:3181,yyy.yyy.97.221:3181 (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:36:11,870] INFO Loading logs. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:36:12,775] WARN Found an corrupted index file, /var/tmp/kafka_2.11-0.9.0.0/kafka-logs/NUMBERS-1/00000000000000000000.index, deleting and rebuilding index... (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:12,776] INFO Recovering unflushed segment 0 in log NUMBERS-1. (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:12,777] INFO Completed load of log NUMBERS-1 with log end offset 14 (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:12,780] WARN Found an corrupted index file, /var/tmp/kafka_2.11-0.9.0.0/kafka-logs/NUMBERS-0/00000000000000000000.index, deleting and rebuilding index... (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:12,781] INFO Recovering unflushed segment 0 in log NUMBERS-0. (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:12,782] INFO Completed load of log NUMBERS-0 with log end offset 13 (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:12,785] INFO Logs loading complete. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:36:12,786] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:36:12,799] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:36:12,858] INFO Awaiting socket connections on SBT-IPO-204.youre.adress.ru:9092. (kafka.network.Acceptor)
   
   [YYYY-MM-DD 17:36:12,861] INFO [Socket Server on Broker 2], Started 1 acceptor threads (kafka.network.SocketServer)
   
   [YYYY-MM-DD 17:36:13,054] INFO [GroupCoordinator 2]: Starting up. (kafka.coordinator.GroupCoordinator)
   
   [YYYY-MM-DD 17:36:13,064] INFO [ExpirationReaper-2], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:36:13,064] INFO [GroupCoordinator 2]: Startup complete. (kafka.coordinator.GroupCoordinator)
   
   [YYYY-MM-DD 17:36:13,073] INFO [Group Metadata Manager on Broker 2]: Removed 0 expired offsets in 17 milliseconds. (kafka.coordinator.GroupMetadataManager)
   
   [YYYY-MM-DD 17:36:13,075] INFO [ExpirationReaper-2], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:36:13,087] INFO [ThrottledRequestReaper-Produce], Starting (kafka.server.ClientQuotaManager\$ThrottledRequestReaper)
   
   [YYYY-MM-DD 17:36:13,088] INFO [ThrottledRequestReaper-Fetch], Starting (kafka.server.ClientQuotaManager\$ThrottledRequestReaper)
   
   [YYYY-MM-DD 17:36:13,094] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader\$)
   
   [YYYY-MM-DD 17:36:13,119] INFO Creating /brokers/ids/2 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:36:13,133] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:36:13,135] INFO Registered broker 2 at path /brokers/ids/2 with addresses: PLAINTEXT -\> EndPoint(yyy.yyy.93.88,9092,PLAINTEXT) (kafka.utils.ZkUtils)
   
   [YYYY-MM-DD 17:36:13,149] INFO [Kafka Server 2], started (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:36:15,082] INFO [ReplicaFetcherManager on broker 2] Removed fetcher for partitions [test2,1],[test,14],[test2,13],[test,15],[test,3],[test2,11],[test,6],[test5,1],[test2,5],[ThrashMetal,2],[test,17],[test5,0],[test,5],[test,11],[test3,1],[test2,15],[test2,18],[test2,8],[test,16],[test,1],[NUMBERS,0],[test,4],[test,19],[test2,3],[test2,10],[ThrashMetal,0],[test,18],[test,13],[test2,2],[test2,17],[test,12],[test2,7],[test,2],[test,8],[test2,16],[NUMBERS,1],[test2,6],[test,0],[test2,4],[test2,0],[test,7],[test2,9],[test,9],[superTest,1],[test3,0],[test2,14],[ThrashMetal,1],[superTest,0],[test2,19],[test,10],[test2,12] (kafka.server.ReplicaFetcherManager)
   
   [YYYY-MM-DD 17:36:15,096] INFO Truncating log NUMBERS-0 to offset 13. (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:15,100] INFO Truncating log NUMBERS-1 to offset 14. (kafka.log.Log)
   
   [YYYY-MM-DD 17:36:15,162] INFO [ReplicaFetcherManager on broker 2] Added fetcher for partitions List([[NUMBERS,1], ...) initOffset 14 to broker BrokerEndPoint(1,yyy.yyy.93.87,9092)]
   
   [YYYY-MM-DD 17:36:15,171] INFO [ReplicaFetcherThread-0-1], Starting (kafka.server.ReplicaFetcherThread)
   
   [YYYY-MM-DD 17:36:15,354] INFO [Group Metadata Manager on Broker 2]: Loading offsets and group metadata from [consumer_offsets,22] (kafka.coordinator.GroupMetadataManager)
   
   [YYYY-MM-DD 17:36:15,424] INFO [Group Metadata Manager on Broker 2]: Finished loading offsets from [consumer_offsets,22] in 69 milliseconds. (kafka.coordinator.GroupMetadataManager)
   
   [YYYY-MM-DD 17:37:14,092] INFO [ReplicaFetcherManager on broker 2] Removed fetcher for partitions [NUMBERS,1] (kafka.server.ReplicaFetcherManager)
   

   Запуск Corax-брокера на сервере Server-2 выполнен успешно.

3. Выполните запуск узла Corax на сервере Server-3 в режиме демона с выводом логов:

   bin/kafka-server-start.sh -daemon
   
   config/server.properties && tail -f logs/server.log
   
   [mon99usr\@SBT-IPO-208 kafka_2.11-0.9.0.0]\$ bin/kafka-server-start.sh -daemon config/server-3-servers.properties && tail -f logs/server.log
   
   Kafka application log tail -f from /home/mon99usr/kafka_2.11-0.9.0.0/logs/server.log
   
   [ [YYYY-MM-DD 17:45:56,259] INFO Creating /brokers/ids/3 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:45:56,280] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:45:56,281] INFO Registered broker 3 at path /brokers/ids/3 with addresses: PLAINTEXT -> EndPoint(yyy.yyy.97.221,9092,PLAINTEXT) (kafka.utils.ZkUtils)
   
   [YYYY-MM-DD 17:45:56,300] INFO [Kafka Server 3], started (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:50:45,138] INFO New leader is 2 (kafka.server.ZookeeperLeaderElector\$LeaderChangeListener)
   
   [YYYY-MM-DD 17:50:57,090] INFO Creating /controller (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:50:57,096] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:50:57,097] INFO 3 successfully elected as leader (kafka.server.ZookeeperLeaderElector)
   
   [YYYY-MM-DD 17:51:00,733] INFO New leader is 3 (kafka.server.ZookeeperLeaderElector\$LeaderChangeListener)
   
   [YYYY-MM-DD 17:52:37,509] INFO KafkaConfig values:
   
   ...
   
   (kafka.server.KafkaConfig)
   
   [YYYY-MM-DD 17:52:37,675] INFO starting (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:52:37,686] INFO Connecting to zookeeper on yyy.yyy.93.87:3181,yyy.yyy.93.88:3181,yyy.yyy.97.221:3181 (kafka.server.KafkaServer)
   
   [YYYY-MM-DD 17:52:38,083] INFO Loading logs.
   
   [YYYY-MM-DD 17:52:38,090] INFO Logs loading complete.
   
   [YYYY-MM-DD 17:52:38,091] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:52:38,103] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
   
   [YYYY-MM-DD 17:52:38,163] INFO Awaiting socket connections on SBT-IPO-208.youre.adress.ru:9092. (kafka.network.Acceptor)
   
   [YYYY-MM-DD 17:52:38,169] INFO [Socket Server on Broker 3], Started 1 acceptor threads (kafka.network.SocketServer)
   
   [YYYY-MM-DD 17:52:38,199] INFO [ExpirationReaper-3], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:52:38,203] INFO [ExpirationReaper-3], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:52:38,378] INFO [GroupCoordinator 3]: Starting up. (kafka.coordinator.GroupCoordinator)
   
   [YYYY-MM-DD 17:52:38,394] INFO [Group Metadata Manager on Broker 3]: Removed 0 expired offsets in 13 milliseconds. (kafka.coordinator.GroupMetadataManager)
   
   [YYYY-MM-DD 17:52:38,396] INFO [ExpirationReaper-3], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:52:38,397] INFO [GroupCoordinator 3]: Startup complete. (kafka.coordinator.GroupCoordinator)
   
   [YYYY-MM-DD 17:52:38,407] INFO [ExpirationReaper-3], Starting (kafka.server.DelayedOperationPurgatory\$ExpiredOperationReaper)
   
   [YYYY-MM-DD 17:52:38,432] INFO [ThrottledRequestReaper-Produce], Starting (kafka.server.ClientQuotaManager\$ThrottledRequestReaper)
   
   [YYYY-MM-DD 17:52:38,437] INFO [ThrottledRequestReaper-Fetch], Starting (kafka.server.ClientQuotaManager\$ThrottledRequestReaper)
   
   [YYYY-MM-DD 17:52:38,445] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader\$)
   
   [YYYY-MM-DD 17:52:38,485] INFO Creating /brokers/ids/3 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:52:38,504] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
   
   [YYYY-MM-DD 17:52:38,506] INFO Registered broker 3 at path /brokers/ids/3 with addresses: PLAINTEXT -\> EndPoint(yyy.yyy.97.221,9092,PLAINTEXT) (kafka.utils.ZkUtils)
   
   [YYYY-MM-DD 17:52:38,524] INFO [Kafka Server 3], started (kafka.server.KafkaServer)
   

   Запуск Corax-брокера на сервере Server-3 выполнен успешно.

Запуск кластеров Corax и Zookeeper выполнен успешно.

Тест доступности Zookeeper / работоспособности Corax при различных условиях работы кластера Zookeeper

1. Тест работоспособности Corax при всех работающих узлах кластера Zookeeper.

   • Вывод списка топиков:

     bin/kafka-topics.sh
     
     ---list
     
     --zookeeper yyy.yyy.93.87:3181,yyy.yyy.93.87:3181,yyy.yyy.93.87:3181
     
     ---topic NUMBERS
     
     NUMBERS
     

     В списке содержится топик NUMBERS.

   • Отправка сообщений в топик NUMBERS:

     bin/kafka-console-producer.sh
     
     --broker-list
     yyy.yyy.93.87:9092,yyy.yyy.93.88:9092,yyy.yyy.97.221:9092
     
     --topic NUMBERS
     
     1
     
     2
     
     3
     

     Запись в топик успешна.

   • Чтение сообщений из топика:

     bin/kafka-console-consumer.sh
     
     --new-consumer
     
     --bootstrap-server yyy.yyy.93.88:9092
     
     --topic NUMBERS
     
     --from-beginning
     
     1
     
     2
     
     3
     

     Чтение из топика успешно.

2. Тест работоспособности Corax при отказе одного узла кластера Zookeeper:

   • список топиков выводится;

   • чтение сообщений из топика выполняется успешно;

   • отправка сообщений в топик выполняется успешно.

3. Тест работоспособности Corax при отказе двух узлов кластера Zookeeper:

   • список топиков не выводится;

   • чтение сообщений из топика выполняется успешно;

   • отправка сообщений в топик выполняется успешно.

4. Тест работоспособности Corax при полном отказе кластера Zookeeper:

   • список топиков не выводится;

   • чтение сообщений из топика выполняется успешно;

   • отправка сообщений в топик выполняется успешно.

Конфигурирование Corax#

Конфигурирование узла (брокера) Corax#

Конфигурирование брокера Corax осуществляется в файле config/server.properties.

Допускается создание файлов конфигурации с различными именами, например:

    config/server-1-node-cluster.properties
    config/server-3-node-cluster.properties
    config/server-3-node-cluster-on-1-machine.properties
    config/server-5-node-cluster.properties

При запуске Corax достаточно указать нужный конфигурационный файл.

Минимальная конфигурация#

Пример файла конфигураций server.properties для одного брокера:

broker.id=1
listeners=PLAINTEXT://localhost:9092
zookeeper.connect=localhost:2181
log.dirs=/var/log/kafka_2.11-0.9.0.0/kafka-logs

Шаги создания конфигурационного файла:

1. Присвойте ключу broker.id необходимый числовой идентификатор. Необходимо соблюдать уникальность данного идентификатора на всех серверах кластера.

   Пример

   broker.id=1

2. Для ключа zookeeper.connect укажите IP:PORT сервиса Zookeeper.

   Пример

   Запущен один экземпляр Zookeeper и его порт clientPort=2181:

   zookeeper.connect=localhost:2181

   Пример

   Запущен кластер из нескольких экземпляров Zookeeper (адреса всех служб Zookeeper указаны через запятую):

   zookeeper.connect=xxx.xxx.1.1:2181,xxx.xxx.1.2:2181

3. Для ключа log.dirs задайте одну или несколько директорий, в которых планируется хранение сообщений, передаваемых в Corax.

   Пример

   log.dirs=/var/log/kafka_2.11-0.9.0.0/kafka-logs

   Примечание

   Если планируется использовать Corax под большой нагрузкой, то должно быть выделено достаточно места, т. к. все сообщения будут сохраняться на диск именно в указанной директории.

   Если на физическом сервере возможно использование нескольких физических жестких дисков для хранения сообщений Corax, то для увеличения производительности перечислите пути к директориям на этих дисках через запятую:

   log.dirs=/var/log/kafka_2.11-0.9.0.0/kafka-logs,/another_disk/kafka_2.11-0.9.0.0/kafka-logs

4. Для ключа listeners присвойте строку формата <ПРОТОКОЛ>://<хост>:<порт>:

   Пример

       listeners=PLAINTEXT://localhost:9092
       listeners=SSL://yyy.yyy.93.88:9092
       listeners=SASL_PLAINTEXT://:9092
   

Расширенная конфигурация#

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

• Если на одном физическом сервере планируется запускать несколько брокеров Corax, то для каждого брокера необходимо также указать уникальный порт, заданный в ключе port. По умолчанию сервер Corax слушает порт 9092.

• Можно изменить значение количества партиций для вновь создаваемых топиков по умолчанию в ключе num.partitions. Если для сервера Corax доступны несколько жестких дисков, увеличение числа партиций увеличит производительность Corax. Количество партиций по умолчанию можно указать равным числу дисков.

  Пример:

  num.partitions=4

• Для повышения надежности необходимо раскомментировать записи, содержащие ключи log.flush.interval.messages и log.flush.interval.ms. В этом случае указывается, как часто должна происходить реальная запись сообщений на диск: либо по истечении заданного времени, либо после достижения заданного количества незаписанных сообщений. Данную настройку можно изменить для конкретного топика при его создании.

  Пример:

  log.flush.interval.messages=10000
  log.flush.interval.ms=1000
  

• Для ключа log.retention.hours задается время хранения партиций в часах, по истечении которого партиция будет доступна для удаления. В данном случае отсчет будет вестись по самому новому сообщению в партиции.

  Пример

  log.retention.hours=168

• Для ключа log.cleaner.enable задается значение true для активации механизма удаления старых партиций.

  Пример

  log.cleaner.enable=true

Запуск узлов кластера 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.

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

Настройки подключения к 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-брокера, и их значений по умолчанию приведен в таблице ниже.

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

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

Конфигурирование кластера Corax#

Конфигурирование кластера серверов Corax, как и конфигурирование одного узла, осуществляется в файле server.properties на каждом сервере, который планируется включить в кластер.

Минимальная конфигурация кластера#

Пример файла конфигураций server.properties для кластера Corax из двух брокеров (приведен пример конфигурации для одного из брокеров):

broker.id=1

host.name=xxx.xxx.1.1

zookeeper.connect=xxx.xxx.1.1:3181,xxx.xxx.1.2:3181

log.dirs=/var/log/kafka_2.11-0.9.0.0/kafka-logs

Шаги создания конфигурационного файла:

1. Присвойте ключу broker.id необходимый числовой идентификатор.

   Необходимо соблюдать уникальность данного идентификатора на всех серверах кластера.

   Пример

   На первом сервере Corax:

   broker.id=1

   На втором сервере:

   broker.id=2

2. Присвойте ключу host.name IP-адрес сервера.

   Пример

   host.name=xxx.xxx.1.1

   Если на одном физическом сервере планируется запускать несколько серверов Corax, то для каждого сервера необходимо также указать уникальный порт, заданный в ключе port. По умолчанию сервер Corax слушает порт 9092.

3. Для ключа zookeeper.connect укажите IP и PORT сервиса Zookeeper в формате zookeeper.connect=IP:PORT. Если кластер Zookeeper запущен, то укажите адреса Zookeeper-сервисов через запятую:

   Пример

   zookeeper.connect=xxx.xxx.1.1:2181,xxx.xxx.1.2:2181

4. Для ключа log.dirs задайте директорию, в которой планируется хранение сообщений, передаваемых в Corax.

   Пример

   log.dirs=/var/log/kafka_2.11-0.9.0.0/kafka-logs

   Примечание

   Если планируется использовать Corax под большой нагрузкой, то должно быть выделено достаточно места, т. к. все сообщения будут сохраняться на диск именно в указанной директории.

   Если на физическом сервере возможно использование нескольких физических жестких дисков для хранения сообщений Corax, то для увеличения производительности перечислите пути к директориям на этих дисках через запятую:

   log.dirs=/var/log/kafka_2.11-0.9.0.0/kafka-logs,/another_disk/kafka_2.11-0.9.0.0/kafka-logs

На данном этапе минимальная конфигурация узлов кластера Corax завершается. Если не существует каких-либо ошибок в конфигурации, а также Zookeeper сконфигурирован корректно и запущен, то этого вполне достаточно для успешного запуска и работы кластера Corax.

Расширенная конфигурация кластера#

Конфигурирование производителя (producer)#

Минимальная конфигурация

Пример файла конфигураций producer.properties для одного брокера:

acks=1

bootstrap.servers=`hostname --f`:9101, `hostname --f`:9102,
`hostname --f`:9103

Шаги создания конфигурационного файла:

1. Присвойте ключу acks значение, указывающее, что сообщение отправлено:

   • 0 — сообщение расположено в TCP-буфере для отправки по сети;

   • 1 — сообщение успешно записано в партицию-лидер;

   • 1 (all) — сообщение успешно записано на все ISR (InSyncReplicas параметр на Kafka broker (server.properties))

   Пример

   acks=1

2. Для ключа bootstrap.servers укажите IP:PORT сервиса Corax.

   Пример

   Если запущены 3 экземпляра Corax на том же сервере:

   bootstrap.servers=`hostname --f`:9101, `hostname --f`:9102, `hostname --f`:9103
   

   Допускается также указывать IP-адреса, полные доменные имена.

Расширенная конфигурация

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

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

./bin/kafka-console-producer --broker-list `hostname --f`:9101, `hostname --f`:9102, `hostname --f`:9103 --topic test1 --producer.config etc/kafka/producer.properties

Конфигурирование потребителя (consumer)#

Минимальная конфигурация#

Пример файла конфигураций consumer.properties для одного брокера:

group.id=test-consumer-group

bootstrap.servers=`hostname --f`:9101, `hostname --f`:9102,
`hostname --f`:9103

Шаги создания конфигурационного файла:

1. Присвойте ключу group.id необходимый идентификатор группы, под которым будет запущен consumer. Можно запускать несколько consumers под одним и тем же group.id, что позволит считывать данные в несколько потоков (с нескольких серверов), обеспечивая отсутствие дублирования вычитываемых данных.

   Пример

   group.id=test-consumer-group

2. Для ключа bootstrap.servers укажите IP:PORT сервиса Corax.

   Пример

   Если запущены 3 экземпляра Corax на том же сервере:

   bootstrap.servers=`hostname --f`:9101, `hostname --f`:9102, `hostname --f`:9103
   

   Допускается также указывать IP-адреса, полные доменные имена.

Расширенная конфигурация#

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

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

./bin/kafka-console-consumer --broker-list `hostname --f`:9101, `hostname --f`:9102, `hostname --f`:9103 --topic test1 --consumer.config etc/kafka/consumer.properties --from-beginning

Конфигурирование сквозного шифрования#

Симметричное шифрование#

На стороне producer#

Producer конфигурируется следующими параметрами:

На стороне consumer#

Consumer конфигурируется следующими параметрами:

Внимание!

Для успешной расшифровки данных, полученных consumer, необходимо, чтобы соль и секретный ключ consumer совпадали с солью и секретным ключом producer.

Примеры конфигурации#

Пример producer.config

value.serializer=youre.adress.kafka.serialization.EncryptedByteArraySerializer
key.serializer=youre.adress.kafka.serialization.EncryptedByteArraySerializer
e2e.serializer.encryption.secret.key=secret
e2e.serializer.encryption.salt.provider.class=youre.adress.kafka.security.FileSaltProvider
e2e.serializer.encryption.class=youre.adress.kafka.security.SimpleDataEncoder
e2e.serializer.encryption.security.encoding.salt.location=config/salt.txt

Пример consumer.config

value.deserializer=youre.adress.kafka.serialization.EncryptedByteArrayDeserializer
key.deserializer=youre.adress.kafka.serialization.EncryptedByteArrayDeserializer
e2e.deserializer.encryption.secret.key=secret
e2e.deserializer.encryption.salt.provider.class=youre.adress.kafka.security.FileSaltProvider
e2e.deserializer.encryption.class=youre.adress.kafka.security.SimpleDataDecoder
e2e.deserializer.encryption.security.encoding.salt.location=config/salt.txt

Асимметричное шифрование#

Конфигурирование producer и consumer, работающих с шифрованием, возможно только путем редактирования кода. Properties - класс в java, с помощью которого происходит конфигурирование.

Для producer и consumer используется алгоритм шифрования/расшифровки RSA/ECB/PKCS1Padding.

На стороне producer#

Producer конфигурируется следующими параметрами:

Внимание!

Пароли в Properties должны быть представлены экземплярами класса org.apache.kafka.common.config.types.Password. Это маскирует пароль в логах. Другие типы не поддерживаются.

На стороне consumer#

Consumer конфигурируется следующими параметрами:

Внимание!

Пароли в Properties должны быть представлены экземплярами класса org.apache.kafka.common.config.types.Password. Это маскирует пароль в логах. Другие типы не поддерживаются.

Обновление SSL сертификатов на кластере Corax без недоступности кластера Corax#

Алгоритм обновления SSL сертификатов на брокерах кластера Corax:

1. Остановите брокер, который еще не обновлялся.

2. Для остановленного брокера обновите SSL сертификат.

3. Запустите брокер, и дождитесь завершения полной репликации (пример команды: kafka-topics --zookeeper host:2181  --describe --under-replicated-partitions).

4. Повторите шаг 1 для следующего брокера. Если это был последний брокер — обновление завершено.

Конфигурирование аудита событий над брокером в log-файл#

1. Для включения аудита над брокером Corax в log-файл, добавьте в конфигурационный файл KAFKA_DIR/config/server.properties строки:

   kafka.se.audit.enable=true
   kafka.se.audit.provider=kafka.audit.provider.LogAuditProvider
   

   где kafka.se.audit.enable — глобальное включение аудита;

   Примечание

   По умолчанию будет включен аудит записи действий на брокере в log-файл kafka-audit.log на уровне INFO.

2. При необходимости указать другой log-файл для аудита и/или задать свои параметры (уровень логирования, вид записи, форматирование или режим записи rolling update), укажите appender в файле config/log4j.properties:

   log4j.appender.auditFileAppender = org.apache.log4j.RollingFileAppender //включение режима записи rolling update
   log4j.appender.auditFileAppender.File = ${kafka.logs.dir}/kafka-audit.log //путь до log-файла
   log4j.appender.auditFileAppender.layout.ConversionPattern = [%d] %p %m (%c)%n //формат даты и времени
   log4j.logger.kafka.audit.provider.LogAuditProvider = INFO, auditFileAppender //уровень логирования
   

Конфигурирование Schema Registry#

Schema Registry REST Server#

REST API SSL#

Настраивается стандартным для Spring-boot способом. Инструкция приведена в официальной документации.

Для шифрования/расшифровки паролей к хранилищам сертификатов необходимо использовать следующие настройки:

Если не указана настройка schema.registry.rest.security.encoding.class, значения обрабатываться не будут.

Значение настройки аналогично настройкам Kafka:

schema.registry.rest.security.encoding.class=youre.adress.kafka.security.SimpleTextPasswordDecoder
schema.registry.rest.security.encoding.salt=youre.adress.kafka.security.FileSaltProvider
schema.registry.rest.security.encoding.key=KEY_1
schema.registry.rest.security.encoding.salt.location=classpath:/salt.txt
simple.encrypted.value=${decrypt:uKLIwVWQXoYgOBcqtN4DXA==} #На старте значение будет дешифровано.

SchemaRegistryClient#

SchemaRegistryClient - Java API для выполнения всех запросов к Schema Registry REST Server.

SchemaRegistryClient кеширует данные, полученные из REST Server. Поэтому, если схемы были пересозданы (удалены и повторно добавлены) или добавлена новая версия схемы, то приложение, которое использует SchemaRegistryClient, нужно перезапустить.

SchemaRegistrySerde#

SchemaRegistrySerde - компонент Schema Registry, сериализатор (десериализатор), который:

• сериализует данные и записывает в топик;

• читает данные из топика и десериализует.

Для расшифровки пароля клиента следующие настройки:

SubjectNameStrategy — реализация данного интерфейса генерирует 2 вида идентификаторов.

Если стратегия имен по умолчанию не подходит, реализуйте этот интерфейс самостоятельно.

1. subject — имя субъекта по имени топика и признаку isKey. По умолчанию для value имя субъекта совпадает с именем топика, в конце добавляется -value. Пример: accounts-value.

   Для ключей добавляется -key. Пример: accounts-key.

2. version — ключ, в котором будет лежать номер версии схемы. По умолчанию [subject].version. Пример:

   • accounts.version=2 — использовать вторую версию схемы accounts;

   • accounts-key.version=3 — использовать третью версию схемы для ключей топика accounts.

Настройки брокера#

Для безопасной конфигурации records policy, настройки могут быть указаны в параметрах брокера:

[name] — название преднастроенной policy. Его надо указывать при создании топика, который будет проверяться с этой policy.

• records.policy.[name].class.name — имя класса или название policy, реализующего проверку сообщений на уровне брокера.

• records.policy.[name].param1, records.policy.[name].param1 — параметры конфигурации policy. Доступны с метода RecordsPolicy#configure.

Пример:

records.policy.secure-policy.class.name=youre.adress.kafka.schemaregistry.SchemaRegistryValuesRecordsPolicy
# Параметры подключения к schema registry REST server.
records.policy.secure-policy.url=<Schema-registry-REST-server-address>
records.policy.secure-policy.user=user
# Пароль зашифрован.
records.policy.secure-policy.password.encrypted=uKLIwVWQXoYgOBcqtN4DXA==
# Параметры для расшифровки пароля.
# На стороне пользователя kafka знаний о параметрах подключения к REST server не требуется.
# Достаточно при создании топика указать `records.policy.name=secure-policy`
records.policy.secure-policy.security.encoding.class=youre.adress.kafka.security.SimpleTextPasswordDecoder
records.policy.secure-policy.security.encoding.salt=youre.adress.kafka.security.FileSaltProvider
records.policy.secure-policy.security.encoding.key=KEY_1
records.policy.secure-policy.security.encoding.salt.location=classpath:/salt.txt

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

Настройки конфигурации Schema Registry при создании или изменении топика:

где

• <sr-host> – хост, на котором запущен сервер Schema Registry;

• <sr-port> – порт, заданный в конфигурационном файле rest.properties (по умолчанию 8080).

Особенности обновления consumers и producers#

• В любом НЕ *_TRANSITIVE режиме одновременно могут работать приложения, которые используют версии схемы, различающиеся не более чем на 1.

• Обновление в режиме BACKWARD или BACKWARD_TRANSISTIVE: сначала обновите consumers, затем producers.

• Обновление в режиме FORWARD или FORWARD_TRANSISTIVE: сначала обновите producers, затем consumers.

• В режимах FULL, FULL_TRANSITIVE, NONE порядок обновления не важен.