DEV Community: Vinicius Pereira de Oliveira

Opentelemetry + Typescript

Vinicius Pereira de Oliveira — Mon, 05 Feb 2024 10:54:00 +0000

Inserindo Observabilidade nos seus códigos Typescript

Nesse artigo irei apresentar funcionalidades básicas de observabilidade para inserir nos seus projetos em Nest Js, para que o gerenciamento da sua aplicação e tratamento de erros se torne algo cada vez mais simples no seu desenvolvimento e a correção de incidentes, mais rápida.

O que é o Opentelemetry

OpenTelemetry é uma ferramenta para observabilidade utilizada para criação e gerenciamento de traços, métricas e logs de uma aplicação. É uma funcionalidade open source com integração disponível a diversos sistemas de APM (Jaeger, Grafana, Dynatrace e Prometheus).

Jaeger

Jaeger é uma plataforma open-source para gerenciamento de trace. Por ser simples, utilizaremos para monitoramento da nossa aplicação em ambiente local.

1 - Organize seu docker compose

Para conseguir testar em ambiente de desenvolvimento, utilizaremos uma ferramenta para gerenciamento de traces open source. Jaeger

Crie um arquivo chamado docker-compose.yaml

e insira o seguinte texto:

version: '3.0'
services:
  jaeger:
    platform: linux/x86_64
    image: jaegertracing/all-in-one:1.53
    environment:
      - COLLECTOR_ZIPKIN_HTTP_PORT=9411
    ports:
      - "5775:5775/udp"
      - "6831:6831/udp"
      - "6832:6832/udp"
      - "5778:5778"
      - "4317:4317"
      - "4318:4318"
      - "16686:16686"
      - "14250:14250"
      - "14268:14268"
      - "14269:14269"
      - "9411:9411"

rode um

docker compose up -d

E o Jaeger já estará instrumentado.

Para testar se a instrumentação está correta abra a página
http://localhost:16686/

2 - Instrumentando os traces no Jaeger

As nossas métricas da aplicação, traces e logs de sistema serão instrumentados a partir de 'exporters' que são configurados a partir do SDK do NodeJS.

Para instrumentar o sdk do OpenTelemetry é necessário instalar a biblioteca necessária para implantação dessa classe

yarn add @opentelemetry/sdk-node

crie um arquivo para inicializar o sdk, inserindo o seguinte código:

import { NodeSDK } from "@opentelemetry/sdk-node";

const sdk = new NodeSDK({

})

sdk.start();

Após a criação do sdk, precisamos dizer ao OpenTelemetry para onde queremos exportar todos os logs que traços de sistema que ele gerar. Para facilitar esse trabalho, dentro do objeto de configuração do NodeSDK, podemos definir qual exporter iremos utilizar. O próprio OpenTelemetry Protocol (OTLP) nos fornece uma boa experiência e facilidade para exportar esses dados.

Há varios protocolos disponíveis para envio dessas informações aos coletores, por exemplo, JSON, protobuf ou gRPC.

Iremos utilizar utilizar o protocolo JSON, mas fique a vontade para utilizar o protocolo que quiser, todos irão funcionar bem.

Instale as dependencias abaixo para gerenciar os exporters

yarn add @opentelemetry/exporter-trace-otlp-http \
  @opentelemetry/exporter-metrics-otlp-http \
  @opentelemetry/sdk-metrics

Atualizando nossa instancia do SDK, ficaremos com o código da seguinte forma:

import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { NodeSDK } from "@opentelemetry/sdk-node";

const sdk = new NodeSDK({
    traceExporter: new OTLPTraceExporter({}),
    metricReader: new PeriodicExportingMetricReader({
        exporter: new OTLPMetricExporter({})
    })
})

sdk.start();

OBS: Dentro de cada coletor, podemos definir URL's de destino da aplicação. Exemplo, ao utilizar o Dynatrace, temos que definir uma URL de destino dos traces que criamos. Para realizar essa modificação. Dentro do objeto de OTLPTraceExporter e OTLPMetricExporter existe um campo, chamado url, onde definimos o destino da informação

