Railander Marques for AWS Community Builders

Posted on Nov 12

CDC do SQL Server para o Amazon MSK com Debezium no MSK Connect: IAM, Secrets Manager e alerta de conexão

#aws #kafka #debezium #sqlserver

1. Objetivo

Capturar as alterações (INSERT, UPDATE, DELETE) de tabelas de um SQL Server e publicar cada evento em tópicos do Amazon MSK, usando o conector Debezium rodando no MSK Connect, com autenticação IAM no cluster, senha do banco no Secrets Manager e alerta por e-mail quando o conector perder conexão com o banco.

Resultado esperado: cada tabela vira um tópico <PREFIXO>.<DATABASE>.dbo.<TABELA>, o snapshot inicial é carregado sozinho e, a partir dali, só as mudanças trafegam. Se o banco cair ou a rede oscilar, alguém recebe um e-mail dizendo qual conector falhou e qual foi o erro.

Esse é o tipo de pipeline que parece simples no diagrama e dá trabalho na hora de subir. A maior parte das falhas que eu vejo não é no Debezium em si, é em três lugares: CDC mal habilitado no banco, autenticação IAM esquecida no tópico de histórico de schema, e security group.

2. Arquitetura

+-------------------+        1433/tcp        +----------------------+
|   SQL Server      | <--------------------- |   MSK Connect        |
|   (RDS ou EC2)    |                        |   Debezium SQL Server|
|   CDC habilitado  |                        |   (worker na VPC)    |
+-------------------+                        +----------+-----------+
                                                        |
                                          9098/tcp      | SASL_SSL + AWS_MSK_IAM
                                                        v
                                             +----------------------+
                                             |   Amazon MSK         |
                                             |   topicos por tabela |
                                             |   + schema history   |
                                             +----------------------+

       logs do worker
MSK Connect -------> CloudWatch Logs --(subscription filter)--> Lambda --> SES (e-mail)

Componentes:

Componente	Função
SQL Server com CDC	Gera as tabelas de change data no schema `cdc`
Custom plugin	Zip com o Debezium, o config provider do Secrets Manager e a lib de IAM do MSK
Worker configuration	Converters e config provider
Connector	Configuração do Debezium apontando para o banco e para o MSK
Lambda + SES	Alerta quando aparece erro de conexão no log

3. Pré-requisitos

[ ] Cluster MSK (provisionado ou serverless) com IAM access control habilitado (porta 9098)
[ ] SQL Server 2016 SP1 ou superior, edição que suporte CDC (Standard ou Enterprise)
[ ] SQL Server Agent rodando (no RDS ele já fica ativo)
[ ] Bucket S3 para o custom plugin
[ ] Subnets privadas com rota até o banco e até o MSK
[ ] Identidade verificada no SES para o remetente do alerta
[ ] AWS CLI v2 configurado com permissão para kafkaconnect, iam, secretsmanager, logs e lambda

Variáveis usadas no tutorial (ajuste e exporte antes de começar):

export AWS_REGION="us-east-1"
export ACCOUNT_ID="<ACCOUNT_ID>"
export MSK_CLUSTER_ARN="arn:aws:kafka:us-east-1:<ACCOUNT_ID>:cluster/<CLUSTER_NAME>/<CLUSTER_UUID>"
export BOOTSTRAP_IAM="b-1.<cluster>.kafka.us-east-1.amazonaws.com:9098,b-2.<cluster>.kafka.us-east-1.amazonaws.com:9098"
export PLUGIN_BUCKET="<BUCKET_PLUGINS>"
export SECRET_NAME="<SECRET_SQLSERVER>"
export CONNECTOR_NAME="prd-cdc-<sistema>"
export TOPIC_PREFIX="<PREFIXO>"
export DATABASE="<DATABASE>"
export SUBNETS="subnet-aaaa,subnet-bbbb"
export SG_CONNECT="sg-<CONNECT>"

4. Passo 1: Habilitar CDC no SQL Server

4.1 Habilitar no banco

RDS for SQL Server (não tem sysadmin, então usa a procedure da AWS):

EXEC msdb.dbo.rds_cdc_enable_db '<DATABASE>';

