Uso do MirrorMaker para sincronizar dados entre clusters

Cenário

Nos cenários a seguir, o MirrorMaker pode ser usado para sincronizar dados entre diferentes clusters do Kafka para garantir a disponibilidade e a confiabilidade dos clusters:

• Backup e recuperação de desastres: uma empresa tem vários data centers. Para evitar a indisponibilidade do serviço causada por uma falha em um data center, os dados do cluster são copiados de forma síncrona em vários data centers.
• Migração de cluster: à medida que as empresas migram serviços para a nuvem, os dados em clusters locais devem ser sincronizados com os dados em clusters de nuvem para garantir a continuidade do serviço.

Arquitetura da solução

O MirrorMaker pode ser usado para espelhar dados do cluster de origem para o cluster de destino. Como mostrado em Figura 1, em essência, o MirrorMaker primeiro consome dados do cluster de origem e, em seguida, produz os dados consumidos para o cluster de destino. Para obter mais informações sobre o MirrorMaker, consulte Espelhamento de dados entre clusters.

Figura 1 Como funciona o MirrorMaker

Restrições

• Os endereços IP e os números de porta dos nós no cluster de origem não podem ser iguais aos dos nós no cluster de destino. Caso contrário, os dados serão replicados infinitamente em um tópico.
• Use o MirrorMaker para sincronizar dados entre pelo menos dois clusters. Se houver apenas um cluster, os dados serão replicados infinitamente em um tópico.

Procedimento

1. Compre um ECS que possa se comunicar com os clusters de origem e de destino. Para obter detalhes, consulte a Documentação do ECS.
2. Faça logon no ECS, instale o JDK e adicione o seguinte conteúdo ao .bash_profile no diretório inicial para configurar as variáveis de ambiente JAVA_HOME e PATH. Neste comando, /opt/java/jdk1.8.0_151 é o caminho de instalação do JDK. Altere-o para o caminho onde você instala o JDK.

   export JAVA_HOME=/opt/java/jdk1.8.0_151 
   export PATH=$JAVA_HOME/bin:$PATH

   Execute o comando source .bash_profile para que a modificação tenha efeito.

   

   Use o JDK Oracle em vez do JDK padrão do ECS (por exemplo, OpenJDK), porque o JDK padrão do ECS pode não ser adequado. Obtenha o JDK Oracle 1.8.111 ou posterior no site oficial de Oracle.

3. Faça o download do pacote de software binário do Kafka 3.3.1.

   wget https://archive.apache.org/dist/kafka/3.3.1/kafka_2.12-3.3.1.tgz

4. Descompacte o pacote de software binário.

   tar -zxvf kafka_2.12-3.3.1.tgz

5. Vá para o diretório do pacote de software binário e especifique os endereços IP e portas dos clusters de origem e de destino e outros parâmetros no arquivo de configuração connect-mirror-maker.properties no diretório config.

   # Specify two clusters.
   clusters = A, B
   A.bootstrap.servers = A_host1:A_port, A_host2:A_port, A_host3:A_port
   B.bootstrap.servers = B_host1:B_port, B_host2:B_port, B_host3:B_port
   
   # Specify the data synchronization direction. The data can be synchronized unidirectionally or bidirectionally.
   A->B.enabled = true
   
   # Specify the topics to be synchronized. Regular expressions are supported. By default, all topics are replicated, for example, foo-.*.
   A->B.topics = .*
   
   # If the following two configurations are enabled, clusters A and B replicate data with each other.
   #B->A.enabled = true
   #B->A.topics = .*
   
   # Specify the number of replicas. If multiple topics need to be synchronized and their replica quantities are different, create topics with the same name and replica quantity before starting MirrorMaker.
   replication.factor=3
   
   # Specify the consumer offset synchronization direction (unidirectionally or bidirectionally).
   A->B.sync.group.offsets.enabled=true
   
   ############################# Internal Topic Settings  #############################
   # The replication factor for mm2 internal topics "heartbeats", "B.checkpoints.internal" and
   # "mm2-offset-syncs.B.internal"
   # In the test environment, the value can be 1. In the production environment, it is recommended that the value be greater than 1, for example, 3.
   checkpoints.topic.replication.factor=3
   heartbeats.topic.replication.factor=3
   offset-syncs.topic.replication.factor=3
   
   # The replication factor for connect internal topics "mm2-configs.B.internal", "mm2-offsets.B.internal" and
   # "mm2-status.B.internal"
   # In the test environment, the value can be 1. In the production environment, it is recommended that the value be greater than 1, for example, 3.
   offset.storage.replication.factor=3
   status.storage.replication.factor=3
   config.storage.replication.factor=3
   
   # customize as needed
   # replication.policy.separator = _
   # sync.topic.acls.enabled = false
   # emit.heartbeats.interval.seconds = 5