Exemplo:

import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { NodeSDK } from "@opentelemetry/sdk-node";

const sdk = new NodeSDK({
    traceExporter: new OTLPTraceExporter({
        url: 'http://localhost:4318/v1/traces'
    }),
    metricReader: new PeriodicExportingMetricReader({
        exporter: new OTLPMetricExporter({
            url: 'http://localhost:4318/v1/metrics'
        })
    })
})

sdk.start();

Antes de finalizar a configuração do NodeSDK. Precisamos dizer ao NodeSDK do OpenTelemetry os dados da nossa aplicação, para conseguirmos identificar os traces de maneira mais rápida e como serão as instrumentações automáticas.

Para definir dados da aplicação para o OpenTelemetry, precisamos instalar a biblioteca @opentelemetry/resources e @opentelemetry/semantic-conventions

yarn add @opentelemetry/resources \
  @opentelemetry/semantic-conventions

Onde o código ficará da seguinte forma:

import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { Resource } from '@opentelemetry/resources';
import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({
    url: 'http://localhost:4318/v1/traces',
  }),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: 'http://localhost:4318/v1/metrics',
    }),
  }),
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAMESPACE]: 'NOME DO SERVICO',
    [SemanticResourceAttributes.SERVICE_VERSION]: 'VERSAO',
  }),
});

sdk.start();

Para finalizar, a instrumentação automática, é criada a partir da @opentelemetry/auto-instrumentations-node

Então

yarn add @opentelemetry/auto-instrumentations-node

E modificamos o código, deixando-o da seguinte forma:

import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { Resource } from '@opentelemetry/resources';
import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({
    url: 'http://localhost:4318/v1/traces',
  }),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: 'http://localhost:4318/v1/metrics',
    }),
  }),
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAMESPACE]: 'NOME',
    [SemanticResourceAttributes.SERVICE_VERSION]: '0.2',
    [SemanticResourceAttributes.SERVICE_NAME]: 'Nome Identificador da aplicação',
  }),
  instrumentations: [getNodeAutoInstrumentations()]
});

sdk.start();

3 - Criando Traces Manuais

Para criarmos os nossos traços na aplicação de forma manual, utilizaremos decorators que vão nos facilitar a não ficar repetindo código na aplicação inteira, para pegar todo o caminho de uma requisição ou lógica.

Criaremos um decorator, que por enquanto apenas executará o método sem modificá-lo em nada

export const Trace = (value?: string) => {
  return function (
    target: any,
    propertyKey: string,
    descriptor: PropertyDescriptor,
  ) {
    const method = descriptor.value

    const args = arguments;

    descriptor.value = function() {
      const that = this
      return method?.apply(that, args)
    }
  };
};

Para iniciarmos a criação dos tracers, é necessário instalar mais uma biblioteca chamada @opentelemetry/api

yarn add @opentelemetry/api

Modificamos o decorator, para ficar do seguinte formato:

import { Span, trace } from '@opentelemetry/api';

export const Trace = (value?: string) => {
  return function (
    target: any,
    propertyKey: string,
    descriptor: PropertyDescriptor,
  ) {
    const method = descriptor.value;

    const args = arguments;

    const tracer = trace.getTracer('');

    descriptor.value = function () {
      const that = this;
      return tracer.startActiveSpan(
        value ?? `${target.constructor.name}.${propertyKey}`,
        (span: Span) => {
          console.log(propertyKey)
          const result = method?.apply(that, args);
          span.end();
          return result
        }
      );
    };
  };
};

Dessa forma conseguiremos todos os traços dos métodos que inserirem esse decorator

Criaremos também uma função de traceError para conseguir ter mapeados todos os erros que acontecerem na aplicação.

import { Exception, SpanStatusCode, trace } from "@opentelemetry/api";

