Forem: Adriano Avelino

📦 ASDF: Gerenciando versões de linguagens e ferramentas num lugar só

Adriano Avelino — Wed, 27 Nov 2024 00:46:07 +0000

Imagine que você está em um restaurante fast-food e você tem a liberdade de escolher os acompanhamentos do seu lanche. Você pode optar por um pão integral, carne de frango grelhada, queijo cheddar e tomate. Mas e se, na próxima vez, você quiser um hambúrguer com bacon e cebola caramelizada?

O asdf funciona como esse fast-food. No lugar de lanches, temos linguagens de programação e ferramentas como Python, Ruby, Node.js, etc. Em vez de ingredientes, temos versões específicas dessas linguagens.

Como funciona:

Você, o cliente: É o desenvolvedor que precisa de uma determinada versão de uma linguagem para um projeto específico.
O menu: É o repositório de plugins do asdf, onde você encontra todas as linguagens e ferramentas disponíveis.
O funcionário: É o próprio asdf, que vai buscar a versão exata da linguagem que você escolheu e configurá-la para você. Exemplo:

Imagine que você está trabalhando em dois projetos: um que usa Python 3.6 e outro que usa Python 3.9. Com o asdf, você pode instalar ambas as versões e configurar cada projeto para usar a versão correta, sem conflitos.

Em resumo:

O asdf é como um restaurante fast-food onde você monta o seu próprio lanche, mas em vez de comida, você está "montando" o seu ambiente de desenvolvimento. É uma ferramenta poderosa e flexível que pode te ajudar a gerenciar múltiplas versões de linguagens de programação de forma eficiente e organizada.

Neste post, vamos explorar como instalar e usar o asdf para gerenciar suas dependências de forma eficaz. Mas antes, vamos entender:

O que é o asdf
Plugins e versões
Principais Comandos do asdf
Instalação do asdf
Guia Prático para Instalação de Plugins e Versões com o asdf
(Opcional) Laboratório

O que é o asdf?

O asdf é um gerenciador de versões universal que permite que você utilize múltiplas versões de linguagens de programação e ferramentas, tudo em um só lugar. Com ele, você pode alternar entre diferentes versões de uma mesma linguagem como se estivesse trocando de roupa, facilitando seu fluxo de trabalho e evitando conflitos.

Ao contrário de outros gerenciadores, como o rbenv para Ruby ou o nvm para Node.js, o asdf se destaca pela sua versatilidade. Ele suporta uma variedade de linguagens através de plugins. Você pode ter várias versões do Node.js, Python ou Java instaladas e alternar entre elas sem esforço.

Se ainda não ficou muito claro, Fábio Akita fala e dá algumas dicas de uso do asdf no seu vídeo The DEFINITIVE UBUNTU Guide for Beginning Devs, no minuto 40:42.

Neste post, vamos explorar como instalar e usar o asdf para gerenciar suas dependências de forma prática. Mas antes, vamos entender dois conceitos-chave: plugins e versões.

Plugins e versões

De acordo com a documentação oficial, plugins são as extensões que permitem ao asdf gerenciar diferentes ferramentas, como Node.js, Ruby e Elixir. Já as versões são as diferentes iterações das dependências que você pode utilizar. Por exemplo, você pode optar pela versão v20.18.0 do Node.js para um projeto específico, enquanto usa uma versão diferente para outro.

Pronto para mergulhar no mundo do asdf? Então vamos à prática!

Principais Comandos do asdf

Esses são os comandos essenciais para instalar, listar e gerenciar versões no asdf.

asdf list: Lista todos os plugins e as versões já instaladas no seu ambiente. Ideal para verificar rapidamente o que está disponível.
asdf plugin add <NOME-DO-PLUGIN>: Adiciona um plugin ao asdf para que ele gerencie uma nova linguagem ou ferramenta. Por exemplo, para adicionar o plugin do Python, digite:

asdf plugin add python

asdf list all <NOME-DO-PLUGIN>: Lista todas as versões disponíveis para um plugin específico, permitindo que você escolha a que deseja instalar. Exemplo para ver as versões do Python:

asdf list all python

asdf install <NOME-DO-PLUGIN> <VERSAO>: Instala uma versão específica de uma linguagem ou ferramenta. Para instalar a versão 3.13.0 do Python, por exemplo, use:

asdf install python 3.13.0

asdf global <NOME-DO-PLUGIN> <VERSAO>: Define uma versão específica de uma linguagem ou ferramenta como padrão para todo o sistema. Para definir a versão 3.13.0 do Python como global, digite:

asdf global python 3.13.0

asdf local <NOME-DO-PLUGIN> <VERSAO>: Define uma versão específica de uma linguagem ou ferramenta apenas para o diretório atual do projeto. Navegue até a pasta do projeto e execute o comando. Por exemplo:

asdf local python 3.13.0

Observação: Este comando cria um arquivo chamado .tool-versions na pasta do projeto, permitindo configurar uma versão específica para cada repositório, independentemente da versão global.

Agora que já conhecemos os comandos básicos, vamos turbinar o nosso ambiente com alguns plugins e versões. Vêm comigo!

Instalação do asdf

Observação: se você é daquelas pessoas que gosta de fazer testes na ferramenta antes de configurar no seu ambiente, você pode ir para o item (Opcional) Laboratório e práticar em um container docker para avaliar o asdf.

Clone o respositorio para a sua pasta home:

git clone https://github.com/asdf-vm/asdf.git ~/.asdf --branch v0.14.1

Execute os comandos abaixo para configurar o seu ~/.bashrc:

echo '. "$HOME/.asdf/completions/asdf.bash"' >> ~/.bashrc
echo '. "$HOME/.asdf/asdf.sh"' >> ~/.bashrc

Aplique as configurações no terminal ativo:

source ~/.bashrc

Para outros tipos de shell como o zsh ou fish shell, acesse a documentação oficial do asdf.

Guia Prático para Instalação de Plugins e Versões com o asdf

Vamos instalar plugins e definir versões de linguagens e ferramentas populares com o asdf! A cada passo, você verá como configurar o ambiente para Node.js, Java, Maven e Python.

🚀 Começando com Node.js

Para configurar o Node.js no seu ambiente, siga estes passos:

1- Instalar dependências: Vamos garantir que o sistema tenha todas as dependências que você precisa.

sudo apt-get install dirmngr gpg curl gawk

2- Adicionar o plugin do Node.js: Com o plugin, o asdf será capaz de gerenciar versões do Node.

asdf plugin add nodejs https://github.com/asdf-vm/asdf-nodejs.git

3- Verificar o plugin instalado:

asdf list

Você deve ver algo assim:

nodejs
  No versions installed

4- Instalar uma versão do Node.js:

