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

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