Conectar o Pub/Sub ao Apache Kafka

Este documento descreve como integrar o Apache Kafka e o Pub/Sub usando o conector do Kafka do grupo do Pub/Sub.

Observação: também é possível usar esse conector com o Pub/Sub Lite. Consulte Conectar o Pub/Sub Lite ao Apache Kafka.

Sobre o conector de Kafka do grupo do Pub/Sub

O Apache Kafka é uma plataforma de código aberto para eventos de streaming. Ele é usado com frequência em arquiteturas distribuídas para permitir a comunicação entre componentes acoplados com flexibilidade. O Pub/Sub é um serviço gerenciado para enviar e receber mensagens de forma assíncrona. Assim como no Kafka, você pode usar o Pub/Sub para se comunicar entre componentes na sua arquitetura de nuvem.

O conector do Kafka do grupo do Pub/Sub permite integrar esses dois sistemas. Os seguintes conectores estão empacotados no JAR do conector:

• O conector do coletor lê registros de um ou mais tópicos do Kafka e os publica no Pub/Sub.
• O conector de origem lê mensagens de um tópico do Pub/Sub e as publica no Kafka.

Confira alguns cenários em que você pode usar o conector de Kafka do grupo do Pub/Sub:

• Você está migrando uma arquitetura baseada em Kafka para o Google Cloud.
• Você tem um sistema de front-end que armazena eventos no Kafka fora do Google Cloud, mas também usa o Google Cloud para executar alguns dos seus serviços de back-end, que precisam receber os eventos do Kafka.
• Você coleta logs de uma solução Kafka local e os envia ao Google Cloud para análise de dados.
• Você tem um sistema de front-end que usa o Google Cloud, mas também armazena dados localmente usando o Kafka.

O conector exige o Kafka Connect, que é uma estrutura para streaming de dados entre o Kafka e outros sistemas. Para usar o conector, execute o Kafka Connect com seu cluster do Kafka.

Este documento pressupõe que você tenha familiaridade com o Kafka e o Pub/Sub. Antes de ler este documento, é recomendável concluir um dos guias de início rápido do Pub/Sub.

O conector do Pub/Sub não oferece suporte a nenhuma integração entre o IAM do Google Cloud e as ACLs do Kafka Connect.

Começar a usar o conector

Esta seção orienta você nas seguintes tarefas:

1. Configure o conector de Kafka do grupo do Pub/Sub.
2. Envie eventos do Kafka para o Pub/Sub.
3. Enviar mensagens do Pub/Sub para o Kafka.

Observação: esta seção usa o Kafka Connect no modo independente, que é adequado para testes. Para sistemas de produção, o modo distribuído é recomendado. Para informações sobre como executar o Kafka Connect no modo distribuído, consulte Como executar o Kafka Connect na documentação do Apache Kafka.

Pré-requisitos

Instalar o Kafka

Siga o Guia de início rápido do Apache Kafka para instalar um nó único do Kafka na sua máquina local. Siga estas etapas no início rápido:

1. Faça o download da versão mais recente do Kafka e extraia o arquivo.
2. Inicie o ambiente do Kafka.
3. Crie um tópico do Kafka.

Autenticar

O conector do Kafka do grupo do Pub/Sub precisa ser autenticado com o Pub/Sub para enviar e receber mensagens do Pub/Sub. Para configurar a autenticação, siga estas etapas:

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
2. Install the Google Cloud CLI.

3. To initialize the gcloud CLI, run the following command:

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Create local authentication credentials for your user account:

   gcloud auth application-default login

6. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/pubsub.admin

   gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

   • Replace PROJECT_ID with your project ID.

   • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

   • Replace ROLE with each individual role.
7. Install the Google Cloud CLI.

8. To initialize the gcloud CLI, run the following command:

   gcloud init

9. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

10. Create local authentication credentials for your user account:

    gcloud auth application-default login

11. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/pubsub.admin

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

    • Replace PROJECT_ID with your project ID.

    • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

    • Replace ROLE with each individual role.

Observação: para cargas de trabalho de produção, recomendamos usar a federação de identidade da carga de trabalho para autenticação.

Fazer o download do JAR do conector

Faça o download do arquivo JAR do conector para sua máquina local. Para mais informações, consulte Adquirir o conector no readme do GitHub.

Copiar os arquivos de configuração do conector

1. Clone ou faça o download do repositório do GitHub para o conector.

   git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git
   cd java-pubsub-group-kafka-connector
   