O primeiro passo antes de instalar uma versão, é escolher uma versão disponível e para isso liste as versões disponíveis:

asdf list all nodejs

Você deve ver uma lista semelhante ao exemplo abaixo:

...
22.11.0
23.0.0
23.1.0

Escolha uma das versões disponíveis e execute o comando abaixo para instalar, por exemplo, a versão 23.0.0:

asdf install nodejs 23.0.0

5- Definir a versão do Node.js:

Globalmente (para todo o sistema):

asdf global nodejs 23.0.0

Não se esqueça, o próximo passo é opcional, mas para executar, instale a versão latest com o comando asdf install nodejs latest.

Localmente (apenas no diretório atual):

asdf local nodejs latest # ou qualquer outra versão disponível

6- Verifique a versão do nodejs:

node --version

☕ Configurando o Java

1- Adicionar o plugin do Java: Isso permite instalar e gerenciar versões do Java no asdf.

asdf plugin add java https://github.com/halcyon/asdf-java.git

2- Verificar as versões disponíveis do Java:

asdf list all java

3- Configurar a variável JAVA_HOME: Para garantir que o sistema identifique o Java.

echo . ~/.asdf/plugins/java/set-java-home.bash >> ~/.bashrc

4- Instalar uma versão específica do Java:

asdf install java adoptopenjdk-8.0.432+6

Observação: você pode executar o comando asdf list all java para ver as versões disponíveis.

5- Verificar a versão do Java instalada:

java -version

6- Definir a versão do Java globalmente:

asdf global java adoptopenjdk-8.0.432+6

🔧 Configurando o Maven

1- Adicionar o plugin do Maven:

asdf plugin add maven

2- Instalar a versão mais recente do Maven:

asdf install maven latest

3- Definir a versão global do Maven:

asdf global maven latest

4- Verifique a versão do maven instalada:

mvn --version

🐍 Configurando o Python

1- Instalar dependências para o Python:

sudo apt-get update && sudo apt-get install -y \
       make build-essential libssl-dev zlib1g-dev \
       libbz2-dev libreadline-dev libsqlite3-dev wget curl \
       llvm libncurses5-dev libncursesw5-dev \
       xz-utils tk-dev libffi-dev liblzma-dev \
       git

2- Adicionar o plugin do Python:

asdf plugin add python

3- Instalar a versão mais recente do Python:

asdf install python latest

4- Definir a versão global do Python:

asdf global python latest

5- Verifique a versão do python:

python --version

Esses passos são suficientes para instalar e configurar as linguagens e ferramentas essenciais no asdf. Com essas instruções, seu ambiente estará preparado para alternar entre versões específicas conforme necessário.

(Opcional) Laboratório

Se você está aqui é porque você ficou interessado em testar o asdf antes de configurar em sua máquina. Sendo assim, vamos lá!

Pré-requisitos:

Docker
Docker Compose

Criando o ambiente

Para facilitar o laboratório com o asdf, preparamos um ambiente configurado e pronto para uso em um container Docker. Você só precisa criar os arquivos indicados abaixo:

Arquivo Dockerfile:

FROM ubuntu:24.04

# Atualiza o sistema e instala as dependências
RUN apt-get update && apt-get install -y \
    make build-essential libssl-dev zlib1g-dev \
    libbz2-dev libreadline-dev libsqlite3-dev wget curl \
    llvm libncurses5-dev libncursesw5-dev \
    xz-utils tk-dev libffi-dev liblzma-dev \
    git nano sudo

# Adiciona um novo usuário
RUN useradd -m -s /bin/bash usuario && \
echo "usuario:123" | chpasswd && \
usermod -aG sudo usuario

# Da permissões sudo sem senha para o novo usuário
RUN echo "usuario ALL=(ALL) NOPASSWD: ALL" >> /etc/sudoers

# Comando inicial para rodar o container no terminal do novo usuário
USER usuario

# Instala e configura o asdf
RUN git clone https://github.com/asdf-vm/asdf.git ~/.asdf --branch v0.14.1
RUN echo '. "$HOME/.asdf/completions/asdf.bash"' >> ~/.bashrc && \
    echo '. "$HOME/.asdf/asdf.sh"' >> ~/.bashrc

CMD ["/bin/bash"]

Arquivo docker-compose.yml:

services:
  asdf:
    build:
      context: .  # Indica o diretório onde está o Dockerfile
      dockerfile: Dockerfile
    container_name: asdf
    tty: true  # Mantém o terminal aberto

Construindo e executando o container

Com os arquivos prontos, execute o container com o comando abaixo:

docker compose up -d --build

Onde:

-d: Executa o container em segundo plano
--build: Recria o container se o Dockerfile mudar

💡 Isso deve demorar um tempinho para instalar todas as dependências... ⏳ Aproveite para esticar as canelas e tomar uma água 💧!

Após a instalação, acesse o container com o seguinte comando:

docker compose exec asdf bash

Agora você está pronto para seguir com o tutorial no tópico Guia Prático para Instalação de Plugins e Versões com o asdf e testar os comandos do asdf!

Conector Kafka Lambda Sink com Localstack

Adriano Avelino — Wed, 09 Oct 2024 01:30:59 +0000

Introdução

Ao trabalhar em projetos corporativos, é comum mover informações de um lugar para outro utilizando o Kafka e seus conectores source e sink. Os conectores source são responsáveis por enviar dados aos tópicos Kafka, enquanto os conectores sink exportam essas informações para sistemas externos, como bancos de dados, ferramentas de log ou serviços na AWS, como uma Lambda. Neste tutorial, abordaremos especificamente o conector sink, mais precisamente o Lambda Sink Connector, utilizando o Localstack para simular o serviço AWS Lambda.

Pré-requisitos

Antes de começar, certifique-se de ter os seguintes itens configurados no seu ambiente:

Docker: para criar os containers que rodarão o Kafka e o Localstack.
Docker Compose: para orquestrar os containers.
AWS CLI: necessário para executar comandos relacionados à AWS.
cURL ou uma ferramenta similar como o Insomnia: para realizar requisições HTTP.
Arquivo de configuração AWS (~/.aws/config) configurado da seguinte forma:

[default]
region = us-east-1
output = json

Sobre o projeto

O objetivo deste projeto é simular um ambiente corporativo usando Kafka e AWS Lambda, de forma acessível e sem custos, utilizando o Localstack. Isso permite que você emule serviços da AWS localmente, facilitando o desenvolvimento e testes.

No fluxo deste projeto, uma mensagem é enviada por um producer para um tópico Kafka, que por sua vez dispara um evento para o conector Kafka Sink. O conector envia essa mensagem para uma função Lambda, conforme ilustrado no diagrama abaixo:

Para isso, utilizaremos o Docker Compose para gerenciar os containers do Kafka e do Localstack. A imagem Docker fast-data-dev será usada para o Kafka, pois já inclui uma interface gráfica e ferramentas como Zookeeper, Kafka Cluster, Schema Registry e Kafka Connect. O Localstack será usado para emular serviços da AWS, incluindo a Lambda.

Se você preferir uma visualização gráfica da Lambda em execução, o Localstack possui uma interface web. Além disso, existe um vídeo introdutório sobre o Localstack no YouTube, caso queira entender mais sobre a ferramenta.

Tela da interface gráfica do Localstack Web mostrando os logs de execução de uma Lambda.

Para configurar o Kafka Connect, utilizaremos o plugin Lambda Sink que oferece suporte ao Localstack. Este plugin é um fork do projeto kafka-connect-lambda, que não funciona com o Localstack. Caso tenha interesse em contribuir, a sua contribuição é bem-vinda!

Tela da interface gráfica do fast-data-dev mostrando o conector Lambda Sink.

Para enviar mensagens ao tópico Kafka, usaremos um producer. Existem diversas opções para isso:

Linha de comando: Usando o kafka-console-producer do container Kafka.
Kafkacat: Ferramenta de linha de comando para interagir com o Kafka.
JMeter com o plugin Kafka: Veja mais detalhes na seção Dicas e Recomendações.
Conector Data Source como o Voluble.
Criar uma aplicação customizada na sua linguagem de programação preferida.

Para simplificar, neste tutorial, utilizaremos o kafka-console-producer.

Tela do terminal executando Kafka Console Producer na linha de comando

Criação dos arquivos de configuração

Antes de tudo começar a funcionar, é necessário criarmos os arquivos de configuração da infraestrutura do Kafka, Lambda e conector Lambda Sink. Vamos começar configurando o Docker Compose para orquestrar Kafka e LocalStack, seguido do CloudFormation para provisionar a função Lambda no Localstack. Em seguida, definimos o conector Kafka e baixamos o plugin Lambda Sink. Por fim, verificamos a estrutura final do projeto, listando os arquivos criados para garantir que estão no local correto e prontos para execução. Vamos nessa!

Docker Compose

Começaremos criando o arquivo docker-compose.yml, que será responsável por orquestrar os containers do Kafka e do Localstack. Ele define os serviços necessários, como o Kafka (com a imagem fast-data-dev) e o Localstack para emular a AWS. Crie o arquivo ./docker-compose.yml com o seguinte conteúdo:

services:
  fast-data-dev:
    image: landoop/fast-data-dev:3.3
    ports:
      - "2181:2181"
      - "3030:3030"
      - "8081-8083:8081-8083"
      - "9581-9585:9581-9585"
      - "9092:9092"
    environment:
      ADV_HOST: 127.0.0.1
      DEBUG: 1
      RUNTESTS: 0
      AWS_ACCESS_KEY_ID: local
      AWS_SECRET_ACCESS_KEY: local
      CONNECT_PLUGIN_PATH: /var/run/connect/connectors/stream-reactor,/var/run/connect/connectors/third-party,/connectors
    volumes:
      - ./plugins:/connectors/kafka-connect-lambda-localstack
    network_mode: host

  localstack:
    container_name: "${LOCALSTACK_DOCKER_NAME-localstack-main}"
    image: localstack/localstack:2.3
    ports:
      - "127.0.0.1:4566:4566"            # LocalStack Gateway
      - "127.0.0.1:4510-4559:4510-4559"  # external services port range
    environment:
      - DEBUG=${DEBUG-}
      - DOCKER_HOST=unix:///var/run/docker.sock
    volumes:
      - "${LOCALSTACK_VOLUME_DIR:-./volume}:/var/lib/localstack"
      - "/var/run/docker.sock:/var/run/docker.sock"
    depends_on: [fast-data-dev]
    network_mode: host

CloudFormation

Agora, criaremos um arquivo cloudformation.yml para provisionar a função Lambda simulada no Localstack. O arquivo também define as permissões necessárias para a execução da Lambda. Crie o arquivo ./cloudformation.yml com o seguinte conteúdo:

AWSTemplateFormatVersion: '2010-09-09'
Description: 'Lambda connector example function'
Resources:
  ExampleFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: example-function
      Handler: index.handler
      Runtime: python3.7
      Role: !GetAtt 'ExampleFunctionRole.Arn'
      Code:
        ZipFile: |
          import json
          def handler(event, context):
            print(json.dumps(event))
            return event

  ExampleFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: example-lambda-role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service: lambda.amazonaws.com
          Action: sts:AssumeRole
      ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

Outputs:
  ExampleFunctionArn:
    Value: !GetAtt 'ExampleFunction.Arn'
  ExampleFunctionRoleArn:
    Value: !GetAtt 'ExampleFunctionRole.Arn'

Configuração do conector Kafka

Agora, crie o arquivo de configuração do conector Lambda Sink. Este arquivo define as propriedades do conector, incluindo o tópico Kafka e a função Lambda a ser invocada. Crie o arquivo ./connector-localstack.json com o seguinte conteúdo:

{
  "name": "example-lambda-connector-localstack",
  "config": {
    "tasks.max": "1",
    "connector.class": "com.nordstrom.kafka.connect.lambda.LambdaSinkConnector",
    "topics": "example-stream",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "aws.region": "us-east-1",
    "aws.lambda.function.arn": "arn:aws:lambda:us-east-1:000000000000:function:example-function",
    "aws.lambda.invocation.timeout.ms": "60000",
    "aws.lambda.invocation.mode": "SYNC",
    "aws.lambda.batch.enabled": "false",
    "localstack.enabled": "true"
  }
}

Download do plugin Lambda Sink

Por fim, faça o download do plugin Lambda Sink e salve-o no diretório ./plugins. Execute o seguinte comando no terminal:

mkdir ./plugins & curl -L -o ./plugins/kafka-connect-lambda-localstack-1.4.0.jar \
https://github.com/adrianoavelino/kafka-connect-lambda-localstack/releases/download/v1.4.0/kafka-connect-lambda-localstack-1.4.0.jar

A última versão do plugin é a 1.4.0. Para versões mais recentes, consulte as releases no GitHub.

Estrutura final do projeto

Ao finalizar esta etapa, a estrutura de arquivos do projeto deverá ser semelhante a esta:

.
├── cloudformation.yml
├── connector-localstack.json
├── docker-compose.yml
├── plugins
│   └── kafka-connect-lambda-localstack-1.4.0.jar

Criação da infrastrutura

Para que nossa aplicação comece a funcionar, precisamos preparar o terreno: vamos criar a infraestrutura que dará vida ao Kafka, à Lambda e ao conector Lambda Sink.

Iniciaremos com a inicialização dos containers. Em seguida, provisionaremos a função Lambda usando o CloudFormation. Depois, criaremos o conector Kafka.