SQL Server em EC2 ou on-premises:

USE <DATABASE>;
EXEC sys.sp_cdc_enable_db;

Confira:

SELECT name, is_cdc_enabled FROM sys.databases WHERE name = '<DATABASE>';

4.2 Habilitar por tabela

Crie uma role de acesso (gating role) para o Debezium e habilite cada tabela:

USE <DATABASE>;
CREATE ROLE cdc_reader;

EXEC sys.sp_cdc_enable_table
    @source_schema        = N'dbo',
    @source_name          = N'pedido',
    @role_name            = N'cdc_reader',
    @supports_net_changes = 0;

EXEC sys.sp_cdc_enable_table
    @source_schema        = N'dbo',
    @source_name          = N'cliente',
    @role_name            = N'cdc_reader',
    @supports_net_changes = 0;

Valide que as capture instances existem e que os jobs foram criados:

EXEC sys.sp_cdc_help_change_data_capture;
EXEC sys.sp_cdc_help_jobs;

Atenção à retenção do CDC. O job de cleanup apaga as alterações depois de 3 dias por padrão (retention = 4320 minutos). Se o conector ficar parado mais tempo que isso, ele perde eventos e você vai precisar refazer o snapshot. Em ambiente onde o conector pode ficar fora no fim de semana, aumente:
EXEC sys.sp_cdc_change_job @job_type = N'cleanup', @retention = 10080; -- 7 dias

4.3 Usuário do Debezium

USE master;
CREATE LOGIN debezium WITH PASSWORD = '<SENHA_FORTE>';

USE <DATABASE>;
CREATE USER debezium FOR LOGIN debezium;
ALTER ROLE cdc_reader ADD MEMBER debezium;
ALTER ROLE db_datareader ADD MEMBER debezium;
GRANT VIEW DATABASE STATE TO debezium;

O db_datareader é para o snapshot inicial. Se quiser restringir mais, troque por GRANT SELECT só nas tabelas capturadas e no schema cdc.

Tabela alterada depois do CDC habilitado. Se alguém fizer ALTER TABLE ADD COLUMN, a capture instance antiga não enxerga a coluna nova. É preciso criar uma segunda capture instance para a tabela (o SQL Server permite duas) e depois desabilitar a antiga. O Debezium troca de uma para a outra sozinho. Combine isso com quem faz deploy de banco, senão a coluna simplesmente não chega no tópico e ninguém percebe.

5. Passo 2: Secret do banco

aws secretsmanager create-secret \
  --name "$SECRET_NAME" \
  --secret-string '{"username":"debezium","password":"<SENHA_FORTE>"}'

A senha nunca vai aparecer em texto claro na configuração do conector. O MSK Connect resolve ${secretsmanager:...} em tempo de execução.

6. Passo 3: Montar o custom plugin

O plugin é um zip único com tudo que o worker precisa carregar. Monte numa máquina com acesso à internet:

DBZ_VERSION="2.7.3.Final"
IAM_AUTH_VERSION="2.2.0"

mkdir -p plugin && cd plugin