2. Copie o conteúdo do diretório config para o subdiretório config da sua instalação do Kafka.

   cp config/* [path to Kafka installation]/config/
   

Esses arquivos contêm configurações para o conector.

Atualizar a configuração do Kafka Connect

1. Navegue até o diretório que contém o binário do Kafka Connect que você fez o download.
2. No diretório binário do Kafka Connect, abra o arquivo chamado config/connect-standalone.properties em um editor de texto.
3. Se o plugin.path property estiver comentado, remova o comentário.

4. Atualize o plugin.path property para incluir o caminho para o JAR do conector.

   Exemplo:

   plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar
   

5. Defina a propriedade offset.storage.file.filename como um nome de arquivo local. No modo independente, o Kafka usa esse arquivo para armazenar dados de deslocamento.

   Exemplo:

   offset.storage.file.filename=/tmp/connect.offsets
   

Encaminhar eventos do Kafka para o Pub/Sub

Esta seção descreve como iniciar o conector de coletor, publicar eventos no Kafka e ler as mensagens encaminhadas do Pub/Sub.

1. Use a Google Cloud CLI para criar um tópico do Pub/Sub com uma assinatura.

   gcloud pubsub topics create PUBSUB_TOPIC
   gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

   Substitua:

   • PUBSUB_TOPIC: o nome de um tópico do Pub/Sub para receber as mensagens do Kafka.
   • PUBSUB_SUBSCRIPTION: o nome de uma assinatura do Pub/Sub para o tópico.

2. Abra o arquivo /config/cps-sink-connector.properties em um editor de texto. Adicione valores para as seguintes propriedades, marcadas como "TODO" nos comentários:

   topics=KAFKA_TOPICS
   cps.project=PROJECT_ID
   cps.topic=PUBSUB_TOPIC

   Substitua:

   • KAFKA_TOPICS: uma lista separada por vírgulas de tópicos do Kafka para leitura.
   • PROJECT_ID: o projeto do Google Cloud que contém seu tópico do Pub/Sub.
   • PUBSUB_TOPIC: o tópico do Pub/Sub para receber as mensagens do Kafka.

3. No diretório do Kafka, execute o seguinte comando:

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/cps-sink-connector.properties
   

4. Siga as etapas no Guia de início rápido do Apache Kafka para gravar alguns eventos no seu tópico do Kafka.

5. Use a CLI gcloud para ler os eventos do Pub/Sub.

   gcloud pubsub subscriptions pull PUBSUB_SUBSCRIPTION --auto-ack

Encaminhar mensagens do Pub/Sub para o Kafka

Esta seção descreve como iniciar o conector de origem, publicar mensagens no Pub/Sub e ler as mensagens encaminhadas do Kafka.

1. Use a CLI gcloud para criar um tópico do Pub/Sub com uma assinatura.

   gcloud pubsub topics create PUBSUB_TOPIC
   gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

   Substitua:

   • PUBSUB_TOPIC: o nome de um tópico do Pub/Sub.
   • PUBSUB_SUBSCRIPTION: o nome de uma assinatura do Pub/Sub.

2. Abra o arquivo chamado /config/cps-source-connector.properties em um editor de texto. Adicione valores para as seguintes propriedades, marcadas como "TODO" nos comentários:

   kafka.topic=KAFKA_TOPIC
   cps.project=PROJECT_ID
   cps.subscription=PUBSUB_SUBSCRIPTION

   Substitua:

   • KAFKA_TOPIC: os tópicos do Kafka para receber as mensagens do Pub/Sub.
   • PROJECT_ID: o projeto do Google Cloud que contém seu tópico do Pub/Sub.
   • PUBSUB_TOPIC: o tópico do Pub/Sub.

3. No diretório do Kafka, execute o seguinte comando:

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/cps-source-connector.properties
   

4. Use a CLI gcloud para publicar uma mensagem no Pub/Sub.

   gcloud pubsub topics publish PUBSUB_TOPIC --message="message 1"

5. Leia a mensagem do Kafka. Siga as etapas no Guia de início rápido do Apache Kafka para ler as mensagens do tópico do Kafka.

Conversão de mensagem

Um registro do Kafka contém uma chave e um valor, que são matrizes de bytes de comprimento variável. Opcionalmente, um registro do Kafka também pode ter cabeçalhos, que são pares de chave-valor. Uma mensagem do Pub/Sub tem duas partes principais: o corpo da mensagem e zero ou mais atributos de chave-valor.

O Kafka Connect usa convertidores para serializar chaves e valores de e para o Kafka. Para controlar a serialização, defina as seguintes propriedades nos arquivos de configuração do conector:

• key.converter: o conversor usado para serializar chaves de registro.
• value.converter: o conversor usado para serializar valores de registro.

O corpo de uma mensagem do Pub/Sub é um objeto ByteString. Portanto, a conversão mais eficiente é copiar o payload diretamente. Por esse motivo, recomendamos usar um conversor que produza tipos de dados primitivos (inteiro, flutuante, string ou esquema de bytes) sempre que possível para evitar a desserialização e a reserialização do mesmo corpo da mensagem.

Conversão do Kafka para o Pub/Sub

O conector do coletor converte os registros do Kafka em mensagens do Pub/Sub da seguinte maneira:

• A chave de registro do Kafka é armazenada como um atributo chamado "key" na mensagem do Pub/Sub.
• Por padrão, o conector descarta todos os cabeçalhos no registro do Kafka. No entanto, se você definir a opção de configuração headers.publish como true, o conector vai gravar os cabeçalhos como atributos do Pub/Sub. O conector pula todos os cabeçalhos que excedem os limites de atributos de mensagem do Pub/Sub.
• Para esquemas de inteiros, flutuantes, strings e bytes, o conector transmite os bytes do valor do registro do Kafka diretamente para o corpo da mensagem do Pub/Sub.
• Para esquemas de estrutura, o conector grava cada campo como um atributo da mensagem do Pub/Sub. Por exemplo, se o campo for { "id"=123 }, a mensagem resultante do Pub/Sub terá um atributo "id"="123". O valor do campo é sempre convertido em uma string. Os tipos de mapa e struct não são compatíveis como tipos de campo em um struct.
• Para esquemas de mapa, o conector grava cada par de chave-valor como um atributo da mensagem do Pub/Sub. Por exemplo, se o mapa for {"alice"=1,"bob"=2}, a mensagem resultante do Pub/Sub terá dois atributos, "alice"="1" e "bob"="2". As chaves e os valores são convertidos em strings.

Os esquemas de estrutura e mapa têm alguns comportamentos adicionais:

• Opcionalmente, você pode especificar um campo de struct ou uma chave de mapa específicos para ser o corpo da mensagem definindo a propriedade de configuração messageBodyName. O valor do campo ou da chave é armazenado como um ByteString no corpo da mensagem. Se você não definir messageBodyName, o corpo da mensagem vai ficar vazio para esquemas de estrutura e mapa.

• Para valores de matriz, o conector oferece suporte apenas a tipos de matriz primitiva. A sequência de valores na matriz é concatenada em um único objeto ByteString.

Conversão do Pub/Sub para o Kafka

O conector de origem converte as mensagens do Pub/Sub em registros do Kafka da seguinte maneira:

• Chave de registro do Kafka: por padrão, a chave é definida como null. Opcionalmente, é possível especificar um atributo de mensagem do Pub/Sub para usar como chave, definindo a opção de configuração kafka.key.attribute. Nesse caso, o conector procura um atributo com esse nome e define a chave do registro como o valor do atributo. Se o atributo especificado não estiver presente, a chave do registro será definida como null.

• Valor do registro do Kafka. O conector grava o valor do registro da seguinte maneira:

  • Se a mensagem do Pub/Sub não tiver atributos personalizados, o conector vai gravar o corpo da mensagem do Pub/Sub diretamente no valor do registro do Kafka como um tipo byte[], usando o conversor especificado por value.converter.

  • Se a mensagem do Pub/Sub tiver atributos personalizados e kafka.record.headers for false, o conector vai gravar uma estrutura no valor do registro. O struct contém um campo para cada atributo e um campo chamado "message", cujo valor é o corpo da mensagem do Pub/Sub armazenado como bytes:

    {
      "message": "<Pub/Sub message body>",
      "<attribute-1>": "<value-1>",
      "<attribute-2>": "<value-2>",
      ....
    }
    

    Nesse caso, use um value.converter compatível com esquemas struct, como org.apache.kafka.connect.json.JsonConverter.

  • Se a mensagem do Pub/Sub tiver atributos personalizados e kafka.record.headers for true, o conector vai gravar os atributos como cabeçalhos de registro do Kafka. Ele grava o corpo da mensagem do Pub/Sub diretamente no valor do registro do Kafka como um tipo byte[], usando o conversor especificado por value.converter.

• Cabeçalhos de registro do Kafka. Por padrão, os cabeçalhos estão vazios, a menos que você defina kafka.record.headers como true.

Observação: na configuração do agente do Kafka, defina message.max.bytes grande o suficiente para processar as mensagens do Pub/Sub. O Pub/Sub tem um tamanho máximo de mensagem.

Opções de configuração

Além das configurações fornecidas pela API Kafka Connect, o conector de Kafka do grupo do Pub/Sub oferece suporte à configuração de origem e de coletor, conforme descrito em Configurações do conector do Pub/Sub.

Como receber suporte

Se precisar de ajuda, crie um tíquete de suporte. Para perguntas e discussões gerais, crie um problema no repositório do GitHub.

A seguir

• Entender as diferenças entre o Kafka e o Pub/Sub.
• Saiba mais sobre o conector do Kafka do grupo do Pub/Sub.
• Consulte o repositório do GitHub do conector do Kafka do Pub/Sub.