export function traceError(error: any) {
  const activeSpan = trace.getActiveSpan();

  activeSpan?.recordException(error as Exception);
  activeSpan?.setAttribute(
    "exception.response",
    JSON.stringify(
      {
        data: "Erro AXIOS",
      },
      null,
      2
    )
  );
  activeSpan?.setStatus({
    code: SpanStatusCode.ERROR,
    message: `Error: ${error.message}`,
  });
}

Simulando a AWS no seu ambiente Local

Vinicius Pereira de Oliveira — Thu, 17 Aug 2023 22:41:38 +0000

Opa, espero que esteja tudo bem com você que está lendo esse artigo no momento.
Passei por situações interessantes, no período em que desenvolvia uma feature na atual empresa em que me encontro.

Tive a necessidade de utilizar recursos da AWS, seja para aprender o que eu iria manusear, como implementar no ambiente corporativo de forma correta.

Durante meus estudos, para evitar gastos, procurei soluções que me permitiriam usar as funções de armazenamento e envio de e-mail da AWS de forma gratuita. Até que finalmente encontro um recurso, que simula o ambiente cloud da AWS dentro da sua máquina local.

O Localstack: https://localstack.cloud/, é um recurso que possibilita simular diversos recursos AWS (dynamoDB, s3, iam, cognito, ses), dentro da sua máquina, utilizando o docker.

Pré-requisitos

Docker & Docker-compose
Node (Minimo na versão 16)
Aws-cli instalada na máquina

Para ativar, basta criar um docker compose no seu projeto com os seguintes comandos:

version: "3.8"

services:
  localstack:
    container_name: "localstack"
    image: localstack/localstack:1.0.4
    ports:
      - "4566:4566"            # LocalStack Gateway
    environment:
      - DOCKER_HOST=unix:///var/run/docker.sock
      - DEFAULT_REGION=us-east-1
    volumes:
      - "/var/run/docker.sock:/var/run/docker.sock"

Rode o comando docker-compose up para inicializar o funcionamento do localstack

Para validar se está realmente funcionando, abra o seu navegador na seguinte url => http://localhost:4566/health

Iniciando um projeto Node
Não iremos nos preocupar com padrão de projetos, apenas queremos criar uma aplicação, para: criar um arquivo num bucket S3, remover o arquivo, receber a url assinada, enviar um e-mail pelo SES

crie uma pasta e abra no editor de texto da sua preferencia

Inicie o seu projeto node com o seguinte comando:

yarn init -y

Instalando as dependencias

yarn add aws-sdk@2.1438 axios@1.4

yarn add @types/node@20.5 @types/axios@0.14 localstack@1 ts-node-dev@2 typescript@5.1.6 -D

Inicializando o Typescript

yarn tsc --init

dentro do package.json, adicione o script para rodar o projeto

"scripts": {
    "dev": "ts-node-dev --respawn --transpile-only index.ts"
},

crie um arquivo chamado index.ts

O projeto terá a seguinte estrutura

Codificando o projeto
Dentro do index.ts, iremos realizar as configurações dos serviços S3 e SES

import AWS from 'aws-sdk'
import axios from 'axios'

const s3config = {
    s3ForcePathStyle: true
}

const sesConfig = {
    apiVersion: '2010-12-01'
}

const isLocal = true

if(isLocal) {
    AWS.config.update({
        credentials: {
            accessKeyId: 'test',
            secretAccessKey: 'test'
        },
        region: 'us-east-1'
    })

    const host = 'localhost'
    const awsEndpoint = new AWS.Endpoint(`http://${host}:4566`);

    //@ts-ignore
    s3config.endpoint = awsEndpoint
    //@ts-ignore
    sesConfig.endpoint = awsEndpoint
}

const s3 = new AWS.S3(s3config);
const ses = new AWS.SES(sesConfig)

const bucketName = 'local-data-files'

OBSERVAÇÕES: Dentro da constante isLocal, poderemos inserir esse valor em uma variável de ambiente, para que quando for enviado para ambientes de homologação e Produção, utilizemos os reais valores da AWS e não do Localstack

