Миграция базы данных из стороннего кластера Apache Kafka® в Yandex Managed Service for Apache Kafka®

Перенести топики из кластера-источника Apache Kafka® в кластер-приемник Managed Service for Apache Kafka® можно двумя способами:

• С помощью встроенного в Yandex Managed Service for Apache Kafka® MirrorMaker-коннектора.

  Этот способ прост в настройке и не требует создания промежуточной виртуальной машины.

• С помощью утилиты MirrorMaker 2.0.

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

Перенос данных с использованием сервиса Yandex Managed Service for Apache Kafka® Connector

1. Создайте коннектор.
2. Проверьте наличие данных в топике кластера-приемника.

Создайте кластер и коннектор

Проверьте наличие данных в топике кластера-приемника

1. Подключитесь к топику кластера-приемника с помощью утилиты kafkacat. К имени топика кластера-источника добавьте префикс source: например, топик mytopic будет перенесен на кластер-приемник как source.mytopic.
2. Убедитесь, что в консоли отображаются сообщения из топика кластера-источника.

Перенос данных с использованием утилиты MirrorMaker

1. Настройте конфигурацию MirrorMaker.
2. Запустите репликацию.
3. Проверьте наличие данных в топике кластера-приемника.

Если созданные ресурсы вам больше не нужны, удалите их.

Перед началом работы

Подготовьте инфраструктуру

Выполните дополнительные настройки

1. Создайте в кластере-источнике пользователя admin-source с правом управления топиками через Admin API.

2. Подключитесь к виртуальной машине по SSH.

   1. Установите JDK:

      sudo apt update && sudo apt install --yes default-jdk
      

   2. Скачайте и распакуйте архив Apache Kafka® той же версии, которая установлена на кластере-приемнике. Например, для версии 2.8:

      wget https://archive.apache.org/dist/kafka/2.8.0/kafka_2.12-2.8.0.tgz && \
      tar -xvf kafka_2.12-2.8.0.tgz
      

   3. Установите утилиту kafkacat:

      sudo apt update && sudo apt install --yes kafkacat
      

      Убедитесь, что можете с ее помощью подключиться к кластеру-источнику и кластеру-приемнику через SSL.

3. Настройте межсетевой экран (firewall) и группы безопасности, если это требуется для подключения MirrorMaker к кластеру-приемнику и кластеру-источнику.

Настройте конфигурацию MirrorMaker

1. Подключитесь к ВМ с MirrorMaker по SSH.

2. Скачайте SSL-сертификат для подключения к кластеру Managed Service for Apache Kafka®.

3. В домашней директории создайте каталог mirror-maker для хранения сертификатов Java Keystore и конфигураций MirrorMaker:

   mkdir --parents /home/<домашняя директория>/mirror-maker
   

4. Выберите пароль для хранилища сертификатов, создайте хранилище и добавьте в него SSL-сертификат для подключения к кластеру:

   sudo keytool --noprompt -importcert -alias YandexCA \
      -file /usr/local/share/ca-certificates/Yandex/YandexCA.crt \
      -keystore /home/<домашняя директория>/mirror-maker/keystore \
      -storepass <пароль хранилища сертификатов, не короче 6 символов>
   