Por fim, vamos conferir se tudo foi criado corretamente. Preparados? Então, vamos lá!

Inicialização dos containers

Inicie os containers do Kafka e do Localstack com o seguinte comando:

docker compose up -d

Se você é uma pessoa como eu que adora acompanhar o passo a passo de cada informação nos logs, é possível utilizar o comando docker compose logs -f.

Para garantir que tudo está funcionando como devia, podemos realizar algumas verificações:

docker compose ps: para validar se os containers estejam no estado healthy
Acesse o endereço http://localhost:3030/ para acessar a interface gráfica do Landoop para visualizar os tópicos, conectores e plugins dos conectores instalados.
curl --url http://localhost:8083/connector-plugins/: listas os plugins de conectores disponíveis, verifique se o plugin com a class com.nordstrom.kafka.connect.lambda.LambdaSinkConnector está disponível.

Criação da lambda

Crie uma Lambda no Localstack utilizando Cloudformation com seguinte comando:

aws cloudformation create-stack \
--stack-name example-lambda-stack \
--capabilities CAPABILITY_NAMED_IAM \
--template-body file://cloudformation.yml \
--endpoint-url http://localhost:4566

Se a stack for criada com sucesso, a saída será semelhante a esta:

{
    "StackId": "arn:aws:cloudformation:us-east-1:000000000000:stack/example-lambda-stack/d61cbd21"
}

Para verificar se a Lambda foi criada corretamente, execute:

aws lambda list-functions --endpoint-url=http://localhost:4566

A saída esperada será semelhante ao exemplo abaixo:

{
    "Functions": [
        {
            "FunctionName": "example-function",
            "FunctionArn": "arn:aws:lambda:us-east-1:000000000000:function:example-function",
            "Runtime": "python3.7",
            "Role": "arn:aws:iam::000000000000:role/example-lambda-role",
            "Handler": "index.handler",
            "CodeSize": 1630,
            "Description": "",
            "Timeout": 3,
            "MemorySize": 128,
            "LastModified": "2024-09-09T01:17:54.471773+0000",
            "CodeSha256": "qDSOE5NTun0FiK+cFAAZGHPqarSjlyJtlGMCPPRpJ8Y=",
            "Version": "$LATEST",
            "TracingConfig": {
                "Mode": "PassThrough"
            },
            "RevisionId": "debe4c55-40b0-46e3-94b4-acfc1aeb17fb",
            "PackageType": "Zip",
            "Architectures": [
                "x86_64"
            ],
            "EphemeralStorage": {
                "Size": 512
            },
            "SnapStart": {
                "ApplyOn": "None",
                "OptimizationStatus": "Off"
            }
        }
    ]
}

Para testar a Lambda, use o seguinte comando:

aws lambda invoke --function-name example-function \
--cli-binary-format raw-in-base64-out \
--payload '{"value": "my example"}' --output text result.txt \
--endpoint-url http://localhost:4566

Se tudo ocorreu bem, você deve receber a seguinte resposta:

$LATEST Unhandled       200

Criação do conector

A criação do conector Lambda Sink pode ser feita via API Kafka Connect REST ou pela interface gráfica do Landoop. Neste exemplo, utilizaremos o curl para enviar as requisições, mas você também pode usar ferramentas como o Insomnia ou similares.

curl -XPOST -H "Content-Type: application/json" \
http://localhost:8083/connectors \
-d @connector-localstack.json

Dica: No Insomnia, você pode importar comandos curl para gerar a requisição automaticamente. Confira este vídeo tutorial ou consulte a documentação oficial para mais detalhes.

Você deve receber a seguinte resposta:

{
  "name": "example-lambda-connector-localstack",
  "config": {
    "tasks.max": "1",
    "connector.class": "com.nordstrom.kafka.connect.lambda.LambdaSinkConnector",
    "topics": "example-stream",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "aws.region": "us-east-1",
    "aws.lambda.function.arn": "arn:aws:lambda:us-east-1:000000000000:function:example-function",
    "aws.lambda.invocation.timeout.ms": "60000",
    "aws.lambda.invocation.mode": "SYNC",
    "aws.lambda.batch.enabled": "false",
    "localstack.enabled": "true",
    "name": "example-lambda-connector-localstack"
  },
  "tasks": [],
  "type": "sink"
}

Para validar o status do conector, execute o seguinte comando curl:

curl --request GET \
  --url http://localhost:8083/connectors/example-lambda-connector-localstack/status

A resposta deve se algo como:

{
  "name": "example-lambda-connector-localstack",
  "connector": {
    "state": "RUNNING",
    "worker_id": "127.0.0.1:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "127.0.0.1:8083"
    }
  ],
  "type": "sink"
}

Observação: Este comando também pode ser utilizado para diagnosticar problemas de integração entre o conector Kafka e a Lambda no Localstack.

Enviar mensagem no kafka

Agora que tudo está configurado e em execução, é hora de fazer o Kafka funcionar de verdade! Vamos aprender a enviar mensagens para o tópico que criamos. Neste momento, vamos explorar duas opções para enviar mensagens: uma forma rápida e direta para quem precisa de agilidade e outra que permite o envio contínuo de múltiplas mensagens. Com essas opções, você poderá interagir facilmente com o Kafka e começar a testar a comunicação entre os componentes da sua aplicação. Vamos lá!

Opção 1: Envio rápido de mensagem única

Utilize o comando abaixo para enviar uma mensagem única ao Kafka de forma rápida:

echo "teste" | docker compose exec -T fast-data-dev \
kafka-console-producer \
--broker-list localhost:9092 \
--topic example-stream

Obs: em caso de problema de comunicação com o broker ou outro serviço utilize o comando: nc -vz localhost <PORTA>. Ex: nc -vz localhost 9092

Opção 2: Envio contínuo de mensagens

Se preferir manter um terminal ativo para enviar várias mensagens ao Kafka, utilize o seguinte comando:

docker compose exec fast-data-dev \
kafka-console-producer \
--broker-list localhost:9092 \
--topic example-stream

Após a execução, você verá o prompt >, onde poderá inserir várias mensagens ao tópico Kafka. Exemplo de mensagens:

{"value": "my example"}
{"value": "my example 2"}

Como visualizar os logs da Lambda no terminal

Para visualizar os logs da execução da Lambda diretamente no terminal, siga os passos abaixo:

Primeiro, obtenha o nome do grupo de logs da Lambda:

LOG_GROUP=`aws logs describe-log-groups \
--endpoint-url http://localhost:4566 \
--query "logGroups[0].logGroupName" | sed 's/"//g'`

Depois, use o comando aws logs tail para seguir os logs em tempo real:

aws logs tail $LOG_GROUP --follow --endpoint-url http://localhost:4566