Para definir que o ambiente utilizado será o localstack, é necessário configurar as chaves da AWS e alterar o endpoint que as classes irão realizar as requisições

AWS.config.update({
        credentials: {
            accessKeyId: 'test',
            secretAccessKey: 'test'
        },
        region: 'us-east-1'
})

const host = 'localhost'
    const awsEndpoint = new AWS.Endpoint(`http://${host}:4566`);

Para realizar a codificação dos demais serviços, iremos criar duas classes

class StorageManagement {

}

class SendEmail {

}

Para as funções de gerenciamento no S3, adicionaremos os seguintes comandos:

class StorageManagement {
    public async execute() {
        const image = 'https://www.purina.co.uk/sites/default/files/2020-12/Dog_1098119012_Teaser.jpg'

        const buffer = await this.downloadImage(image);
        const upload = await this.upload(buffer);

        await this.remove(upload.Key);
    }

    private async downloadImage(imageUrl: string) {
        const response = await axios.get(imageUrl, {
            responseType: 'arraybuffer'
        })

        const buffer = Buffer.from(response.data, 'base64');
        return buffer
    }

    private async upload(buffer: Buffer) {
        const s3Upload = await s3.upload({
            Bucket: bucketName,
            Key: `FILE_${(new Date()).getTime()}`,
            Body: buffer
        }).promise();

        return s3Upload
    }

    private async remove(key: string) {
        await s3.deleteObject({
            Bucket: bucketName,
            Key: key
        })
    }
}

E para o envio de e-mail, temos o seguinte código:

class SendEmail {
    public async execute() {
        var params = {
            Destination: {
             ToAddresses: [
                'envio@teste.com', 
             ]
            }, 
            Message: {
             Body: {
              Html: {
               Charset: "UTF-8", 
               Data: "This message body contains HTML formatting. It can, for example, contain links like this one: <a class=\"ulink\" href=\"http://docs.aws.amazon.com/ses/latest/DeveloperGuide\" target=\"_blank\">Amazon SES Developer Guide</a>."
              }, 
              Text: {
               Charset: "UTF-8", 
               Data: "This is the message body in text format."
              }
             }, 
             Subject: {
              Charset: "UTF-8", 
              Data: "Test email"
             }
            },
            Source: "teste@mailinator.com", 
           };

        const emailSend = ses.sendEmail(params).promise()
    }
}

Antes de rodar nossa aplicação e ver o funcionamento, precisamos realizar alguns comandos via script para que o código funcione

crie um arquivo script.sh e adicione os seguintes comandos

# Nome do Bucket
BUCKET_NAME=local-data-files

# Comando AWS para criar um bucket no S3
# o Endpoint é para definir que iremos criar dentro do localstack
aws \
    s3 mb s3://$BUCKET_NAME \
    --endpoint http://localhost:4566

# Comando para validar o e-mail que iremos utilizar para teste no Localstack
aws ses verify-email-identity \
    --email-address teste@mailinator.com \
    --endpoint-url=http://localhost:4566

Para executar digite no terminal

sh script.sh

Testando
Para Testar cada uma das funcionalidades, instancie as classes e chame a função execute

const test = new StorageManagement()
const testSES = new SendEmail()

testSES.execute()

test.execute()

Validando
Para Validar se o arquivo está presente no s3, deve-se executar o seguinte comando sh

BUCKET_NAME=local-data-files
aws \
    s3 ls s3://$BUCKET_NAME\
    --endpoint http://localhost:4566

Para visualizar se o email foi enviado, entre na seguinte url

http://localhost:4566/_localstack/ses

Espero que esse artigo tenha sido útil para você. Deixe seu comentário e sugestão de próximo tema.

Até mais! ❤️

Repositório GitHub do artigo
https://github.com/DEVinicius/use-s3-ses-localstack

Minhas Redes Sociais

Instagram📗
LinkedIn💼

KAFKA CONNECT: Conectando em um banco de Dados Oracle

Vinicius Pereira de Oliveira — Sun, 27 Nov 2022 01:45:40 +0000