6. No diretório do pacote de software binário, inicie o MirrorMaker para sincronizar dados.

   ./bin/connect-mirror-maker.sh config/connect-mirror-maker.properties

7. (Opcional) Se um tópico for criado no cluster de origem após o MirrorMaker ter sido iniciado e os dados do tópico precisarem ser sincronizados, reinicie o MirrorMaker. Para obter detalhes sobre como reiniciar o MirrorMaker, consulte 6. Você também pode adicionar configurações listadas em Tabela 1 para sincronizar novos tópicos periodicamente sem reiniciar o MirrorMaker. refresh.topics.interval.seconds é obrigatório. Outros parâmetros são opcionais.

   Tabela 1 Configurações do MirrorMaker

   Parâmetro	Valor padrão	Descrição
   sync.topic.configs.enabled	true	Se deve monitorar o cluster de origem para alterações de configuração.
   sync.topic.acls.enabled	true	Se deve monitorar o cluster de origem para alterações de ACL.
   emit.heartbeats.enabled	true	Se deve permitir que o conector envie pulsações periodicamente.
   emit.heartbeats.interval.seconds	5 segundos	Frequência de pulsações.
   emit.checkpoints.enabled	true	Se deixar o conector enviar periodicamente as informações de deslocamento do consumidor.
   emit.checkpoints.interval.seconds	5 segundos	Frequência do ponto de verificação.
   refresh.topics.enabled	true	Se deixar o conector verificar periodicamente novos tópicos.
   refresh.topics.interval.seconds	5 segundos	Frequência de verificação de novos tópicos no cluster de origem.
   refresh.groups.enabled	true	Se deixar o conector verificar periodicamente se há novos grupos de consumidores.
   refresh.groups.interval.seconds	5 segundos	Frequência de verificação de novos grupos de consumidores no cluster de origem.
   replication.policy.class	org.apache.kafka.connect.mirror.DefaultReplicationPolicy	Use o LegacyReplicationPolicy para imitar o MirrorMaker de uma versão anterior.
   heartbeats.topic.retention.ms	Um dia	Usado quando tópicos de pulsação são criados pela primeira vez.
   checkpoints.topic.retention.ms	Um dia	Usado quando os tópicos do ponto de verificação são criados pela primeira vez.
   offset.syncs.topic.retention.ms	max long	Usado quando tópicos de sincronização de deslocamento são criados pela primeira vez.

Verificação da sincronização de dados

1. Exiba a lista de tópicos no cluster de destino para verificar se há tópicos de origem.
   

   Os nomes de tópico no cluster de destino têm um prefixo (por exemplo, A.) adicionado ao nome do tópico de origem. Esta é uma configuração do MirrorMaker 2 para evitar o backup cíclico de tópicos.

2. Produza e consuma mensagens no cluster de origem, visualize o progresso do consumo no cluster de destino e verifique se os dados foram sincronizados do cluster de origem para o cluster de destino.

   Se o cluster de destino for uma instância do Kafka da Huawei Cloud, visualize o progresso do consumo na página Consumer Groups.