A resposta esperada é algo semelhante ao exempĺo abaixo:

2024-09-15T18:27:34.888000+00:00 2024/09/15/[$LATEST]57661289d19ebedfe4a6782395866989 START RequestId: 61ab5b4f-569a-4348-905b-b15ceadfcc26 Version: $LATEST
2024-09-15T18:27:34.902000+00:00 2024/09/15/[$LATEST]57661289d19ebedfe4a6782395866989 {"key": "", "keySchemaName": null, "value": "teste", "valueSchemaName": null, "topic": "example-stream", "partition": 0, "offset": 1, "timestamp": 1726424854776, "timestampTypeName": "CreateTime"}
2024-09-15T18:27:34.917000+00:00 2024/09/15/[$LATEST]57661289d19ebedfe4a6782395866989 END RequestId: 61ab5b4f-569a-4348-905b-b15ceadfcc26
2024-09-15T18:27:34.932000+00:00 2024/09/15/[$LATEST]57661289d19ebedfe4a6782395866989 REPORT RequestId: 61ab5b4f-569a-4348-905b-b15ceadfcc26    Duration: 7.85 ms   Billed Duration: 8 msMemory Size: 128 MB    Max Memory Used: 128 MB

Observação: também é possível visualizar os logs da Lambda diretamente na interface gráfica do Localstack.

Automatização

Parabéns! Se você chegou até aqui, já percorreu um longo caminho e fez tudo manualmente ou ... você foi uma pessoa "espertinha" e encontrou um atalho aqui para chegar no parque de diversão.

Aqui não vamos precisar criar os arquivos de configuração e nem a criação da infraestrutura de forma manual, tudo está automatizado. Seguindo os passo abaixo você terá o ambiente completo configurado para enviar mensagens ao tópico Kafka e validar os logs. Vamos a prática:

Clone o repositório:

git clone https://github.com/adrianoavelino/posts.git

Entre no diretório automation:

cd posts/dev.to/2024-09-01-conector-lambda-sink-localstack/automation/

Adicione a permissão de execução nos arquivos de scrips:

chmod +x init-scripts/*.sh

Inicie os containers:

docker compose up

Agora é só aguardar a inicialização e criação da lambda e conector Lambda Sink para inicar os testes. Se o procorreu como planejado, você deve ver algo parecido com o exemplo abaixo:

localstack-main  | ########### script 02 - Lambda function has been invoked ###########
fast-data-dev-1  | Sat 28 Sep 2024 02:15:31 AM UTC  Kafka Connect listener HTTP state:  000  (waiting for 200)
fast-data-dev-1  | Sat 28 Sep 2024 02:15:36 AM UTC  Kafka Connect listener HTTP state:  000  (waiting for 200)
fast-data-dev-1  | Sat 28 Sep 2024 02:15:41 AM UTC  Kafka Connect listener HTTP state:  000  (waiting for 200)
fast-data-dev-1  | Sat 28 Sep 2024 02:15:47 AM UTC  Kafka Connect listener HTTP state:  200  (waiting for 200)
fast-data-dev-1  | 
fast-data-dev-1  | --
fast-data-dev-1  | +> Creating Lambda Sink Connector with avro
fast-data-dev-1  | {"name":"example-lambda-connector-localstack","config":{"tasks.max":"1","connector.class":"com.nordstrom.kafka.connect.lambda.LambdaSinkConnector","topics":"example-stream","key.converter":"org.apache.kafka.connect.storage.StringConverter","value.converter":"org.apache.kafka.connect.storage.StringConverter","aws.region":"us-east-1","aws.lambda.function.arn":"arn:aws:lambda:us-east-1:000000000000:function:example-function","aws.lambda.invocation.timeout.ms":"60000","aws.lambda.invocation.mode":"SYNC","aws.lambda.batch.enabled":"false","localstack.enabled":"true","name":"example-lambda-connector-localstack"},"tasks":[],"type":"sink"}2024-09-28 02:15:53,267 INFO exited: logs-to-kafka (exit status 0; expected)

Execute o seguinte comando para enviar um evento ao tópico Kafka:

echo "teste" | docker compose exec -T fast-data-dev \
kafka-console-producer \
--broker-list localhost:9092 \
--topic example-stream

Agora você pode começar a realizar seus testes e ajustar a configuração conforme necessário. Se encontrar algum problema, não se preocupe, basta verificar os logs dos containers para identificar possíveis erros. E deixa eu te contar mais um segredo, deixei várias dicas legais logo abaixo.

Erros e Soluções: Desvendando Problemas com Estilo

Vamos lá, todo mundo já passou por isso: algo dá errado e você fica ali, encarando a tela, tentando entender o que aconteceu. Mas calma, eu tô aqui pra te ajudar a resolver esses pepinos! 😎

1. Logs: Seus Melhores Amigos

Primeira dica de ouro: sempre dê uma olhada nos logs dos containers. Eles são tipo aquele amigo que te conta o que realmente está acontecendo por trás das cortinas. Você pode conferir os logs de cada serviço separadamente, assim:

No LocalStack:

docker compose logs -f localstack

Ou no fast-data-dev:

docker compose logs -f fast-data-dev

Agora, se você estiver lidando com o fast-data-dev e não estiver vendo todos os logs, não se desespere! Basta rodar o seguinte comando pra dar uma espiada no arquivo de logs dentro do container:

docker compose exec fast-data-dev cat /var/log/broker.log

E se quiser saber mais sobre o fast-data-dev, dá uma olhada no Docker Hub. Eles têm tudo lá!

2. Erro Clássico: "Function not found"

Ah, o famoso erro "Function not found arn:aws:lambda:us-east-1:000000000000:function:example-function". Isso geralmente significa que a Lambda que você está tentando chamar não foi encontrada. Provavelmente, ela foi criada em outra região. Então, dá uma conferida no seu arquivo de configuração do AWS CLI (~/.aws/config) e veja se a região está certinha.

3. Erros no Conector Kafka: Como Encontrar?

Se o problema for no conector Kafka, você pode usar a api do Kafka Connect pra descobrir o que tá rolando. Aqui vai o comando mágico:

curl --request GET \
  --url http://localhost:8083/connectors/example-lambda-connector-localstack/status

Exemplo de Erro: Lambda Não Encontrada

Se a Lambda não foi criada no Localstack, você vai ver algo assim:

{
  "name": "example-lambda-connector-localstack",
  "connector": {
    "state": "RUNNING",
    "worker_id": "127.0.0.1:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "FAILED",
      "worker_id": "127.0.0.1:8083",
      "trace": "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:611)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:333)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:244)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: com.nordstrom.kafka.connect.lambda.InvocationClient$InvocationException: java.util.concurrent.ExecutionException: com.amazonaws.services.lambda.model.ResourceNotFoundException: Function not found: arn:aws:lambda:us-east-1:000000000000:function:example-function (Service: AWSLambda; Status Code: 404; Error Code: ResourceNotFoundException; Request ID: 52512fac-927b-4db0-a910-907270c4166f; Proxy: null)\n\tat com.nordstrom.kafka.connect.lambda.InvocationClient.invoke(InvocationClient.java:71)\n\tat com.nordstrom.kafka.connect.lambda.LambdaSinkTask.invoke(LambdaSinkTask.java:190)\n\tat com.nordstrom.kafka.connect.lambda.LambdaSinkTask.put(LambdaSinkTask.java:86)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:581)\n\t... 10 more\n"
    }
  ],
  "type": "sink"
}

4. Interface Gráfica: Porque Nem Todo Mundo Gosta de Linha de Comando

Se você é do tipo que prefere ver as coisas de forma visual (e quem não gosta, né?), o fast-data-dev tem uma interface gráfica bem bacana pra você acompanhar tudo que está rolando com seus conectores e tópicos do Kafka.

Pronto! Agora você já sabe como lidar com esses erros chatos e seguir em frente com seu projeto. E lembre-se: os logs são seus amigos, e a interface gráfica tá aí pra te ajudar quando a linha de comando não for suficiente. 🚀

Dicas e recomendações

No nosso passo a passo, colocamos o código-fonte da Lambda diretamente dentro do template do CloudFormation. Fizemos isso porque é um exemplo bem simples. No entanto, é possível criar uma Lambda com várias dependências usando ferramentas como Terraform ou Serverless, ambas integradas ao Localstack.

Se você quiser ver um exemplo prático, recomendo o vídeo no YouTube do Localstack, que ensina como criar uma Lambda com hot reload. Isso significa que você não precisa empacotar todo o código-fonte e suas dependências a cada alteração para testar a execução da Lambda no Localstack. Aqui está o repositório no GitHub utilizado no vídeo, além de um exemplo de uso que fiz, onde testei uma Lambda com um conversor de Avro (no nosso guia, usamos um conversor de string).

Outra maneira de acelerar a criação de Lambdas é utilizando o Serverless Framework. Ele simplifica a implantação de aplicações serverless. Para ajudar, deixo aqui um repositório no GitHub com exemplos de uso com Localstack, tanto em Node.js quanto em Python.

Para produzir mensagens em um tópico Kafka via linha de comando, você pode usar o kcat. Se preferir uma interface gráfica, o JMeter com o plugin Kafka é uma ótima opção.

Tela da interface gráfica do JMeter com um plugin producer Kafka enviando diversas mensagens a um tópico Kafka.

Para facilitar, aqui estão alguns tutoriais de instalação:

Para instalar o plugin Kafka no JMeter, recomendo usar o Plugin Manager, que facilita bastante o processo. Também deixo aqui um arquivo de configuração para o plugin no JMeter, para agilizar ainda mais. Com ele você pode realizar o download e importar as configurações do producer Kafka no JMeter.

Se você está buscando um desafio maior, pode experimentar o conector de data source Voluble. Ele permite gerar eventos automaticamente para um tópico Kafka. Confira um exemplo de uso no vídeo 🎄 Twelve Days of SMT 🎄 - Day 1: InsertField (timestamp) ou veja algumas anotações que podem te ajudar na prática.

Links

Observação: todo conteúdo foi criado com auxílio da Inteligência Artificial Stackspot AI.

Multiplos schemas no mesmo tópico Kafka na linha de comando

Adriano Avelino — Thu, 04 Jan 2024 01:28:23 +0000

Recentemente, tive a oportunidade de trabalhar em um projeto que utiliza três tipos distintos de eventos (schemas) no mesmo tópico do Kafka. Até então, só havia utilizado um único schema Avro por tópico. Isso só foi possível devido ao uso da estratégia de nome do subject (Subject name strategy) no Schema Registry.

Teoria

Um subject no Schema Registry é uma coleção de schemas associados a um tópico ou a um schema específico usado na serialização e desserialização de dados em um tópico do Kafka. Em termos simples, um "subject" é uma chave sob a qual os schemas são registrados no Schema Registry. Como exemplos de nomes de subjects podemos citar: transacoes-value e transacoes-key, repectivamente relacionados ao valor e a chave de identificação de um evento enviado a um tópico Kafka.

O Schema Registry suporta diversos tipos de schemas, mas estaremos abordando a utilização de schemas avro, por exemplo:



{
  "type": "record",
  "name": "Transacao",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "valor", "type": "double"},
    {"name": "data", "type": "string"},
    {"name": "tipo", "type": "string"},
    {"name": "conta_origem", "type": "string"},
    {"name": "conta_destino", "type": "string"},
    {"name": "descricao", "type": "string"}
  ]
}



{
  "type": "record",
  "name": "DetalhesConta",
  "fields": [
    {"name": "conta_id", "type": "string"},
    {"name": "titular", "type": "string"},
    {"name": "saldo", "type": "double"},
    {"name": "tipo_conta", "type": "string"},
    {"name": "agencia", "type": "string"},
  ]
}

Mais informações relacionadas a arquivos avro podem ser encontradas na documentação do Apache Avro.

Seguindo a estratégia de nome de subject padrão, o nome do subject num tópico chamado transacoes seria: transacoes-value.

O uso de mais de um schema no mesmo tópico Kafka é possível através do uso da estratégia de nome do subject (Subject name strategy). Existem três opções: TopicNameStrategy, RecordNameStrategy e TopicRecordNameStrategy.

O TopicNameStrategy é o padrão e gera o nome do subject com base no nome do tópico seguido do caractere -, mais a palavra key ou value. No RecordNameStrategy o nome do subject é baseado no nome do record do schema avro. Já o TopicRecordNameStrategy, o nome subject é baseado no nome do tópico seguido do caractere -, mais o nome do record do schema avro. Veja a tabela abaixo com exemplos dos nomes gerados num tópico chamado transacoes e nome do record br.com.DetalhesConta:

Nome da estratégia do Subject	Exemplo Subject gerado
TopicNameStrategy	transacoes-value ou transacoes-key
RecordNameStrategy	DetalhesConta
TopicRecordNameStrategy	transacoes-DetalhesConta

Seguindo essa lógica fica fácil entender porque não conseguimos utilizar mais de um schema utilizando a estratégia padrão (TopicNameStrategy). A formação do nome do subject é a mesma, tanto para um schema Transacao ou DetalhesConta, por exemplo. Ao cadastrar os schemas, ambos geram o mesmo nome do subject: transacoes-value.

O problema

Vamos à prática, executando o ambiente Kafka no docker-compose.yml com o seguinte conteúdo:



version: '3.7'

services:
  fast-data-dev:
    image: landoop/fast-data-dev:3.3
    ports:
      - "2181:2181"
      - "3030:3030"
      - "8081-8083:8081-8083"
      - "9581-9585:9581-9585"
      - "9092:9092"
    environment:
      ADV_HOST: 127.0.0.1

E executando o comando abaixo:



docker-compose up -d

Agora vamos acessar o bash do container:



docker-compose exec fast-data-dev bash

Dentro do container podemos criar um producer que por padrão cria os subjects e os schemas de forma automática, primeiramente criamos um producer para o schema Transacoes:



kafka-avro-console-producer \
--broker-list localhost:9092 \
--topic transacoes \
--property value.schema='{ "type": "record", "name": "Transacao", "fields": [ {"name": "id", "type": "string"}, {"name": "valor", "type": "double"}, {"name": "data", "type": "string"}, {"name": "tipo", "type": "string"}, {"name": "conta_origem", "type": "string"}, {"name": "conta_destino", "type": "string"}, {"name": "descricao", "type": "string"} ] }' \
--property schema.registry.url=http://localhost:8081

Logo após, colamos o seguinte código:



{ "id": "123456789", "valor": 500.0, "data": "2023-01-01", "tipo": "transferencia", "conta_origem": "987654321", "conta_destino": "567890123", "descricao": "Transferência entre contas" }

Com isso podemos visualizar os subjects cadastrados através da interface gráfica do container landoop, através do terminal utilizando o curl ou qualquer outra aplicação como o Postman ou Insomnia. Para acessar via interface gráfica do landoop, acesse o endereço http://localhost:3030/schema-registry-ui/#/ e filtre pelo nome do subject transacoes-value, conforme a imagem abaixo:

Se preferir também pode utilizar o curl em um novo terminal:



curl -X GET http://localhost:8081/subjects

O resultado deve ser algo semelhante a imagem abaixo:

Obs: a reposta deve conter diversos subjects, mas vamos focar somente no subject transacoes-value para realizar a validação.

Vamos ver o que acontece ao criarmos mais um producer, mas para o schema DetalhesConta. Como comentado anteriormente, por padrão, o producer cria os subjects e os schemas de forma automática. O nosso objetivo ao utilizar um segundo producer é adicionar mais um schema no mesmo tópico e para isso primeiro abrimos um novo terminal e iniciamos o bash:



docker-compose exec fast-data-dev bash

Agora iniciamos o nosso producer:



kafka-avro-console-producer \
--broker-list localhost:9092 \
--topic transacoes \
--property value.schema='{ "type": "record", "name": "DetalhesConta", "fields": [ {"name": "conta_id", "type": "string"}, {"name": "titular", "type": "string"}, {"name": "saldo", "type": "double"}, {"name": "tipo_conta", "type": "string"}, {"name": "agencia", "type": "string"} ] }' \
--property schema.registry.url=http://localhost:8081

Logo após, colamos o seguinte evento para o schema DetalhesConta:



{ "conta_id": "987654321", "titular": "Joaquim", "saldo": 1000.0, "tipo_conta": "corrente", "agencia": "123" }

Devemos receber uma mensagem de erro semelhante ao erro abaixo:



org.apache.kafka.common.errors.InvalidConfigurationException: Schema being registered is incompatible with an earlier schema for subject "transacoes-value", 
details: [Incompatibility{type:NAME_MISMATCH, location:/name, message:expected: Transacao, reader:{"type":"record","name":"DetalhesConta","fields":
[{"name":"conta_id","type":"string"},{"name":"titular","type":"string"},{"name":"saldo","type":"double"},{"name":"tipo_conta","type":"string"},
{"name":"agencia","type":"string"}]}, writer:{"type":"record","name":"Transacao","fields":[{"name":"id","type":"string"},{"name":"valor","type":"double"},
{"name":"data","type":"string"},{"name":"tipo","type":"string"},{"name":"conta_origem","type":"string"},{"name":"conta_destino","type":"string"},
{"name":"descricao","type":"string"}]}}, Incompatibility{type:READER_FIELD_MISSING_DEFAULT_VALUE, location:/fields/0, message:conta_id, reader:{"type":"record",
"name":"DetalhesConta","fields":[{"name":"conta_id","type":"string"},{"name":"titular","type":"string"},{"name":"saldo","type":"double"},{"name":"tipo_conta",
"type":"string"},{"name":"agencia","type":"string"}]}, writer:{"type":"record","name":"Transacao","fields":[{"name":"id","type":"string"},{"name":"valor",
"type":"double"},{"name":"data","type":"string"},{"name":"tipo","type":"string"},{"name":"conta_origem","type":"string"},{"name":"conta_destino",
"type":"string"},{"name":"descricao","type":"string"}]}}, Incompatibility{type:READER_FIELD_MISSING_DEFAULT_VALUE, location:/fields/1, message:titular, reader:
{"type":"record","name":"DetalhesConta","fields":[{"name":"conta_id","type":"string"},{"name":"titular","type":"string"},{"name":"saldo","type":"double"},
{"name":"tipo_conta","type":"string"},{"name":"agencia","type":"string"}]}, writer:{"type":"record","name":"Transacao","fields":[{"name":"id","type":"string"},
{"name":"valor","type":"double"},{"name":"data","type":"string"},{"name":"tipo","type":"string"},{"name":"conta_origem","type":"string"},
{"name":"conta_destino","type":"string"},{"name":"descricao","type":"string"}]}}, Incompatibility{type:READER_FIELD_MISSING_DEFAULT_VALUE, location:/fields/2, 
message:saldo, reader:{"type":"record","name":"DetalhesConta","fields":[{"name":"conta_id","type":"string"},{"name":"titular","type":"string"},{"name":"saldo",
"type":"double"},{"name":"tipo_conta","type":"string"},{"name":"agencia","type":"string"}]}, writer:{"type":"record","name":"Transacao","fields":[{"name":"id",
"type":"string"},{"name":"valor","type":"double"},{"name":"data","type":"string"},{"name":"tipo","type":"string"},{"name":"conta_origem","type":"string"},
{"name":"conta_destino","type":"string"},{"name":"descricao","type":"string"}]}}, Incompatibility{type:READER_FIELD_MISSING_DEFAULT_VALUE, location:/fields/3, 
message:tipo_conta, reader:{"type":"record","name":"DetalhesConta","fields":[{"name":"conta_id","type":"string"},{"name":"titular","type":"string"},
{"name":"saldo","type":"double"},{"name":"tipo_conta","type":"string"},{"name":"agencia","type":"string"}]}, writer:{"type":"record","name":"Transacao",
"fields":[{"name":"id","type":"string"},{"name":"valor","type":"double"},{"name":"data","type":"string"},{"name":"tipo","type":"string"},{"name":"conta_origem",
"type":"string"},{"name":"conta_destino","type":"string"},{"name":"descricao","type":"string"}]}}, Incompatibility{type:READER_FIELD_MISSING_DEFAULT_VALUE, 
location:/fields/4, message:agencia, reader:{"type":"record","name":"DetalhesConta","fields":[{"name":"conta_id","type":"string"},{"name":"titular",
"type":"string"},{"name":"saldo","type":"double"},{"name":"tipo_conta","type":"string"},{"name":"agencia","type":"string"}]}, writer:{"type":"record",
"name":"Transacao","fields":[{"name":"id","type":"string"},{"name":"valor","type":"double"},{"name":"data","type":"string"},{"name":"tipo","type":"string"},
{"name":"conta_origem","type":"string"},{"name":"conta_destino","type":"string"},{"name":"descricao","type":"string"}]}}]; error code: 409

A mensagem de erro é extensa, porém, resumidamente, o problema ocorreu ao tentar registrar automaticamente um novo schema, seguindo a estratégia de nome TopicNameStrategy que gera um subject chamado transacoes-value. Como este subject já foi registrado anteriormente, foi realizada uma tentativa de atualizar um schema tentado adicionar campos obrigatórios gerando erros de incompatibilidade. Esse erro está relacionado aos tipos de compatibilidade dos schemas, onde o padrão é BACKWARD. Esse padrão permite a exclusão de campos e a adição de campos opcionais. Mais informações sobre compatibilidade dos schemas podem ser encontradas na documentação sobre Compatibility types.

Dica: para finalizar o producer em qualquer um dos terminais abertos, basta digitar Ctrl + C

Solução

Para resolver esse problema devemos utilizar a estratégia de nome de subject TopicRecordNameStrategy ou RecordNameStrategy. Aqui vamos utilizar a estratégia TopicRecordNameStrategy, mas antes vamos recriar a nossa infraestrutura seguindo alguns passos:

Obs: esses passos não são obrigatórios, mas para mantermos o ambiente sem estados iremos recriar a infraestrutura

Desligue o container:



docker-compose down -v

Recrie a infraestrutura:



docker-compose up -d

Agora vamos registrar dois schemas para o mesmo tópico da seguinte forma:
Acesse o bash do container:



docker-compose exec fast-data-dev bash

Crie um producer para o schema Transacoes:



kafka-avro-console-producer \
--broker-list localhost:9092 \
--topic transacoes \
--property value.schema='{ "type": "record", "name": "Transacao", "fields": [ {"name": "id", "type": "string"}, {"name": "valor", "type": "double"}, {"name": "data", "type": "string"}, {"name": "tipo", "type": "string"}, {"name": "conta_origem", "type": "string"}, {"name": "conta_destino", "type": "string"}, {"name": "descricao", "type": "string"} ] }' \
--property schema.registry.url=http://localhost:8081 \
--property value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

Agora cole o seguinte evento:



{ "id": "123456789", "valor": 500.0, "data": "2023-01-01", "tipo": "transferencia", "conta_origem": "987654321", "conta_destino": "567890123", "descricao": "Transferência entre contas" }

Agora vamos criar um producer para o schema DetalhesConta, primeiro abrimos um novo terminal e inciamos um bash para o container:



docker-compose exec fast-data-dev bash

Agora iniciamos o nosso producer:



kafka-avro-console-producer \
--broker-list localhost:9092 \
--topic transacoes \
--property value.schema='{ "type": "record", "name": "DetalhesConta", "fields": [ {"name": "conta_id", "type": "string"}, {"name": "titular", "type": "string"}, {"name": "saldo", "type": "double"}, {"name": "tipo_conta", "type": "string"}, {"name": "agencia", "type": "string"} ] }' \
--property schema.registry.url=http://localhost:8081 \
--property value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

Logo após, colamos o seguinte evento para o schema DetalhesConta:



{ "conta_id": "987654321", "titular": "Joaquim", "saldo": 1000.0, "tipo_conta": "corrente", "agencia": "123" }

Com isso podemos validar os subjects na interface gráfica do landoop e pesquisar pela palavra transacoes:

Ou através do curl com o seguinte comando:



curl -X GET http://localhost:3030/schema-registry-ui/#/subjects

Ao analisar a resposta do comando acima podemos identificar que foram criados 2 subjects para o tópico transacoes: transacoes-Transacao e transacoes-DetalhesConta.

Também podemos visualizar os eventos recebidos no tópico Kafka usando a interface gráfica do landoop:

Ou através de um consumer na linha de comando, num outro terminal, após iniciar o bash com o comando docker-compose exec fast-data-dev bash:



kafka-avro-console-consumer \
--bootstrap-server localhost:9092 \
--topic transacoes \
--from-beginning \
--property schema.registry.url=http://localhost:8081

Ao finalizar o passo a passo será possível registrar dois schemas no mesmo tópico utilizando o kafka-avro-console-producer, de forma automática, através da estratégia de nome de subject TopicRecordNameStrategy para não sofrer com problemas de conflito de subjects no mesmo tópico. Você pode encontrar um exemplo dessa implantação em Multi schemas in one Kafka topic utilizando a linguagem Java ou outro exemplo em python

Forem: Adriano Avelino

📦 ASDF: Gerenciando versões de linguagens e ferramentas num lugar só

O que é o asdf?

Plugins e versões

Principais Comandos do asdf

Instalação do asdf

Guia Prático para Instalação de Plugins e Versões com o asdf

🚀 Começando com Node.js

☕ Configurando o Java

🔧 Configurando o Maven

🐍 Configurando o Python

(Opcional) Laboratório

Pré-requisitos:

Criando o ambiente

Construindo e executando o container

Links

Conector Kafka Lambda Sink com Localstack

Introdução

Pré-requisitos

Sobre o projeto

Criação dos arquivos de configuração

Docker Compose

CloudFormation

Configuração do conector Kafka

Download do plugin Lambda Sink

Estrutura final do projeto

Criação da infrastrutura

Inicialização dos containers

Criação da lambda

Criação do conector

Enviar mensagem no kafka

Opção 1: Envio rápido de mensagem única

Opção 2: Envio contínuo de mensagens

Como visualizar os logs da Lambda no terminal

Automatização

Erros e Soluções: Desvendando Problemas com Estilo

1. Logs: Seus Melhores Amigos

2. Erro Clássico: "Function not found"

3. Erros no Conector Kafka: Como Encontrar?

Exemplo de Erro: Lambda Não Encontrada

4. Interface Gráfica: Porque Nem Todo Mundo Gosta de Linha de Comando

Dicas e recomendações

Links

Multiplos schemas no mesmo tópico Kafka na linha de comando

Teoria

O problema

Solução

Referências