# Debezium SQL Server
curl -sLO "https://repo1.maven.org/maven2/io/debezium/debezium-connector-sqlserver/${DBZ_VERSION}/debezium-connector-sqlserver-${DBZ_VERSION}-plugin.tar.gz"
tar xzf "debezium-connector-sqlserver-${DBZ_VERSION}-plugin.tar.gz"
rm -f ./*.tar.gz

# Lib de autenticacao IAM do MSK (usada pelo producer/consumer do schema history)
curl -sLO "https://github.com/aws/aws-msk-iam-auth/releases/download/v${IAM_AUTH_VERSION}/aws-msk-iam-auth-${IAM_AUTH_VERSION}-all.jar"
mv aws-msk-iam-auth-*-all.jar debezium-connector-sqlserver/

# Config provider do Secrets Manager (baixe o zip "with-dependencies" da pagina de releases)
# https://github.com/aws-samples/msk-config-providers/releases
unzip -q msk-config-providers-*-with-dependencies.zip -d debezium-connector-sqlserver/

cd .. && (cd plugin && zip -qr ../debezium-sqlserver-plugin.zip .)
aws s3 cp debezium-sqlserver-plugin.zip "s3://${PLUGIN_BUCKET}/plugins/debezium-sqlserver-plugin.zip"

Crie o plugin:

aws kafkaconnect create-custom-plugin \
  --name "debezium-sqlserver-${DBZ_VERSION//./-}" \
  --content-type ZIP \
  --location "s3Location={bucketArn=arn:aws:s3:::${PLUGIN_BUCKET},fileKey=plugins/debezium-sqlserver-plugin.zip}"

Aguarde o estado ACTIVE:

aws kafkaconnect list-custom-plugins \
  --query "customPlugins[].{nome:name,estado:customPluginState,arn:customPluginArn}" --output table

Versão do Kafka Connect. O Debezium 2.x precisa de Java 11 ou superior. Crie o conector com kafkaConnectVersion 3.7.x. Com a 2.7.1 você ganha erros de classe que não dizem nada com nada.

7. Passo 4: Role de execução do conector

Trust policy (trust-kafkaconnect.json):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "Service": "kafkaconnect.amazonaws.com" },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": { "aws:SourceAccount": "<ACCOUNT_ID>" }
      }
    }
  ]
}

Permissões (policy-connector.json). Repare que tópico e grupo têm ARN próprio, com cluster/ trocado por topic/ e group/:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Cluster",
      "Effect": "Allow",
      "Action": ["kafka-cluster:Connect", "kafka-cluster:DescribeCluster"],
      "Resource": "arn:aws:kafka:us-east-1:<ACCOUNT_ID>:cluster/<CLUSTER_NAME>/<CLUSTER_UUID>"
    },
    {
      "Sid": "Topicos",
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:CreateTopic",
        "kafka-cluster:DescribeTopic",
        "kafka-cluster:WriteData",
        "kafka-cluster:ReadData"
      ],
      "Resource": "arn:aws:kafka:us-east-1:<ACCOUNT_ID>:topic/<CLUSTER_NAME>/<CLUSTER_UUID>/*"
    },
    {
      "Sid": "Grupos",
      "Effect": "Allow",
      "Action": ["kafka-cluster:AlterGroup", "kafka-cluster:DescribeGroup"],
      "Resource": "arn:aws:kafka:us-east-1:<ACCOUNT_ID>:group/<CLUSTER_NAME>/<CLUSTER_UUID>/*"
    },
    {
      "Sid": "SecretBanco",
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "arn:aws:secretsmanager:us-east-1:<ACCOUNT_ID>:secret:<SECRET_SQLSERVER>-*"
    }
  ]
}

aws iam create-role --role-name msk-connect-debezium \
  --assume-role-policy-document file://trust-kafkaconnect.json

aws iam put-role-policy --role-name msk-connect-debezium \
  --policy-name debezium-msk \
  --policy-document file://policy-connector.json

Se quiser travar ainda mais, troque o /* dos tópicos por /<PREFIXO>.* e /schema-history.*.

8. Passo 5: Security groups

Origem	Destino	Porta	Motivo
`sg-CONNECT`	SG do SQL Server	1433/tcp	Debezium lendo o banco
`sg-CONNECT`	SG do MSK	9098/tcp	Producer IAM
`sg-CONNECT`	VPC endpoint Secrets Manager (ou NAT)	443/tcp	Resolver o secret

O worker do MSK Connect roda dentro das suas subnets. Se as subnets não têm NAT, crie um VPC endpoint de interface para o Secrets Manager. Sem isso o conector fica em CREATING por um bom tempo e depois falha com timeout no config provider.

aws ec2 authorize-security-group-ingress --group-id <SG_SQLSERVER> \
  --protocol tcp --port 1433 --source-group "$SG_CONNECT"

aws ec2 authorize-security-group-ingress --group-id <SG_MSK> \
  --protocol tcp --port 9098 --source-group "$SG_CONNECT"

9. Passo 6: Worker configuration

cat > worker.properties <<'EOF'
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
config.providers=secretsmanager
config.providers.secretsmanager.class=com.amazonaws.kafka.config.providers.SecretsManagerConfigProvider
config.providers.secretsmanager.param.region=us-east-1
EOF

aws kafkaconnect create-worker-configuration \
  --name "debezium-json-secrets" \
  --properties-file-content "$(base64 -w0 worker.properties)"

10. Passo 7: Criar o conector

Configuração do Debezium (connector.json). O bloco do schema.history.internal.* é o ponto onde 90% das primeiras tentativas quebram:

{
  "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
  "tasks.max": "1",

  "database.hostname": "<SQLSERVER_HOST>",
  "database.port": "1433",
  "database.user": "${secretsmanager:<SECRET_SQLSERVER>:username}",
  "database.password": "${secretsmanager:<SECRET_SQLSERVER>:password}",
  "database.names": "<DATABASE>",
  "database.encrypt": "true",
  "database.trustServerCertificate": "true",

  "topic.prefix": "<PREFIXO>",
  "table.include.list": "dbo.pedido,dbo.cliente",
  "snapshot.mode": "initial",

  "topic.creation.default.replication.factor": "3",
  "topic.creation.default.partitions": "3",

  "schema.history.internal.kafka.bootstrap.servers": "<BOOTSTRAP_IAM>",
  "schema.history.internal.kafka.topic": "schema-history.<PREFIXO>",

  "schema.history.internal.producer.security.protocol": "SASL_SSL",
  "schema.history.internal.producer.sasl.mechanism": "AWS_MSK_IAM",
  "schema.history.internal.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
  "schema.history.internal.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",

  "schema.history.internal.consumer.security.protocol": "SASL_SSL",
  "schema.history.internal.consumer.sasl.mechanism": "AWS_MSK_IAM",
  "schema.history.internal.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
  "schema.history.internal.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
}

Por que isso importa: o MSK Connect configura IAM para o producer principal do worker, mas o Debezium abre outro producer e outro consumer só para o tópico de histórico de schema. Esses dois não herdam nada. Sem o bloco acima o conector sobe, tenta falar com o broker em 9098 sem SASL e fica preso em TimeoutException na criação do tópico de histórico.

Sobre database.encrypt: o driver JDBC da Microsoft a partir da versão 10 liga criptografia por padrão. Se o SQL Server usa certificado autoassinado (o padrão em EC2 e on-prem), sem trustServerCertificate=true o erro é PKIX path building failed. Em produção, o ideal é importar a CA do banco no truststore, mas isso dá outro tutorial.

Crie o conector:

PLUGIN_ARN=$(aws kafkaconnect list-custom-plugins \
  --query "customPlugins[?starts_with(name,'debezium-sqlserver')].customPluginArn | [0]" --output text)

WORKER_ARN=$(aws kafkaconnect list-worker-configurations \
  --query "workerConfigurations[?name=='debezium-json-secrets'].workerConfigurationArn | [0]" --output text)

ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/msk-connect-debezium"

aws logs create-log-group --log-group-name "/msk/connect/${CONNECTOR_NAME}"
aws logs put-retention-policy --log-group-name "/msk/connect/${CONNECTOR_NAME}" --retention-in-days 14

aws kafkaconnect create-connector \
  --connector-name "$CONNECTOR_NAME" \
  --kafka-connect-version "3.7.x" \
  --service-execution-role-arn "$ROLE_ARN" \
  --plugins "customPlugin={customPluginArn=${PLUGIN_ARN},revision=1}" \
  --worker-configuration "workerConfigurationArn=${WORKER_ARN},revision=1" \
  --capacity "provisionedCapacity={mcuCount=1,workerCount=1}" \
  --kafka-cluster "apacheKafkaCluster={bootstrapServers=${BOOTSTRAP_IAM},vpc={subnets=[${SUBNETS}],securityGroups=[${SG_CONNECT}]}}" \
  --kafka-cluster-client-authentication "authenticationType=IAM" \
  --kafka-cluster-encryption-in-transit "encryptionType=TLS" \
  --log-delivery "workerLogDelivery={cloudWatchLogs={enabled=true,logGroup=/msk/connect/${CONNECTOR_NAME}}}" \
  --connector-configuration file://connector.json

Um conector Debezium SQL Server por banco com tasks.max=1 resolve a maioria dos casos. Mais de uma task só faz diferença quando database.names tem vários bancos.

Mais de um conector no mesmo log group. Funciona, mas o alerta do passo 9 precisa descobrir qual conector gerou a linha. Prefira um log group por conector. Se herdar um ambiente em que todos escrevem no mesmo, o nome do conector aparece no contexto da linha de log ([<nome-conector>|task-0]) e dá para extrair com regex.

11. Passo 8: Validar

Estado do conector:

aws kafkaconnect list-connectors \
  --query "connectors[?connectorName=='${CONNECTOR_NAME}'].{estado:connectorState,arn:connectorArn}" --output table

Acompanhe o snapshot pelo log:

aws logs tail "/msk/connect/${CONNECTOR_NAME}" --follow \
  --filter-pattern '?"Snapshot" ?"snapshot" ?ERROR'

As linhas que importam são Snapshot step 7 - Snapshotting data e depois Snapshot completed. A partir dali ele entra em streaming.

Consumir o tópico a partir de uma EC2 na mesma VPC (com o jar aws-msk-iam-auth em kafka/libs e uma role com ReadData):

cat > client.properties <<'EOF'
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
EOF

./bin/kafka-topics.sh --bootstrap-server "$BOOTSTRAP_IAM" \
  --command-config client.properties --list | grep "$TOPIC_PREFIX"

./bin/kafka-console-consumer.sh --bootstrap-server "$BOOTSTRAP_IAM" \
  --consumer.config client.properties \
  --topic "${TOPIC_PREFIX}.${DATABASE}.dbo.pedido" --from-beginning --max-messages 5

Faça um UPDATE de teste no banco e confira o evento com "op":"u" e os blocos before e after.

12. Passo 9: Alerta de erro de conexão com o banco

Os erros que aparecem na prática quando o banco some ou a rede oscila:

SQLServerException: Connection timed out
SQLServerException: Connection reset
SQLServerException: The connection is closed
Login failed for user (senha trocada sem atualizar o secret)

O Debezium tenta reconectar sozinho, mas se a task morrer ela fica FAILED e o MSK Connect não reinicia. Então o alerta precisa existir.

12.1 Lambda

lambda_function.py:

import base64
import gzip
import json
import os
import re

import boto3

ses = boto3.client("ses", region_name=os.environ.get("SES_REGION", "us-east-1"))

REMETENTE = os.environ["REMETENTE"]
DESTINATARIOS = [d.strip() for d in os.environ["DESTINATARIOS"].split(",") if d.strip()]
PADROES = [
    "Connection timed out",
    "Connection reset",
    "The connection is closed",
    "Login failed for user",
]
RE_CONECTOR = re.compile(r"\[([A-Za-z0-9._-]+)\|task-\d+\]")


def descobrir_conector(mensagem, log_group, log_stream):
    achou = RE_CONECTOR.search(mensagem)
    if achou:
        return achou.group(1)
    return f"{log_group} ({log_stream})"


def lambda_handler(event, context):
    dados = json.loads(gzip.decompress(base64.b64decode(event["awslogs"]["data"])))
    if dados.get("messageType") != "DATA_MESSAGE":
        return

    falhas = {}
    for ev in dados["logEvents"]:
        msg = ev["message"]
        if not any(p in msg for p in PADROES):
            continue
        conector = descobrir_conector(msg, dados["logGroup"], dados["logStream"])
        falhas.setdefault(conector, []).append(msg.strip()[:1500])

    for conector, linhas in falhas.items():
        corpo = (
            f"O conector {conector} registrou erro de conexao com o banco.\n\n"
            f"Ocorrencias neste lote: {len(linhas)}\n\n"
            "Primeira ocorrencia:\n"
            f"{linhas[0]}\n\n"
            "Verifique o estado do conector no MSK Connect e a conectividade com o SQL Server.\n"
        )
        ses.send_email(
            Source=REMETENTE,
            Destination={"ToAddresses": DESTINATARIOS},
            Message={
                "Subject": {"Data": f"[CDC] Falha de conexao: {conector}"},
                "Body": {"Text": {"Data": corpo}},
            },
        )

Variáveis de ambiente:

Variável	Exemplo
`REMETENTE`	`alertas@<dominio-verificado>`
`DESTINATARIOS`	`time-dados@<empresa>,plantao@<empresa>`
`SES_REGION`	`us-east-1`

Destinatários mudam. Deixe a lista só na variável de ambiente e não no código, e não deixe o seu script de deploy sobrescrever as variáveis. Quem opera o ambiente costuma ajustar direto no console e não vai gostar de ver a lista voltar ao valor antigo no próximo deploy.

A role da Lambda precisa de ses:SendEmail na identidade do remetente e da policy gerenciada AWSLambdaBasicExecutionRole.

12.2 Subscription filter

LAMBDA_ARN="arn:aws:lambda:us-east-1:${ACCOUNT_ID}:function:cdc-alerta-conexao"

aws lambda add-permission \
  --function-name cdc-alerta-conexao \
  --statement-id cwlogs-${CONNECTOR_NAME} \
  --principal logs.amazonaws.com \
  --action lambda:InvokeFunction \
  --source-arn "arn:aws:logs:us-east-1:${ACCOUNT_ID}:log-group:/msk/connect/${CONNECTOR_NAME}:*"

aws logs put-subscription-filter \
  --log-group-name "/msk/connect/${CONNECTOR_NAME}" \
  --filter-name "erro-conexao-banco" \
  --filter-pattern '?"Connection timed out" ?"Connection reset" ?"The connection is closed" ?"Login failed for user"' \
  --destination-arn "$LAMBDA_ARN"

O filtro já corta o volume no CloudWatch, então a Lambda só é invocada quando tem erro. A checagem dentro do código é redundante de propósito: se alguém afrouxar o filtro depois, o e-mail continua só saindo para erro de conexão.

12.3 Testar sem derrubar o banco

Remova temporariamente a regra de 1433 do SG do banco:

aws ec2 revoke-security-group-ingress --group-id <SG_SQLSERVER> \
  --protocol tcp --port 1433 --source-group "$SG_CONNECT"

Em alguns minutos aparece o Connection timed out no log e o e-mail chega. Recoloque a regra e confirme que o conector voltou (se a task tiver ido para FAILED, será preciso recriar ou atualizar o conector para ela subir de novo).

Evite spam de alerta. Um banco fora por 30 minutos gera dezenas de linhas. Se isso incomodar, guarde o horário do último envio por conector numa tabela DynamoDB com TTL e só envie de novo depois de N minutos.

13. Troubleshooting

Sintoma	Causa provável	Ação
Conector fica em `CREATING` e falha depois de ~20 min	Worker sem saída para o Secrets Manager ou sem rota para o MSK	VPC endpoint do Secrets Manager, SG do MSK em 9098
`TimeoutException` ao criar `schema-history.*`	Producer/consumer do schema history sem SASL IAM	Bloco `schema.history.internal.*` do passo 7
`PKIX path building failed`	Driver JDBC com encrypt ligado e certificado autoassinado	`database.trustServerCertificate=true` ou CA no truststore
Snapshot termina e nenhum evento novo chega	SQL Server Agent parado ou jobs de capture desabilitados	`EXEC sys.sp_cdc_help_jobs` e status do Agent
`TopicAuthorizationException`	ARN de tópico errado na policy	Conferir `topic/<cluster>/<uuid>/*`
Coluna nova não aparece no evento	Capture instance antiga	Criar nova capture instance para a tabela
Conector parado por dias e voltou com erro de LSN	Cleanup do CDC apagou o histórico	Aumentar retenção e refazer snapshot

14. Referências

Debezium connector for SQL Server: https://debezium.io/documentation/reference/stable/connectors/sqlserver.html
MSK Connect, custom plugins: https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html
MSK IAM access control: https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html
Config providers para MSK Connect: https://github.com/aws-samples/msk-config-providers
CDC no RDS for SQL Server: https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appendix.SQLServer.CommonDBATasks.CDC.html

DEV Community