Clima natalino chegando, na verdade, mais perto do que em janeiro. Vamos apenas pensar em resolver problemas que um dia foram complicados para muitos, mas por meio do artigo certo, conseguimos resolver nossos empecilhos facilmente.

Já passou pelo problema de tentar se conectar com um banco oracle, para conseguir captar todos os eventos de insert ou update de uma tabela específica a partir do kafka-connect?

Caso tenha ou já passou por problemas semelhantes a esses, venho aqui lhe trazer boas novas sobre formas de conexão a partir do conector jdbc.

O connector jdbc, é Open source, presente na documentação oficial da confluent, com todas as descrições bem claras dos campos que são utilizados, e seus níveis de prioridades.
JDBC
Documentação do Connector

Vamos deixar de enrolação e partir logo pro abraço, porque estamos no fim de ano, e quanto mais rápido aprendermos essa funcionalidade, mais cedo comemoraremos a ceia de natal.

Antes de tudo, certifique-se que seu kafka-connect possui o seguinte connector instalado na sua máquina:

confluent-hub install confluentinc/kafka-connect-jdbc:latest

Confirmando a instalação, vamos para alguns casos em que o connector pode ser utilizado:

1 - Fazer Dump completo do banco de dados, ou de tabelas específicas;

2 - Buscar apenas dados que foram inseridos na tabela;

3 - Receber apenas os dados que sofreram updates;

4 - Receber mensagens de itens que foram inseridos ou atualizados na tabela;

Esses 4 cenários conseguimos tratar com o kafka-connect, cada um com sua forma de configurar um connector.

1 - Dump completo do banco, ou de tabelas específicas

Esse processo requer que configuremos o nosso connector da seguinte forma:

Obs: Estarei especificando o significado de cada campo do objeto

{
    "name": "nome do connector",
     "config": {
       "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
       "connection.url": "jdbc:oracle:thin:@<host do banco>:<porta>/<nome do banco de dados>",
       "connection.user": "<username do banco>",
       "connection.password": "<senha do banco>",
       "topic.prefix": "<prefixo do tópico kafka que será produzido>",
       "table.whitelist": "<tabelas que serão lidas>",
       "poll.interval.ms": 5000,
      "value.converter":"org.apache.kafka.connect.json.JsonConverter",
       "mode": "bulk"
   }
}

Para a criação de um connector jdbc, definimos dois campos primários (name e config). "name" se baseia no nome do nosso connector e "config" são as definições do comportamento desse connector.

Config

connector.class => Classe que tem as funções do nosso connector, dentro do pacote que instalamos anteriormente
connection.url => URL de conexão com o banco dados, devendo seguir a seguinte sintaxe do comando jdbc -> jdbc:oracle:thin:@:/
connection.user => usuário do banco de dados
connection.password => senha para acessar ao banco
topic.prefix => Inicio do nome do nosso tópico. Geralmente ao criarmos um connector, o nosso tópico vem escrito com:
table.whitelist => Tabelas que serão mapeadas no processo. Caso o número de tabelas seja muito grande, podemos definir quais tabelas não serão lidas, a partir do parâmetro "table.blacklist". Não obstante, deve existir apenas um desses campos, ou utilizamos a "whitelist" ou "blacklist"
poll.interval.ms => Intervalo em milisegundos que o nosso connector fará o dump das tabelas
value.converter => Definindo para o formato que os nossos dados serão convertidos. Com essa informação no campo: org.apache.kafka.connect.json.JsonConverter. Sempre iremos transformar nossas mensagens em formato json;
mode => Modo que será feito o mapeamento. O modo bulk se refere a produção dos dados completos da(s) tabela(s), partir de x tempo (definido pelo poll.interval.ms)

2 - Buscar apenas dados que foram inseridos na tabela

Configuração do connector, para que apenas eventos de inserts sejam capturados pelo connector

Obs: Os campos repetidos, não serão replicados na descrição