5. Создайте в каталоге mirror-maker файл конфигурации MirrorMaker mm2.properties:

   # Kafka clusters
   clusters=cloud, source
   source.bootstrap.servers=<FQDN брокера кластера-источника>:9092
   cloud.bootstrap.servers=<FQDN брокера 1 кластера-приемника>:9091, ..., <FQDN брокера N кластера-приемника>:9091
   
   # Source and target cluster settings
   source->cloud.enabled=true
   cloud->source.enabled=false
   source.cluster.alias=source
   cloud.cluster.alias=cloud
   
   # Internal topics settings
   source.config.storage.replication.factor=<R>
   source.status.storage.replication.factor=<R>
   source.offset.storage.replication.factor=<R>
   source.offsets.topic.replication.factor=<R>
   source.errors.deadletterqueue.topic.replication.factor=<R>
   source.offset-syncs.topic.replication.factor=<R>
   source.heartbeats.topic.replication.factor=<R>
   source.checkpoints.topic.replication.factor=<R>
   source.transaction.state.log.replication.factor=<R>
   cloud.config.storage.replication.factor=<R>
   cloud.status.storage.replication.factor=<R>
   cloud.offset.storage.replication.factor=<R>
   cloud.offsets.topic.replication.factor=<R>
   cloud.errors.deadletterqueue.topic.replication.factor=<R>
   cloud.offset-syncs.topic.replication.factor=<R>
   cloud.heartbeats.topic.replication.factor=<R>
   cloud.checkpoints.topic.replication.factor=<R>
   cloud.transaction.state.log.replication.factor=<R>
   
   # Topics
   topics=.*
   groups=.*
   topics.blacklist=.*[\-\.]internal, .*\replica, __consumer_offsets
   groups.blacklist=console-consumer-.*, connect-.*, __.*
   replication.factor=<M>
   refresh.topics.enable=true
   sync.topic.configs.enabled=true
   refresh.topics.interval.seconds=10
   
   # Tasks
   tasks.max=<T>
   
   # Source cluster authentication parameters. Comment out if no authentication required
   source.client.id=mm2_consumer_test
   source.group.id=mm2_consumer_group
   source.security.protocol=SASL_PLAINTEXT
   source.sasl.mechanism=SCRAM-SHA-512
   source.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin-source" password="<пароль>";
   
   # Target cluster authentication parameters
   cloud.client.id=mm2_producer_test
   cloud.group.id=mm2_producer_group
   cloud.ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
   cloud.ssl.truststore.location=/home/<домашняя директория>/mirror-maker/keystore
   cloud.ssl.truststore.password=<пароль хранилища сертификатов>
   cloud.ssl.protocol=TLS
   cloud.security.protocol=SASL_SSL
   cloud.sasl.mechanism=SCRAM-SHA-512
   cloud.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin-cloud" password="<пароль>";
   
   # Enable heartbeats and checkpoints
   source->target.emit.heartbeats.enabled=true
   source->target.emit.checkpoints.enabled=true
   

   Пояснения к конфигурации MirrorMaker:

   • Выполняется односторонняя репликация (параметры source->cloud.enabled = true, cloud->source.enabled = false).
   • В параметре topics указывается список топиков, которые нужно перенести. Также можно указать регулярное выражение для выбора топиков. Для переноса всех топиков укажите .*. В данной конфигурации реплицируются все топики.
   • Названия топиков в кластере-приемнике совпадают с их названиями в кластере-источнике.
   • Параметр <R> — фактор репликации служебных топиков MirrorMaker. Значение этого параметра не должно превышать меньшего между количеством брокеров в кластере-источнике и количеством брокеров в кластере-приемнике.
   • Параметр <M> — фактор репликации по умолчанию, установленный для топиков в кластере-приемнике.
   • Параметр <T> — количество одновременно работающих процессов MirrorMaker. Рекомендуется указывать не менее 2 для равномерного распределения нагрузки репликации. Подробнее см. в документации Apache Kafka®.

   FQDN брокеров Managed Service for Apache Kafka® можно запросить со списком хостов в кластере.

Запустите репликацию

Запустите MirrorMaker на ВМ командой:

<путь установки Kafka>/bin/connect-mirror-maker.sh /home/<домашняя директория>/mirror-maker/mm2.properties

Проверьте наличие данных в топике кластера-приемника

1. Подключитесь к топику кластера-приемника с помощью утилиты kafkacat. К имени топика кластера-источника добавьте префикс source: например, топик mytopic будет перенесен на кластер-приемник как source.mytopic.
2. Убедитесь, что в консоли отображаются сообщения из топика кластера-источника.

Подробнее о работе с сервисом MirrorMaker 2.0 см. в документации Apache Kafka®.

Удалите созданные ресурсы