{
    "name": "nome do connector",
     "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url": "jdbc:oracle:thin:@<host do banco>:<porta>/<nome do banco de dados>",
        "connection.user": "<username do banco>",
        "connection.password": "<senha do banco>",
        "topic.prefix": "<prefixo do tópico kafka que será produzido>",
        "value.converter":"org.apache.kafka.connect.json.JsonConverter",
        "mode":"incrementing",
        "query": "<QUERY SELECT PARA BUSCAR OS DADOS ESPECIFICOS DA TABELA>",
        "numeric.mapping":"best_fit",
        "incrementing.column.name": "<NOME DO CAMPO DA TABELA, QUE OBEDEÇA VALOR DE CHAVE PRIMARIA>",
        "validate.non.null": true
   }
}

Config

mode => Incrementing, tem o significado que apenas será executado, a partir do momento que a chave primária (Obrigatoriamente sendo de um tipo número), ao se produzir um novo dado, será usada como parâmetro para verificar se é um novo dado.
query => Select que será executado, quando essa nova mensagem for gerada. Nesse momento, esse select será replicado em forma de mensagem, então o select contendo os dados corretos e úteis que o seu consumidor fará a leitura, estará presente nesse comando
incrementing.column.name => Campo que o connector terá como parâmetro, para validar se a informação é nova na tabela;
numeric.mapping => Validação se o campo definido referência para o connector é do tipo primitivo correto. O valor "best_fit", significa que será levado em conta vários tipos como parâmetro, desde INT até DECIMAL.
validate.non.null => Parâmetro booleano, que exigirá ou não que o campo utilizado como referência seja não nulo

3 - Buscar apenas dados que foram atualizados na tabela

Tem um corpo semelhante ao item anterior, apenas modificando alguns campos

{
    "name": "nome do connector",
     "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url": "jdbc:oracle:thin:@<host do banco>:<porta>/<nome do banco de dados>",
        "connection.user": "<username do banco>",
        "connection.password": "<senha do banco>",
        "topic.prefix": "<prefixo do tópico kafka que será produzido>",
        "value.converter":"org.apache.kafka.connect.json.JsonConverter",
        "mode":"timestamp",
        "query": "<QUERY SELECT PARA BUSCAR OS DADOS ESPECIFICOS DA TABELA>",
        "timestamp.column.name": "<NOME DO CAMPO DA TABELA, QUE OBEDEÇA VALOR DE TIMESTAMP>",
        "validate.non.null": false
   }
}

Config

mode: Timestamp, para apenas captar itens de atualização
timestamp.column.name => Coluna da tabela, que será utilizada como referência pelo connector. Esse campo deve ser do tipo datetime para que o connector funcione

Os demais campos se repetem na lógica

4 - Receber mensagens de itens que foram inseridos ou atualizados na tabela

Se baseia na junção dos dois itens anteriores. Tendo o seguinte corpo do objeto

{
    "name": "nome do connector",
     "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url": "jdbc:oracle:thin:@<host do banco>:<porta>/<nome do banco de dados>",
        "connection.user": "<username do banco>",
        "connection.password": "<senha do banco>",
        "topic.prefix": "<prefixo do tópico kafka que será produzido>",
        "value.converter":"org.apache.kafka.connect.json.JsonConverter",
        "mode":"timestamp+incrementing",
        "query": "SELECT CAST(p.PERSON_ID AS NUMERIC(8,0)) AS PERSON_ID, p.CREATED_AT, p.UPDATED_AT FROM PERSONS p",
        "numeric.mapping":"best_fit",
        "incrementing.column.name": "PERSON_ID",
        "timestamp.column.name": "UPDATED_AT",
        "validate.non.null": false
    }
}

Config

mode => timestamp+incrementing Deve seguir esse nome para que funcione no connector.

Agradecimentos

Espero que esse artigo seja útil para você, facilitando seu trabalho, evitando horas de pesquisa para um problema que já esteve no dia-a-dia de outros.

Muito Obrigado! ❤️

Minhas Redes Sociais

Instagram📗
LinkedIn💼