# Usar o Lambda com o Amazon MSK
O [Amazon Managed Streaming for Apache Kafka (Amazon MSK)](https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html) é um serviço totalmente gerenciado que você pode usar para criar e executar aplicações que usam o Apache Kafka para processar dados de streaming. O Amazon MSK simplifica a configuração, o dimensionamento e o gerenciamento de clusters do Kafka. O Amazon MSK também facilita a configuração da aplicação para várias zonas de disponibilidade e para segurança com o AWS Identity and Access Management (IAM).
Este capítulo explica como usar um cluster do Amazon MSK como origem de eventos para a função do Lambda. O processo geral de integração do Amazon MSK com o Lambda envolve as seguintes etapas:
1. **[Configuração do cluster e da rede](with-msk-cluster-network.md)**: primeiro, configure o [cluster do Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html). Isso inclui a configuração da rede correta para permitir que o Lambda acesse o cluster.
1. **[Configuração do mapeamento da origem do evento](with-msk-configure.md)**: depois, crie o recurso de [mapeamento da origem do evento](invocation-eventsourcemapping.md) de que o Lambda precisa para se conectar ao cluster Amazon MSK à função com segurança.
1. **[Configuração da função e das permissões](with-msk-permissions.md)**: por fim, certifique-se de que a função esteja configurada corretamente e tenha as permissões exigidas em seu [perfil de execução](lambda-intro-execution-role.md).
**nota**
Agora é possível criar e gerenciar seus mapeamentos da origem do evento do Amazon MSK diretamente do console do Lambda ou do Amazon MSK. Ambos os consoles oferecem a opção de lidar automaticamente com a configuração das permissões necessárias do perfil de execução do Lambda para um processo de configuração mais simplificado.
Para obter exemplos de como configurar uma integração do Lambda com um cluster do Amazon MSK, consulte [Tutorial: Usar um mapeamento de origem do evento do Amazon MSK para invocar uma função do Lambda](services-msk-tutorial.md), [Using Amazon MSK as an event source for AWS Lambda](https://aws.amazon.com/blogs/compute/using-amazon-msk-as-an-event-source-for-aws-lambda/) no AWS Compute Blog e [Amazon MSK Lambda Integration](https://amazonmsk-labs.workshop.aws/en/msklambda.html) no Amazon MSK Labs.
**Topics**
+ [Evento de exemplo](#msk-sample-event)
+ [Configurar o cluster do Amazon MSK e a rede do Amazon VPC para o Lambda](with-msk-cluster-network.md)
+ [Configuração de permissões do Lambda para mapeamentos da origem do evento do Amazon MSK](with-msk-permissions.md)
+ [Configurar as origens de eventos do Amazon MSK para o Lambda](with-msk-configure.md)
+ [Tutorial: Usar um mapeamento de origem do evento do Amazon MSK para invocar uma função do Lambda](services-msk-tutorial.md)
## Evento de exemplo
O Lambda envia o lote de mensagens no parâmetro de evento quando ele chama sua função. O payload do evento contém uma matriz de mensagens. Cada item de array contém detalhes do tópico do Amazon MSK e do identificador de partição, juntamente com um carimbo de data/hora e uma mensagem codificada em base64.
```
{
"eventSource":"aws:kafka",
"eventSourceArn":"arn:aws:kafka:us-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
"bootstrapServers":"b-2.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092,b-1.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092",
"records":{
"mytopic-0":[
{
"topic":"mytopic",
"partition":0,
"offset":15,
"timestamp":1545084650987,
"timestampType":"CREATE_TIME",
"key":"abcDEFghiJKLmnoPQRstuVWXyz1234==",
"value":"SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
"headers":[
{
"headerKey":[
104,
101,
97,
100,
101,
114,
86,
97,
108,
117,
101
]
}
]
}
]
}
}
```
# Configurar o cluster do Amazon MSK e a rede do Amazon VPC para o Lambda
Para conectar a função do AWS Lambda ao cluster do Amazon MSK, é preciso configurar corretamente o cluster e a [Amazon Virtual Private Cloud (VPC)](https://docs.aws.amazon.com/vpc/latest/userguide/what-is-amazon-vpc.html) em que ele reside. Esta página descreve como configurar o cluster e a VPC. Se o cluster e a VPC já estiverem configurados corretamente, consulte [Configurar as origens de eventos do Amazon MSK para o Lambda](with-msk-configure.md) para configurar o mapeamento da origem do evento.
**Topics**
+ [Visão geral dos requisitos de configuração de rede para integrações entre o Lambda e o MSK](#msk-network-requirements)
+ [Configurar um gateway NAT para uma origem de eventos do MSK](#msk-nat-gateway)
+ [Configurar endpoints AWS PrivateLink para uma origem de eventos do MSK](#msk-vpc-privatelink)
## Visão geral dos requisitos de configuração de rede para integrações entre o Lambda e o MSK
A configuração de rede necessária para uma integração entre o Lambda e o MSK depende da arquitetura de rede da aplicação. Há três recursos principais envolvidos nessa integração: o cluster do Amazon MSK, a função do Lambda e o mapeamento da origem do evento do Lambda. Cada um desses recursos reside em uma VPC diferente:
+ O cluster do Amazon MSK normalmente reside em uma sub-rede privada de uma VPC que você gerencia.
+ A função do Lambda reside em uma VPC gerenciada pela AWS e cujo proprietário é o Lambda.
+ O mapeamento da origem do evento do Lambda reside em outra VPC gerenciada pela AWS, cujo proprietário é o Lambda, em separado da VPC que contém a função.
O [mapeamento da origem do evento](invocation-eventsourcemapping.md) é o recurso intermediário entre o cluster do MSK e a função do Lambda. O mapeamento da origem do evento tem duas tarefas principais. Primeiro, ele pesquisa novas mensagens no cluster do MSK. Depois, ele invoca a função do Lambda com essas mensagens. Como esses três recursos estão em VPCs diferentes, as operações de pesquisa e invocação exigem chamadas de rede entre VPCs.
Os requisitos de configuração de rede para o mapeamento da origem do evento variam conforme o uso do [modo provisionado](invocation-eventsourcemapping.md#invocation-eventsourcemapping-provisioned-mode) ou do modo sob demanda, conforme indicado no seguinte diagrama:
![\[\]](http://docs.aws.amazon.com/pt_br/lambda/latest/dg/images/MSK-esm-network-overview.png)
A maneira como o mapeamento da origem do evento do Lambda pesquisa novas mensagens no cluster do MSK é igual em ambos os modos. Para estabelecer uma conexão entre o mapeamento da origem do evento e o cluster do MSK, o Lambda cria uma [ENI de hiperplano](configuration-vpc.md#configuration-vpc-enis) (ou reutiliza uma existente, se disponível) em a sub-rede privada para estabelecer uma conexão segura. Como ilustrado no diagrama, essa ENI de hiperplano usa a configuração de sub-rede e de grupo de segurança do seu cluster do MSK, não da função do Lambda.
Após pesquisar a mensagem do cluster, a maneira como o Lambda invoca a função é diferente em cada modo:
+ No modo provisionado, o Lambda processa automaticamente a conexão entre a VPC do mapeamento da origem do evento e a VPC da função. Assim, não é preciso nenhum componente de rede adicional para invocar a função.
+ No modo sob demanda, o mapeamento da origem do evento do Lambda invoca a função por um caminho através da VPC gerenciada pelo cliente. Por isso, você precisa configurar um [gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) na sub-rede pública da VPC ou endpoints [AWS PrivateLink](https://docs.aws.amazon.com/vpc/latest/privatelink/what-is-privatelink.html) na sub-rede privada da VPC que forneçam acesso ao Lambda, ao [AWS Security Token Service (STS)](https://docs.aws.amazon.com/STS/latest/APIReference/welcome.html) e, opcionalmente, ao [AWS Secrets Manager](https://docs.aws.amazon.com/secretsmanager/latest/userguide/intro.html). A configuração correta dessas opções permite uma conexão entre a sua VPC e a VPC de runtime gerenciada pelo Lambda, que é necessária para invocar a função.
Um gateway NAT permite que recursos na sub-rede privada acessem a Internet pública. Usar essa configuração significa que o tráfego atravessa a Internet antes da invocação da função do Lambda. Os endpoints AWS PrivateLink permitem que as sub-redes privadas se conectem com segurança aos serviços da AWS ou a outros recursos da VPC privada sem atravessar a Internet pública. Consulte [Configurar um gateway NAT para uma origem de eventos do MSK](#msk-nat-gateway) ou [Configurar endpoints AWS PrivateLink para uma origem de eventos do MSK](#msk-vpc-privatelink) para obter detalhes sobre como configurar esses recursos.
Até agora, o pressuposto é que o cluster do MSK reside em uma sub-rede privada dentro da VPC, que é o caso mais comum. Porém, mesmo que o cluster do MSK esteja em uma sub-rede pública dentro da VPC, você deve configurar endpoints AWS PrivateLink para permitir uma conexão segura. A tabela a seguir resume os requisitos de configuração de rede com base em como você configura o cluster do MSK e o mapeamento da origem do evento do Lambda:
| Localização do cluster do MSK (em VPC gerenciada pelo cliente) | Modo de escalação do mapeamento da origem do evento do Lambda | Configuração de rede exigida |
| --- | --- | --- |
| Sub-rede privada | Modo sob demanda | Gateway NAT (na sub-rede pública da VPC) ou endpoints AWS PrivateLink (na sub-rede privada da VPC) para permitir o acesso ao Lambda, ao AWS STS e, opcionalmente, ao Secrets Manager. |
| Sub-rede pública | Modo sob demanda | Endpoints AWS PrivateLink (na sub-rede pública da VPC) para permitir o acesso ao Lambda, ao AWS STS e, opcionalmente, ao Secrets Manager. |
| Sub-rede privada | Modo provisionado | Nenhum |
| Sub-rede pública | Modo provisionado | Nenhum |
Além disso, os grupos de segurança associados ao cluster do MSK devem permitir tráfego nas portas corretas. Configure as seguintes regras de grupo de segurança:
+ **Regras de entrada**: permitir todo o tráfego na porta padrão do agente. A porta que o MSK usa depende do tipo de autenticação no cluster: `9098` para autenticação do IAM, `9096` para SASL/SCRAM e `9094` para TLS. Como alternativa, é possível usar uma regra de grupo de segurança autorreferenciada para permitir o acesso de instâncias dentro do mesmo grupo de segurança.
+ **Regras de saída**: permitir todo o tráfego na porta `443` para destinos externos se a função precisar se comunicar com outros serviços da AWS. Como alternativa, você também pode usar uma regra de grupo de segurança autorreferenciada para limitar o acesso ao agente se não precisar se comunicar com outros serviços da AWS.
+ **Regras de entrada para o endpoint da Amazon VPC**: caso você esteja usando um endpoint da Amazon VPC, o grupo de segurança associado ao endpoint deve permitir tráfego de entrada na porta `443` vindo do grupo de segurança do cluster.
## Configurar um gateway NAT para uma origem de eventos do MSK
Você pode configurar um gateway NAT para permitir que o mapeamento da origem do evento pesquise mensagens do cluster e invoque a função por um caminho via VPC. Isso é necessário apenas se o mapeamento da origem do evento usar o modo sob demanda e o cluster residir em uma sub-rede privada da VPC. Se o cluster residir em uma sub-rede pública da VPC ou se o mapeamento da origem do evento usar o modo provisionado, não será necessário configurar um gateway NAT.
Um [gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) permite que os recursos de uma sub-rede privada acessem a Internet pública. Se você precisar de conectividade privada com o Lambda, consulte [Configurar endpoints AWS PrivateLink para uma origem de eventos do MSK](#msk-vpc-privatelink).
Depois de configurar o gateway NAT, configure as tabelas de rotas apropriadas. Isso permite que o tráfego da sub-rede privada seja roteado para a Internet pública via gateway NAT.
![\[\]](http://docs.aws.amazon.com/pt_br/lambda/latest/dg/images/MSK-NAT-Gateway.png)
As etapas a seguir guiam você pela configuração de um gateway NAT usando o console. Repita essas etapas conforme necessário para cada zona de disponibilidade (AZ).
**Para configurar um gateway NAT e o roteamento adequado (console)**
1. Siga as etapas em [Criar um gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/nat-gateway-working-with.html), observando o seguinte:
+ Os gateways NAT sempre devem residir em uma sub-rede pública. Crie gateways NAT com [conectividade pública](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html).
+ Se o cluster do MSK for replicado em várias AZs, crie apenas um gateway NAT por AZ. Por exemplo, em cada AZ, a VPC deve ter uma sub-rede privada contendo o cluster e uma sub-rede pública contendo o gateway NAT. Para uma configuração com três AZs, você terá três sub-redes privadas, três sub-redes públicas e três gateways NAT.
1. Depois de criar o gateway NAT, abra o console da [Amazon VPC](https://console.aws.amazon.com/vpc/) e escolha **Tabelas de rotas** no menu à esquerda.
1. Escolha **Create Route Table (Criar tabela de rotas)**.
1. Associe essa tabela de rotas à VPC que contém o cluster do MSK. Opcionalmente, insira um nome para a tabela de rotas.
1. Escolha **Create Route Table (Criar tabela de rotas)**.
1. Escolha a tabela de rotas que acabou de criar.
1. Na guia **Associações da sub-rede** selecione **Editar associações da sub-rede**.
+ Associe essa tabela de rotas à sub-rede privada que contém o cluster do MSK.
1. Escolha **Edit routes (Editar rotas)**.
1. Selecione **Adicionar rota**:
1. Em **Destination (Destino)**, selecione `0.0.0.0/0`.
1. Em **Destino**, escolha **Gateway NAT**.
1. Na caixa de pesquisa, escolha o gateway NAT que você criou na etapa 1. Esse deve ser o gateway NAT da mesma AZ que a sub-rede privada que contém o cluster do MSK (a sub-rede privada que você associou a essa tabela de rotas na etapa 6).
1. Escolha **Salvar alterações**.
## Configurar endpoints AWS PrivateLink para uma origem de eventos do MSK
Você pode configurar endpoints AWS PrivateLink para pesquisar mensagens do cluster e invocar a função por um caminho via VPC. Esses endpoints devem permitir que o cluster do MSK acesse o seguinte:
+ O serviço Lambda
+ O [AWS Security Token Service (STS)](https://docs.aws.amazon.com/STS/latest/APIReference/welcome.html)
+ Opcionalmente, o serviço [AWS Secrets Manager](https://docs.aws.amazon.com/secretsmanager/latest/userguide/intro.html). Isso será necessário se o segredo exigido para autenticação do cluster estiver armazenado no Secrets Manager.
A configuração dos endpoints PrivateLink será necessária apenas se o mapeamento da origem do evento usar o modo sob demanda. Se o mapeamento da origem do evento usar o modo provisionado, o Lambda estabelecerá as conexões necessárias para você.
Os endpoints PrivateLink permitem acesso privado seguro aos serviços da AWS por AWS PrivateLink. Como alternativa, para configurar um gateway NAT para conceder ao cluster do MSK acesso à Internet pública, consulte [Configurar um gateway NAT para uma origem de eventos do MSK](#msk-nat-gateway).
Após a configuração dos endpoints da VPC, o cluster do MSK deverá ter acesso direto e privado ao Lambda, ao STS e, opcionalmente, ao Secrets Manager.
![\[\]](http://docs.aws.amazon.com/pt_br/lambda/latest/dg/images/MSK-PrivateLink-Endpoints.png)
As etapas a seguir guiam você pela configuração de um endpoint PrivateLink usando o console. Repita essas etapas conforme necessário para cada endpoint (Lambda, STS, Secrets Manager).
**Para configurar um endpoints PrivateLink da VPC (console)**
1. Abra o [console da Amazon VPC](https://console.aws.amazon.com/vpc/) e escolha **Endpoints ** no menu à esquerda.
1. Escolha **Criar endpoint**.
1. Opcionalmente, insira um nome para o endpoint.
1. Em **Tipo**, escolha **Serviços da AWS**.
1. Em **Serviços**, comece a digitar o nome do serviço. Por exemplo, para criar um endpoint para se conectar ao Lambda, digite `lambda` na caixa de pesquisa.
1. Nos resultados, você deve ver o endpoint do serviço na região atual. Por exemplo, na região Leste dos EUA (Norte da Virgínia), você deve ver `com.amazonaws.us-east-2.lambda`. Selecione este serviço.
1. Em **Configurações de rede**, selecione a VPC que contém o cluster do MSK.
1. Em **Sub-redes**, selecione as AZs em que seu cluster do MSK está.
+ Para cada AZ, em **ID da sub-rede**, escolha a sub-rede privada que contém o cluster do MSK.
1. Em **grupos de segurança** selecione o grupo de segurança associado ao cluster do MSK.
1. Escolha **Criar endpoint**.
Por padrão, os endpoints da Amazon VPC têm políticas do IAM abertas que permitem amplo acesso aos recursos. A prática recomendada é restringir essas políticas para executar as ações necessárias usando esse endpoint. Por exemplo, para o endpoint do Secrets Manager, você pode modificar a política para que permita apenas que o perfil de execução da função acesse o segredo.
**Example Política de endpoint da VPC: endpoint do Secrets Manager**
```
{
"Statement": [
{
"Action": "secretsmanager:GetSecretValue",
"Effect": "Allow",
"Principal": {
"AWS": [
"arn:aws::iam::123456789012:role/my-role"
]
},
"Resource": "arn:aws::secretsmanager:us-west-2:123456789012:secret:my-secret"
}
]
}
```
Para os endpoints do AWS STS e do Lambda, você pode restringir a entidade principal chamadora à entidade principal do serviço Lambda. Porém, certifique-se de usar `"Resource": "*"` nessas políticas.
**Example Política de endpoint da VPC: endpoint do AWS STS**
```
{
"Statement": [
{
"Action": "sts:AssumeRole",
"Effect": "Allow",
"Principal": {
"Service": [
"lambda.amazonaws.com"
]
},
"Resource": "*"
}
]
}
```
**Example Política de endpoint da VPC: endpoint do Lambda**
```
{
"Statement": [
{
"Action": "lambda:InvokeFunction",
"Effect": "Allow",
"Principal": {
"Service": [
"lambda.amazonaws.com"
]
},
"Resource": "*"
}
]
}
```
# Configuração de permissões do Lambda para mapeamentos da origem do evento do Amazon MSK
Além de acessar o cluster do Amazon MSK, a função e o mapeamento da origem do evento precisam de permissões para realizar várias ações de API do Amazon MSK. Adicione essas permissões ao [perfil de execução](lambda-intro-execution-role.md) da função. Se os usuários precisarem de acesso, adicione as permissões necessárias à política de identidades para o usuário ou o perfil.
A política gerenciada [AWSLambdaMSKExecutionRole](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaMSKExecutionRole.html) contém as permissões mínimas necessárias para mapeamentos da origem do evento do Lambda do Amazon MSK. Para simplificar o processo de permissões, é possível:
+ Anexe a política gerenciada [AWSLambdaMSKExecutionRole](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaMSKExecutionRole.html) ao seu perfil de execução.
+ Deixe que o console do Lambda gere as permissões para você. Quando você [cria um mapeamento da origem do evento do Amazon MSK no console](msk-esm-create.md#msk-console), o Lambda avalia seu perfil de execução e alerta você caso alguma permissão esteja faltando. Escolha **Gerar permissões** para atualizar automaticamente seu perfil de execução. Isso não funciona se você criou ou modificou manualmente suas políticas de perfil de execução, ou se as políticas estiverem anexadas a vários perfis. Observe que permissões adicionais ainda podem ser necessárias em seu perfil de execução ao usar recursos avançados, como [Destino em caso de falha](kafka-on-failure.md) ou [Registro de esquemas do AWS Glue](services-consume-kafka-events.md).
**Topics**
+ [Permissões obrigatórias](#msk-required-permissions)
+ [Permissões opcionais.](#msk-optional-permissions)
## Permissões obrigatórias
O perfil de execução da função do Lambda deve ter as permissões exigidas para mapeamentos da origem do evento do Amazon MSK a seguir. Essas permissões estão incluídas na política [AWSLambdaMSKExecutionRole](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaMSKExecutionRole.html).
### Permissões do CloudWatch Logs
As permissões a seguir permitem que o Lambda crie e armazene logs no Amazon CloudWatch Logs.
+ [logs:CreateLogGroup](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_CreateLogGroup.html)
+ [logs:CreateLogStream](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_CreateLogStream.html)
+ [logs:PutLogEvents](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html)
### Permissões de cluster do MSK
As permissões a seguir permitem que o Lambda acesse o cluster do Amazon por você:
+ [kafka:DescribeCluster](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn.html)
+ [kafka:DescribeClusterV2](https://docs.aws.amazon.com/MSK/2.0/APIReference/v2-clusters-clusterarn.html)
+ [kafka:GetBootstrapBrokers](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn-bootstrap-brokers.html)
Recomendamos usar [kafka:DescribeClusterV2](https://docs.aws.amazon.com/MSK/2.0/APIReference/v2-clusters-clusterarn.html) em vez de [kafka:DescribeCluster](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn.html). A permissão v2 funciona com clusters do Amazon MSK provisionados e com tecnologia sem servidor. Você só precisará de uma dessas permissões na sua política.
### Permissões da VPC
As permissões a seguir permitem que o Lambda crie e gerencie interfaces de rede ao se conectar ao seu cluster do Amazon MSK:
+ [ec2:CreateNetworkInterface](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_CreateNetworkInterface.html)
+ [ec2:DescribeNetworkInterfaces](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeNetworkInterfaces.html)
+ [ec2:DescribeVpcs](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeVpcs.html)
+ [ec2:DeleteNetworkInterface](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DeleteNetworkInterface.html)
+ [ec2:DescribeSubnets](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeSubnets.html)
+ [ec2:DescribeSecurityGroups](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeSecurityGroups.html)
## Permissões opcionais.
Sua função do Lambda também pode precisar dessas permissões para:
+ Acesse clusters do Amazon MSK entre contas. Para mapeamentos da origem do evento entre contas, você precisará de [kafka:DescribeVpcConnection](https://docs.aws.amazon.com/msk/1.0/apireference/vpc-connection-arn.html) no perfil de execução. Uma entidade principal do IAM que crie um mapeamento da origem do evento entre contas precisa de [kafka:ListVpcConnections](https://docs.aws.amazon.com/msk/1.0/apireference/vpc-connections.html).
+ Acesse o segredo SCRAM, se estiver usando [autenticação SASL/SCRAM](msk-cluster-auth.md#msk-sasl-scram). Com isso, a função pode usar um nome de usuário e uma senha para se conectar ao Kafka.
+ Descreva seu segredo do Secrets Manager, se estiver usando SASL/SCRAM ou [autenticação mTLS](msk-cluster-auth.md#msk-mtls). Com isso, a função pode recuperar as credenciais ou os certificados necessários para conexões seguras.
+ Acesse sua chave gerenciada pelo cliente do AWS KMS, se o seu segredo do AWS Secrets Manager estiver criptografado com uma chave gerenciada pelo cliente do AWS KMS.
+ Acesse os segredos do registro de esquemas se estiver usando um registro de esquema com autenticação:
+ Para o AWS Glue Schema Registry: sua função precisa das permissões `glue:GetRegistry` e `glue:GetSchemaVersion`. Elas permitem que a função pesquise e use as regras de formato de mensagens armazenadas no AWS Glue.
+ Para o [Confluent Schema Registry](https://docs.confluent.io/platform/current/schema-registry/security/index.html) com `BASIC_AUTH` ou `CLIENT_CERTIFICATE_TLS_AUTH`: a função precisa da permissão `secretsmanager:GetSecretValue` para o segredo que contém credenciais de autenticação. Com isso, a função pode recuperar o nome de usuário/senha ou os certificados necessários para acessar o Confluent Schema Registry.
+ Para certificados de CA privados: a função precisa da permissão secretsmanager:GetSecretValue para o segredo que contém o certificado. Com isso, a função pode verificar a identidade dos registros de esquema que usam certificados personalizados.
+ Acesse grupos de consumidores do cluster do Kafka e mensagens de sondagem do tópico se você estiver usando a autenticação do IAM para o mapeamento da origem do evento.
Eles correspondem às seguintes permissões necessárias:
+ [kafka:ListScramSecrets](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn-scram-secrets.html): permite a listagem de segredos SCRAM para autenticação do Kafka
+ [secretsmanager:GetSecretValue](https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html): permite a recuperação de segredos do Secrets Manager
+ [kms:Decrypt](https://docs.aws.amazon.com/kms/latest/APIReference/API_Decrypt.html): permite a descriptografia de dados criptografados usando o AWS KMS
+ [glue:GetRegistry](https://docs.aws.amazon.com/glue/latest/webapi/API_GetRegistry.html): permite o acesso ao AWS Glue Schema Registry
+ [glue:GetSchemaVersion](https://docs.aws.amazon.com/glue/latest/webapi/API_GetSchemaVersion.html): permite a recuperação de versões específicas de esquema do AWS Glue Schema Registry
+ [kafka-cluster:Connect](https://docs.aws.amazon.com/service-authorization/latest/reference/list_apachekafkaapisforamazonmskclusters.html): concede permissão para se conectar e se autenticar no cluster.
+ [kafka-cluster:AlterGroup](https://docs.aws.amazon.com/service-authorization/latest/reference/list_apachekafkaapisforamazonmskclusters.html): concede permissão para entrar em grupos de um cluster, equivalente a READ GROUP ACL do Apache Kafka.
+ [kafka-cluster:DescribeGroup](https://docs.aws.amazon.com/service-authorization/latest/reference/list_apachekafkaapisforamazonmskclusters.html): concede permissão para descrever os grupos de um cluster, equivalente à ACL DESCRIBE GROUP do Apache Kafka.
+ [kafka-cluster:DescribeTopic](https://docs.aws.amazon.com/service-authorization/latest/reference/list_apachekafkaapisforamazonmskclusters.html): concede permissão para descrever os grupos de um cluster, equivalente à ACL DESCRIBE TOPIC do Apache Kafka.
+ [kafka-cluster:ReadData](https://docs.aws.amazon.com/service-authorization/latest/reference/list_apachekafkaapisforamazonmskclusters.html): concede permissão para ler dados dos tópicos de um cluster, equivalente a READ TOPIC ACL do Apache Kafka.
Além disso, se quiser enviar registros cuja invocação não teve êxito para um destino em caso de falha, você precisará das seguintes permissões, dependendo do tipo de destino:
+ Para destinos do Amazon SQS: [sqs:SendMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html): permite enviar mensagens a uma fila do Amazon SQS
+ Para destinos do Amazon SNS: [sns:Publish](https://docs.aws.amazon.com/sns/latest/api/API_Publish.html): permite publicar mensagens em um tópico do Amazon SNS
+ Para destinos de buckets do Amazon S3: [s3:PutObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html) e [s3:ListBucket](https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListBucket.html): permite escrever e listar objetos em um bucket do Amazon S3
Para solução de problemas de erros comuns de autenticação e de autorização, consulte [Solução de problemas de erros do mapeamento da origem do evento do Kafka](with-kafka-troubleshoot.md).
# Configurar as origens de eventos do Amazon MSK para o Lambda
Para usar um cluster do Amazon MSK como origem de eventos para a função do Lambda, você cria [um mapeamento da origem do evento](invocation-eventsourcemapping.md) que conecta os dois recursos. Esta página descreve como criar um mapeamento da origem do evento para o Amazon MSK.
Esta página pressupõe que você já configurou adequadamente o cluster do MSK e a [Amazon Virtual Private Cloud (VPC)](https://docs.aws.amazon.com/vpc/latest/userguide/what-is-amazon-vpc.html) em que ele reside. Se você precisar configurar o cluster ou a VPC, consulte [Configurar o cluster do Amazon MSK e a rede do Amazon VPC para o Lambda](with-msk-cluster-network.md). Para configurar o comportamento de novas tentativas para tratamento de erros, consulte [Configuração dos controles de tratamento de erros para origens de eventos do Kafka](kafka-retry-configurations.md).
**Topics**
+ [Usar um cluster do Amazon MSK como uma origem de eventos](#msk-esm-overview)
+ [Configuração de métodos de autenticação de cluster do Amazon MSK no Lambda](msk-cluster-auth.md)
+ [Criar um mapeamento da origem do evento do Lambda para uma origem de evento do Amazon MSK](msk-esm-create.md)
+ [Criar mapeamentos da origem do evento entre contas no Lambda](msk-cross-account.md)
+ [Parâmetros de configuração de origem do evento do Amazon MSK no Lambda](msk-esm-parameters.md)
## Usar um cluster do Amazon MSK como uma origem de eventos
Quando você adiciona seu cluster do Apache Kafka ou do Amazon MSK como um gatilho para a função do Lambda, o cluster é usado como uma [origem de eventos](invocation-eventsourcemapping.md).
O Lambda lê os dados de eventos dos tópicos do Kafka que você especifica como `Topics` em uma solicitação de [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) com base na [posição inicial](kafka-starting-positions.md) especificada. Após o processamento bem-sucedido, seu tópico do Kafka é confirmado no cluster do Kafka.
O Lambda lê as mensagens sequencialmente para cada partição de tópico do Kafka. Uma única carga do Lambda pode conter mensagens de várias partições. Quando mais registros ficam disponíveis, o Lambda continua processando os registros em lotes, com base no valor de BatchSize especificado na solicitação de [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html), até a função estar atualizada com o tópico.
Depois que o Lambda processa cada lote, ele confirma os deslocamentos das mensagens nesse lote. Se sua função retorna um erro para qualquer uma das mensagens em um lote, o Lambda tenta novamente todo o lote de mensagens até que o processamento seja bem-sucedido ou as mensagens expiram. É possível enviar registros que apresentaram falha em todas as tentativas a um destino em caso de falha para processamento posterior.
**nota**
Embora as funções do Lambda normalmente tenham um limite máximo de tempo de 15 minutos, os mapeamentos da origem dos eventos para o Amazon MSK, o Apache Kafka autogerenciado, o Amazon DocumentDB e o Amazon MQ para ActiveMQ e RabbitMQ são compatíveis somente com funções com limites máximos de tempo limite de 14 minutos.
# Configuração de métodos de autenticação de cluster do Amazon MSK no Lambda
O Lambda precisa de permissão para acessar o cluster do Amazon MSK, recuperar registros e realizar outras tarefas. O Amazon MSK é compatível com vários métodos de autenticação com o cluster do MSK.
**Topics**
+ [Acesso não autenticado](#msk-unauthenticated)
+ [Autenticação SASL/SCRAM](#msk-sasl-scram)
+ [Autenticação TLS mútua](#msk-mtls)
+ [Autenticação do IAM](#msk-iam-auth)
+ [Como o Lambda escolhe um operador de bootstrap](#msk-bootstrap-brokers)
## Acesso não autenticado
Se nenhum cliente acessar o cluster pela Internet, você poderá usar o acesso não autenticado.
## Autenticação SASL/SCRAM
O Lambda é compatível com autenticação [Simple Authentication and Security Layer/Salted Challenge Response Authentication Mechanism (SASL/SCRAM)](https://docs.aws.amazon.com/msk/latest/developerguide/msk-password-tutorial.html), com a função de hash SHA-512 e com a criptografia de Transport Layer Security (TLS). Para que o Lambda se conecte ao cluster, armazene as credenciais de autenticação (nome de usuário e senha) em um segredo do Secrets Manager e referencie esse segredo quando configurar o mapeamento da origem do evento.
Para obter mais informações sobre o uso do Secrets Manager, consulte [Sign-in credentials authentication with Secrets Manager](https://docs.aws.amazon.com/msk/latest/developerguide/msk-password.html) no *Amazon Managed Streaming for Apache Kafka Developer Guide*.
**nota**
O Amazon MSK não oferece suporte a autenticação SASL/PLAIN.
## Autenticação TLS mútua
O Mutual TLS (mTLS) fornece autenticação bidirecional entre o cliente e o servidor. O cliente envia um certificado ao servidor para que o servidor verifique o cliente. O servidor também envia um certificado para o cliente para que ele verifique o servidor.
Para integrações do Amazon MSK com o Lambda, o cluster do MSK atua como servidor e o Lambda atua como cliente.
+ Para que o Lambda verifique o cluster do MSK, você configura um certificado de cliente como um segredo no Secrets Manager e referencia esse certificado na configuração do mapeamento da origem do evento. O certificado do servidor deve ser assinado por uma autoridade de certificação (CA) no armazenamento de confiança da .
+ O cluster do MSK também envia um certificado de servidor ao Lambda. O certificado do servidor deve ser assinado por uma autoridade de certificação (CA) no armazenamento de confiança da AWS.
O Amazon MSK não é compatível com certificados de servidor auto-assinados. Todos os agentes no Amazon MSK usam [certificados públicos](https://docs.aws.amazon.com/msk/latest/developerguide/msk-encryption.html) assinados por [CAs do Amazon Trust Services](https://www.amazontrust.com/repository/), em que o Lambda confia por padrão.
### Configurar o segredo de mTLS
O segredo CLIENT\$1CERTIFICATE\$1TLS\$1AUTH requer um campo de certificado e um campo de chave privada. Para uma chave privada criptografada, o segredo requer uma senha de chave privada. Tanto o certificado como a chave privada devem estar no formato PEM.
**nota**
O Lambda oferece suporte a algoritmos de criptografia de chave privada [PBES1](https://datatracker.ietf.org/doc/html/rfc2898/#section-6.1) (mas não PBES2).
O campo certificate (certificado) deve conter uma lista de certificados, começando pelo certificado do cliente, seguido por quaisquer certificados intermediários e terminando com o certificado raiz. Cada certificado deve iniciar em uma nova linha com a seguinte estrutura:
```
-----BEGIN CERTIFICATE-----
-----END CERTIFICATE-----
```
O Secrets Manager oferece suporte a segredos de até 65.536 bytes, que é espaço suficiente para cadeias de certificados longas.
A chave privada deve estar no formato [PKCS \$18](https://datatracker.ietf.org/doc/html/rfc5208), com a seguinte estrutura:
```
-----BEGIN PRIVATE KEY-----
-----END PRIVATE KEY-----
```
Para uma chave privada criptografada, use a seguinte estrutura:
```
-----BEGIN ENCRYPTED PRIVATE KEY-----
-----END ENCRYPTED PRIVATE KEY-----
```
O exemplo a seguir exibe o conteúdo de um segredo para autenticação mTLS usando uma chave privada criptografada. Para uma chave privada criptografada, inclua a senha da chave privada no segredo.
```
{
"privateKeyPassword": "testpassword",
"certificate": "-----BEGIN CERTIFICATE-----
MIIE5DCCAsygAwIBAgIRAPJdwaFaNRrytHBto0j5BA0wDQYJKoZIhvcNAQELBQAw
...
j0Lh4/+1HfgyE2KlmII36dg4IMzNjAFEBZiCRoPimO40s1cRqtFHXoal0QQbIlxk
cmUuiAii9R0=
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
MIIFgjCCA2qgAwIBAgIQdjNZd6uFf9hbNC5RdfmHrzANBgkqhkiG9w0BAQsFADBb
...
rQoiowbbk5wXCheYSANQIfTZ6weQTgiCHCCbuuMKNVS95FkXm0vqVD/YpXKwA/no
c8PH3PSoAaRwMMgOSA2ALJvbRz8mpg==
-----END CERTIFICATE-----",
"privateKey": "-----BEGIN ENCRYPTED PRIVATE KEY-----
MIIFKzBVBgkqhkiG9w0BBQ0wSDAnBgkqhkiG9w0BBQwwGgQUiAFcK5hT/X7Kjmgp
...
QrSekqF+kWzmB6nAfSzgO9IaoAaytLvNgGTckWeUkWn/V0Ck+LdGUXzAC4RxZnoQ
zp2mwJn2NYB7AZ7+imp0azDZb+8YG2aUCiyqb6PnnA==
-----END ENCRYPTED PRIVATE KEY-----"
}
```
Para obter mais informações sobre o mTLS para Amazon MSK e instruções para gerar um certificado de cliente, consulte [Mutual TLS client authentication for Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html) no *Amazon Managed Streaming for Apache Kafka Developer Guide*.
## Autenticação do IAM
É possível usar o AWS Identity and Access Management (IAM) para autenticar a identidade dos clientes que se conectam ao cluster do MSK. Com a autenticação do IAM, o Lambda depende das permissões do [perfil de execução](lambda-intro-execution-role.md) da função para se conectar ao cluster, recuperar registros e realizar outras ações necessárias. Para ver um exemplo de política com as permissões necessárias, consulte [Create authorization policies for the IAM role](https://docs.aws.amazon.com/msk/latest/developerguide/create-iam-access-control-policies.html) no *Amazon Managed Streaming for Apache Kafka Developer Guide*.
Se a autenticação do IAM estiver ativa no cluster do MSK e você não fornecer um segredo, o Lambda usará automaticamente a autenticação do IAM por padrão.
Para obter mais informações sobre a autenticação do IAM no Amazon MSK, consulte [IAM access control](https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html).
## Como o Lambda escolhe um operador de bootstrap
O Lambda escolhe um [operador de bootstrap](https://docs.aws.amazon.com/msk/latest/developerguide/msk-get-bootstrap-brokers.html) com base nos métodos de autenticação disponíveis no cluster e dependendo de você fornecer um segredo para autenticação. Se você fornecer um segredo para mTLS ou SASL/SCRAM, o Lambda escolherá esse método de autenticação automaticamente. Se você não fornecer um segredo, o Lambda selecionará o método de autenticação mais forte que estiver ativo no seu cluster. Veja a seguir a ordem de prioridade na qual o Lambda seleciona um operador, da autenticação mais forte para a mais fraca:
+ mTLs (segredo fornecido para mTLs)
+ SASL/SCRAM (segredo fornecido para SASL/SCRAM)
+ SASL IAM (nenhum segredo fornecido, e a autenticação do IAM está ativa)
+ TLS não autenticado (nenhum segredo fornecido, e a autenticação do IAM não está ativa)
+ Texto simples (nenhum segredo fornecido, e a autenticação do IAM e o TLS não autenticado não estão ativos)
**nota**
Se o Lambda não conseguir se conectar ao tipo de operador mais seguro, o Lambda não tentará se conectar a um tipo de operador diferente (mais fraco). Se quiser que o Lambda escolha um tipo de operador mais fraco, desative todos os métodos de autenticação mais fortes no seu cluster.
# Criar um mapeamento da origem do evento do Lambda para uma origem de evento do Amazon MSK
Para criar um mapeamento da origem do evento, você pode usar o console do Lambda, a [AWS Command Line Interface (CLI)](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) ou um [SDK da AWS](https://aws.amazon.com/getting-started/tools-sdks/).
**nota**
Quando você cria o mapeamento da origem do evento, o Lambda cria uma [ENI de hiperplano](configuration-vpc.md#configuration-vpc-enis) na sub-rede privada que contém o cluster do MSK, permitindo que o Lambda estabeleça uma conexão segura. Essa ENI de hiperplano permite usar a configuração de sub-rede e grupo de segurança do cluster MSK, não a função do Lambda.
As etapas do console a seguir adicionam um cluster do Amazon MSK como acionador da função do Lambda. Nos bastidores, isso cria um recurso de mapeamento da origem do evento.
**Para adicionar um acionador do Amazon MSK à sua função do Lambda (console)**
1. Abra a [página Funções](https://console.aws.amazon.com/lambda/home#/functions) no console do Lambda.
1. Escolha o nome da função do Lambda à qual você deseja adicionar um acionador. do Amazon MSK.
1. Em **Visão geral da função**, escolha **Adicionar gatilho**.
1. Em **Configuração do acionador**, escolha **MSK**.
1. Para especificar os detalhes do cluster do Kafka, faça o seguinte:
1. Em **Cluster do MSK**, selecione seu cluster.
1. Em **Nome do tópico**, insira o nome do tópico do Kafka do qual as mensagens serão consumidas.
1. Em **ID do grupo de consumidores**, insira o ID de um grupo de consumidores do Kafka para ingressar, se aplicável. Para obter mais informações, consulte [ID de grupo de consumidores personalizável no Lambda](kafka-consumer-group-id.md).
1. Em **Autenticação de cluster**, faça as configurações necessárias. Para obter mais informações sobre autenticação de cluster, consulte [Configuração de métodos de autenticação de cluster do Amazon MSK no Lambda](msk-cluster-auth.md).
+ Ative a opção **Usar autenticação** se quiser que o Lambda realize a autenticação com o cluster MSK quando estabelecer uma conexão. A autenticação é recomendada.
+ Se você usar autenticação, em **Método de autenticação**, escolha o método de autenticação a ser usado.
+ Se você usar autenticação, em **Chave do Secrets Manager**, escolha a chave do Secrets Manager que contém as credenciais de autenticação necessárias para acessar o cluster.
1. Em **Configuração do agente de sondagem de eventos**, faça as configurações necessárias.
+ Escolha **Ativar acionador** para habilitar o acionador imediatamente após a criação.
+ Escolha se deseja **Configurar o modo provisionado** para o mapeamento da origem do evento. Para obter mais informações, consulte [Modos de escalabilidade de sondagem de eventos do Apache Kafka no Lambda](kafka-scaling-modes.md).
+ Se você configurar o modo provisionado, insira um valor para **Mínimo de agentes de sondagem de eventos**, um valor para **Máximo de agentes de sondagem de eventos** e um valor opcional para PollerGroupName para especificar o agrupamento de vários ESMs na mesma VPC de origem dos eventos.
+ Em **Posição inicial**, escolha como quer que o Lambda comece a ler o fluxo. Para obter mais informações, consulte [Posições iniciais de sondagem e fluxo do Apache Kafka no Lambda](kafka-starting-positions.md).
1. Em **Processamento em lotes**, faça as configurações necessárias. Para obter mais informações sobre processamento em lotes, consulte [Comportamento de lotes](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching).
1. Em **Batch size** (Tamanho do lote), insira o número máximo de mensagens a serem recebidas em um único lote.
1. Em **Janela de lotes**, insira o número máximo de segundos que o Lambda leva coletando registros antes de invocar a função.
1. Em **Filtragem**, faça as configurações necessárias. Para obter mais informações sobre filtragem, consulte [Filtrar as origens dos eventos do Amazon MSK e do Apache Kafka autogerenciado](kafka-filtering.md).
+ Em **Critérios de filtro**, adicione as definições de critérios de filtro para definir se um evento deve ou não ser processado.
1. Em **Tratamento de falhas**, faça as configurações necessárias. Para obter mais informações sobre tratamento de falhas, consulte [Capturar lotes descartados para uma origem do evento do Amazon MSK e do Apache Kafka autogerenciado](kafka-on-failure.md).
+ Em **Destino em caso de falha**, especifique o ARN do destino em caso de falha.
1. Em **Tags**, insira as tags a serem associadas a esse mapeamento da origem do evento.
1. Para criar o acionador, selecione **Add** (Adicionar).
Também é possível criar o mapeamento da origem do evento usando a AWS CLI com o comando [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html). O exemplo a seguir cria um mapeamento da origem do evento para mapear a função do Lambda `my-msk-function` para o tópico `AWSKafkaTopic`, começando pela mensagem `LATEST`. Esse comando também usa o objeto [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) para instruir o Lambda a usar a autenticação [SASL/SCRAM](msk-cluster-auth.md#msk-sasl-scram) quando se conectar ao cluster.
```
aws lambda create-event-source-mapping \
--event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
--topics AWSKafkaTopic \
--starting-position LATEST \
--function-name my-kafka-function
--source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
```
Se o cluster usar a [autenticação mTLS](msk-cluster-auth.md#msk-mtls), inclua um objeto [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) que especifique a `CLIENT_CERTIFICATE_TLS_AUTH` e o ARN de uma chave do Secrets Manager. Isso é mostrado no seguinte comando:
```
aws lambda create-event-source-mapping \
--event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
--topics AWSKafkaTopic \
--starting-position LATEST \
--function-name my-kafka-function
--source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
```
Quando o cluster usa a [autenticação do IAM](msk-cluster-auth.md#msk-iam-auth), você não precisa de um objeto [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html). Isso é mostrado no seguinte comando:
```
aws lambda create-event-source-mapping \
--event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
--topics AWSKafkaTopic \
--starting-position LATEST \
--function-name my-kafka-function
```
# Criar mapeamentos da origem do evento entre contas no Lambda
É possível usar a [conectividade privada com várias VPCs](https://docs.aws.amazon.com/msk/latest/developerguide/aws-access-mult-vpc.html) para conectar uma função do Lambda a um cluster DO MSK provisionado em outra Conta da AWS. A conectividade com várias VPCs usa AWS PrivateLink, que mantém todo o tráfego na rede da AWS.
**nota**
Não é possível criar mapeamentos da origem do evento entre contas para clusters do MSK sem servidor.
Para criar um mapeamento da origem de eventos entre contas, primeiro você precisa [configurar a conectividade com várias VPCs](https://docs.aws.amazon.com/msk/latest/developerguide/aws-access-mult-vpc.html#mvpc-cluster-owner-action-turn-on) para o cluster do MSK. Ao criar o mapeamento da origem de eventos, use o ARN da conexão VPC gerenciada, em vez do ARN do cluster, conforme mostrado nos exemplos a seguir. A operação [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) também difere dependendo do tipo de autenticação usado pelo cluster do MSK.
**Example : criar um mapeamento da origem de eventos entre contas para o cluster que usa autenticação do IAM**
Quando o cluster usa a [autenticação baseada em perfil do IAM](msk-cluster-auth.md#msk-iam-auth), você não precisa de um objeto [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html). Exemplo:
```
aws lambda create-event-source-mapping \
--event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
--topics AWSKafkaTopic \
--starting-position LATEST \
--function-name my-kafka-function
```
**Example : criar um mapeamento da origem de eventos entre contas para o cluster que usa autenticação SASL/SCRAM**
Se o cluster usar [autenticação SASL/SCRAM](msk-cluster-auth.md#msk-sasl-scram), você precisa incluir um objeto [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) que especifique `SASL_SCRAM_512_AUTH` e um ARN secreto do Secrets Manager.
Há duas maneiras de usar segredos para mapeamentos da origem do evento entre contas do Amazon MSK com autenticação SASL/SCRAM:
+ Crie um segredo na conta da função do Lambda e sincronize-o com o segredo do cluster. [Crie uma rotação](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html) para manter os dois segredos sincronizados. Essa opção permite controlar o segredo pela conta da função.
+ Use o segredo associado ao cluster do MSK. Esse segredo deve permitir o acesso entre contas à conta da função do Lambda. Para obter mais informações, consulte [Permissões para segredos do AWS Secrets Manager para usuários em uma conta diferente](https://docs.aws.amazon.com/secretsmanager/latest/userguide/auth-and-access_examples_cross.html).
```
aws lambda create-event-source-mapping \
--event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
--topics AWSKafkaTopic \
--starting-position LATEST \
--function-name my-kafka-function \
--source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'
```
**Example : criar um mapeamento da origem de eventos entre contas para o cluster que usa autenticação mTLS**
Se o cluster usar [autenticação mTLS](msk-cluster-auth.md#msk-mtls), você precisa incluir um objeto [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) que especifique `CLIENT_CERTIFICATE_TLS_AUTH` e um ARN secreto do Secrets Manager. O segredo pode ser armazenado na conta do cluster ou na conta da função do Lambda.
```
aws lambda create-event-source-mapping \
--event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
--topics AWSKafkaTopic \
--starting-position LATEST \
--function-name my-kafka-function \
--source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'
```
# Parâmetros de configuração de origem do evento do Amazon MSK no Lambda
Todos os tipos de origem de evento Lambda compartilham o mesmo[CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html)e[UpdateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html)Operações de API do. Porém, apenas alguns dos parâmetros se aplicam ao Amazon MSK, como mostra a tabela a seguir.
| Parâmetro | Obrigatório | Padrão | Observações |
| --- | --- | --- | --- |
| AmazonManagedKafkaEventSourceConfig | N | Contém o campo ConsumerGroupId, que assume, por padrão, um valor exclusivo. | Pode definir apenas em Criar |
| BatchSize | N | 100 | Máximo: 10.000. |
| DestinationConfig | N | N/D | [Capturar lotes descartados para uma origem do evento do Amazon MSK e do Apache Kafka autogerenciado](kafka-on-failure.md) |
| Habilitado | N | Verdadeiro | |
| BisectBatchOnFunctionError | N | Falso | [Configuração dos controles de tratamento de erros para origens de eventos do Kafka](kafka-retry-configurations.md) |
| FunctionResponseTypes | N | N/D | [Configuração dos controles de tratamento de erros para origens de eventos do Kafka](kafka-retry-configurations.md) |
| MaximumRecordAgeInSeconds | N | -1 (infinito) | [Configuração dos controles de tratamento de erros para origens de eventos do Kafka](kafka-retry-configurations.md) |
| MaximumRetryAttempts | N | -1 (infinito) | [Configuração dos controles de tratamento de erros para origens de eventos do Kafka](kafka-retry-configurations.md) |
| EventSourceArn | S | N/D | Pode definir apenas em Criar |
| FilterCriteria | N | N/D | [Controlar quais eventos o Lambda envia para a função](invocation-eventfiltering.md) |
| FunctionName | S | N/D | |
| KMSKeyArn | N | N/D | [Criptografia de critérios de filtro](invocation-eventfiltering.md#filter-criteria-encryption) |
| MaximumBatchingWindowInSeconds | N | 500 ms | [Comportamento de lotes](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching) |
| ProvisionedPollersConfig | N | `MinimumPollers`: o valor padrão será 1 se não for especificado. `MaximumPollers`: o valor padrão será 200 se não for especificado. `PollerGroupName`: N/D | [Modo provisionado](kafka-scaling-modes.md#kafka-provisioned-mode) |
| SourceAccessConfigurations | N | Sem credenciais do | Credenciais de autenticação SASL/SCRAM ou CLIENT\$1CERTIFICATE\$1TLS\$1AUTH (MutualTLS) para a fonte do evento |
| StartingPosition | S | N/D | AT\$1TIMESTAMP, TRIM\$1HORIZON, ou LATEST Pode definir apenas em Criar |
| StartingPositionTimestamp | N | N/D | Obrigatório se StartingPosition estiver definido como AT\$1TIMESTAMP |
| Tags | N | N/D | [Uso de tags em mapeamentos da origem do evento](tags-esm.md) |
| Tópicos | S | N/D | Nome do tópico do Kafka Pode definir apenas em Criar |
**nota**
Quando você especifica um `PollerGroupName`, vários ESMs dentro da mesma Amazon VPC podem compartilhar a capacidade da unidade de sondagem de evento (EPU). É possível usar essa opção para otimizar os custos do modo provisionado para seus ESMs. Requisitos para o agrupamento de ESM:
Os ESMs devem estar na mesma Amazon VPC
Máximo de 100 ESMs por grupo de agentes de sondagem
O máximo agregado de agentes de sondagem em todos os ESMs em um grupo não pode exceder a 2.000
É possível atualizar o `PollerGroupName` para mover um ESM para um grupo diferente ou remover um ESM de um grupo configurando `PollerGroupName` como uma string vazia (“”).
# Tutorial: Usar um mapeamento de origem do evento do Amazon MSK para invocar uma função do Lambda
Neste tutorial, você fará o seguinte:
+ Criar uma função do Lambda na mesma conta da AWS que um cluster do Amazon MSK existente.
+ Configurar a rede e a autenticação para que o Lambda se comunique com o Amazon MSK.
+ Configurar um mapeamento de origem do evento do Amazon MSK do Lambda, o qual executa sua função do Lambda quando os eventos aparecem no tópico.
Depois de concluir essas etapas, quando os eventos forem enviados para o Amazon MSK, você poderá configurar uma função do Lambda para processar esses eventos automaticamente com seu próprio código do Lambda personalizado.
**O que você pode fazer com esse recurso?**
**Exemplo de solução: use um mapeamento de origem de evento do MSK para fornecer pontuações ao vivo para seus clientes.**
Considere o seguinte cenário: sua empresa hospeda uma aplicação Web em que seus clientes podem ver informações sobre eventos ao vivo, como competições esportivas. As atualizações de informações do jogo são fornecidas à sua equipe por meio de um tópico do Kafka no Amazon MSK. Você deseja criar uma solução que consuma atualizações do tópico do MSK para fornecer uma visão atualizada do evento ao vivo aos clientes dentro de uma aplicação desenvolvida por você. Você optou pela seguinte abordagem de design: suas aplicações clientes se comunicarão com um backend sem servidor hospedado na AWS. Os clientes se conectarão por meio de sessões de websocket usando a API de WebSocket do Amazon API Gateway.
Nessa solução, você precisa de um componente que leia eventos do MSK, execute uma lógica personalizada para preparar esses eventos para a camada de aplicação e, em seguida, encaminhe essas informações para a API do API Gateway. É possível implementar esse componente com o AWS Lambda, fornecendo sua lógica personalizada em uma função do Lambda e, em seguida, chamando-a com um mapeamento de origem de eventos do Amazon MSK do AWS Lambda.
Para obter mais informações sobre a implementação de soluções usando a API de WebSocket do Amazon API Gateway, consulte os [tutoriais da API de WebSocket](https://docs.aws.amazon.com/apigateway/latest/developerguide/websocket-api-chat-app.html) na documentação do API Gateway.
## Pré-requisitos
Uma conta da AWS com os seguintes recursos pré-configurados:
**Para cumprir esses pré-requisitos, recomendamos seguir a seção [Conceitos básicos do uso do Amazon MSK](https://docs.aws.amazon.com//msk/latest/developerguide/getting-started.html) na documentação do Amazon MSK.**
+ Um cluster do Amazon MSK. Consulte [Criar um cluster do Amazon MSK](https://docs.aws.amazon.com//msk/latest/developerguide/create-cluster.html) em *Conceitos básicos do uso do Amazon MSK*.
+ A seguinte configuração:
+ Certifique-se de que a **autenticação baseada em perfis do IAM** esteja **habilitada** nas configurações de segurança do cluster. Isso melhora sua segurança ao limitar sua função do Lambda para acessar somente os recursos necessários do Amazon MSK. Esse comportamento é habilitado por padrão em todos os novos clusters do Amazon MSK.
+ Certifique-se de que o **acesso público** esteja desativado nas configurações de rede do cluster. Restringir o acesso do cluster do Amazon MSK à Internet melhora sua segurança ao limitar o número de intermediários que lidam com seus dados. Esse comportamento é habilitado por padrão em todos os novos clusters do Amazon MSK.
+ Um tópico do Kafka no cluster do Amazon MSK para usar nesta solução. Consulte [Criar um tópico](https://docs.aws.amazon.com//msk/latest/developerguide/create-topic.html) em *Conceitos básicos do uso do Amazon MSK*.
+ Um host administrativo do Kafka configurado para recuperar informações do cluster do Kafka e enviar eventos do Kafka para seu tópico para teste, como uma instância do Amazon EC2 com a CLI de administração do Kafka e a biblioteca do IAM do Amazon MSK instalada. Consulte [Criar uma máquina cliente](https://docs.aws.amazon.com//msk/latest/developerguide/create-client-machine.html) em *Conceitos básicos do uso do Amazon MSK*.
Depois de configurar esses recursos, reúna as seguintes informações da sua conta da AWS para confirmar que está tudo pronto para continuar.
+ O nome do cluster do Amazon MSK. Essas informações podem ser encontradas no console do Amazon MSK.
+ O UUID do cluster, parte do ARN do seu cluster do Amazon MSK, o qual pode ser encontrado no console do Amazon MSK. Siga os procedimentos em [Listar clusters](https://docs.aws.amazon.com/msk/latest/developerguide/msk-list-clusters.html) na documentação do Amazon MSK para encontrar essas informações.
+ Os grupos de segurança associados ao cluster do Amazon MSK. Essas informações podem ser encontradas no console do Amazon MSK. Nas etapas a seguir, faça referência a elas como *clusterSecurityGroups*.
+ O ID da Amazon VPC contendo seu cluster do Amazon MSK. É possível encontrar essas informações identificando sub-redes associadas ao seu cluster do Amazon MSK no console do Amazon MSK e, em seguida, identificando a Amazon VPC associada à sub-rede no console da Amazon VPC.
+ O nome do tópico do Kafka usado na solução. É possível encontrar essas informações chamando o cluster do Amazon MSK com a CLI de `topics` do Kafka no host administrativo do Kafka. Para obter mais informações sobre a CLI de tópicos, consulte [Adicionar e remover tópicos](https://kafka.apache.org/documentation/#basic_ops_add_topic) na documentação do Kafka.
+ O nome de um grupo de consumidores para seu tópico do Kafka, adequado para uso por sua função do Lambda. Esse grupo pode ser criado automaticamente pelo Lambda, então não é necessário criá-lo com a CLI do Kafka. Se você precisar gerenciar seus grupos de consumidores, para saber mais sobre a CLI de grupos de consumidores, consulte [Gerenciar grupos de consumidores](https://kafka.apache.org/documentation/#basic_ops_consumer_group) na documentação do Kafka.
As seguintes permissões em sua conta da AWS:
+ Permissão para criar e gerenciar uma função do Lambda
+ Permissão para criar políticas do IAM e associá-las à sua função do Lambda.
+ Permissão para criar endpoints da Amazon VPC e alterar a configuração de rede na Amazon VPC que hospeda seu cluster do Amazon MSK.
### Instalar o AWS Command Line Interface
Se você ainda não instalou a AWS Command Line Interface, siga as etapas em [Instalar ou atualizar a versão mais recente da AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) para instalá-la.
O tutorial requer um terminal de linha de comando ou um shell para executar os comandos. No Linux e no macOS, use o gerenciador de pacotes e de shell de sua preferência.
**nota**
No Windows, alguns comandos da CLI do Bash que você costuma usar com o Lambda (como `zip`) não são compatíveis com os terminais integrados do sistema operacional. Para obter uma versão do Ubuntu com o Bash integrada no Windows, [instale o Subsistema do Windows para Linux](https://docs.microsoft.com/en-us/windows/wsl/install-win10).
## Configurar a conectividade de rede para que o Lambda se comunique com o Amazon MSK
Use AWS PrivateLink para conectar o Lambda e o Amazon MSK. Isso pode ser feito com a criação de endpoints da Amazon VPC de interface no console do Amazon VPC. Para obter mais informações sobre a configuração de rede, consulte [Configurar o cluster do Amazon MSK e a rede do Amazon VPC para o Lambda](with-msk-cluster-network.md).
Quando um mapeamento de origem de evento do Amazon MSK é executado em nome de uma função do Lambda, ele assume o perfil de execução da função do Lambda. Essa função do IAM autoriza o mapeamento a acessar recursos protegidos pelo IAM, como seu cluster do Amazon MSK. Embora os componentes compartilhem uma função de execução, o mapeamento do Amazon MSK e sua função do Lambda têm requisitos de conectividade separados para suas respectivas tarefas, conforme mostrado no diagrama a seguir.
![\[\]](http://docs.aws.amazon.com/pt_br/lambda/latest/dg/images/msk_tut_network.png)
Seu mapeamento da origem do evento pertence ao seu grupo de segurança do cluster do Amazon MSK. Nesta etapa de rede, crie endpoints do Amazon VPC da VPC do cluster do Amazon MSK para conectar o mapeamento da origem do evento aos serviços do Lambda e STS. Proteja esses endpoints para aceitar tráfego do seu grupo de segurança de cluster do Amazon MSK. Em seguida, ajuste os grupos de segurança do cluster do Amazon MSK para permitir que o mapeamento da origem do evento se comunique com o cluster do Amazon MSK.
É possível configurar as etapas a seguir usando o Console de gerenciamento da AWS.
**Para configurar os endpoints da Amazon VPC de interface para conectar o Lambda e o Amazon MSK**
1. Crie um grupo de segurança para os endpoints da Amazon VPC de interface, *endpointSecurityGroup*, que permite tráfego TCP de entrada na porta 443 proveniente de *clusterSecurityGroups*. Siga o procedimento em [Criar um grupo de segurança](https://docs.aws.amazon.com//AWSEC2/latest/UserGuide/working-with-security-groups.html#creating-security-group) na documentação do Amazon EC2 para criar um grupo de segurança. Em seguida, siga o procedimento em [Adicionar regras a um grupo de segurança](https://docs.aws.amazon.com//AWSEC2/latest/UserGuide/working-with-security-groups.html#adding-security-group-rule) na documentação do Amazon EC2 para adicionar as regras apropriadas.
**Crie um grupo de segurança com as seguintes informações:**
Ao adicionar suas regras de entrada, crie uma regra para cada grupo de segurança em *clusterSecurityGroups*. Para cada regra:
+ Em **Tipo**, selecione **HTTPS**.
+ Em **Origem**, selecione um dos *clusterSecurityGroups*.
1. Crie um endpoint conectando o serviço do Lambda à Amazon VPC contendo seu cluster do Amazon MSK. Siga o procedimento em [Criar um endpoint de interface](https://docs.aws.amazon.com//vpc/latest/privatelink/create-interface-endpoint.html).
**Crie um endpoint de interface com as seguintes informações:**
+ Em **Nome do serviço**, selecione `com.amazonaws.regionName.lambda`, onde *regionName* hospeda sua função do Lambda.
+ Para **VPC**, selecione a Amazon VPC que contém seu cluster do Amazon MSK.
+ Em **Grupos de segurança**, selecione o *endpointSecurityGroup* que você criou anteriormente.
+ Para **Sub-redes**, selecione as sub-redes que hospedam seu cluster do Amazon MSK.
+ Para **Política**, forneça o documento de política a seguir, que protege o endpoint para uso pela entidade principal do serviço do Lambda para a ação `lambda:InvokeFunction`.
```
{
"Statement": [
{
"Action": "lambda:InvokeFunction",
"Effect": "Allow",
"Principal": {
"Service": [
"lambda.amazonaws.com"
]
},
"Resource": "*"
}
]
}
```
+ Certifique-se de que a opção **Habilitar nome de DNS** permaneça selecionada.
1. Crie um endpoint conectando o serviço do AWS STS à Amazon VPC que contém seu cluster do Amazon MSK. Siga o procedimento em [Criar um endpoint de interface](https://docs.aws.amazon.com//vpc/latest/privatelink/create-interface-endpoint.html).
**Crie um endpoint de interface com as seguintes informações:**
+ Em **Nome do serviço**, selecione AWS STS.
+ Para **VPC**, selecione a Amazon VPC que contém seu cluster do Amazon MSK.
+ Em **Grupos de segurança**, selecione *endpointSecurityGroup*.
+ Para **Sub-redes**, selecione as sub-redes que hospedam seu cluster do Amazon MSK.
+ Para **Política**, forneça o documento de política a seguir, que protege o endpoint para uso pela entidade principal do serviço do Lambda para a ação `sts:AssumeRole`.
```
{
"Statement": [
{
"Action": "sts:AssumeRole",
"Effect": "Allow",
"Principal": {
"Service": [
"lambda.amazonaws.com"
]
},
"Resource": "*"
}
]
}
```
+ Certifique-se de que a opção **Habilitar nome de DNS** permaneça selecionada.
1. Para cada grupo de segurança associado ao seu cluster do Amazon MSK, ou seja, em *clusterSecurityGroups*, permita o seguinte:
+ Permita todo o tráfego TCP de entrada e saída na porta 9098 para todos os *clusterSecurityGroups*, inclusive dentro dele mesmo.
+ Permita todo o tráfego TCP de saída na porta 443.
Parte desse tráfego é permitido por regras padrão do grupo de segurança. Portanto, se seu cluster estiver vinculado a um único grupo de segurança e esse grupo tiver regras padrão, regras adicionais não serão necessárias. Para adicionar regras do grupo de segurança, siga o procedimento em [Adicionar regras a um grupo de segurança](https://docs.aws.amazon.com//AWSEC2/latest/UserGuide/working-with-security-groups.html#adding-security-group-rule) na documentação do Amazon EC2.
**Adicione regras aos seus grupos de segurança com as seguintes informações:**
+ Para cada regra de entrada ou regra de saída para a porta 9098, forneça
+ Em **Tipo**, selecione **TCP personalizado**.
+ Em **Intervalo de portas**, forneça 9098.
+ Em **Origem**, forneça um dos *clusterSecurityGroups*.
+ Para cada regra de entrada da porta 443, em **Tipo**, selecione **HTTPS**.
## Crie um perfil do IAM para que o Lambda leia seu tópico do Amazon MSK
Identifique os requisitos de autenticação para que o Lambda leia seu tópico do Amazon MSK e, em seguida, defina-os em uma política. Crie uma função, *lambdaAuthRole*, que autorize o Lambda a usar essas permissões. Autorize ações no seu cluster Amazon MSK usando ações do IAM do `kafka-cluster`. Em seguida, autorize o Lambda a realizar as ações do Amazon MSK `kafka` e do Amazon EC2 necessárias para descobrir e se conectar ao seu cluster do Amazon MSK, bem como as ações do CloudWatch para que o Lambda possa registrar em log o que fez.
**Para descrever os requisitos de autenticação para que o Lambda leia a partir do Amazon MSK**
1. Escreva um documento de política do IAM (um documento JSON), *clusterAuthPolicy*, que permita que o Lambda leia seu tópico do Kafka no seu cluster do Amazon MSK usando seu grupo de consumidores do Kafka. O Lambda exige que um grupo de consumidores do Kafka seja definido durante a leitura.
Altere o modelo a seguir para se alinhar aos seus pré-requisitos:
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kafka-cluster:Connect",
"kafka-cluster:DescribeGroup",
"kafka-cluster:AlterGroup",
"kafka-cluster:DescribeTopic",
"kafka-cluster:ReadData",
"kafka-cluster:DescribeClusterDynamicConfiguration"
],
"Resource": [
"arn:aws:kafka:us-east-1:111122223333:cluster/mskClusterName/cluster-uuid",
"arn:aws:kafka:us-east-1:111122223333:topic/mskClusterName/cluster-uuid/mskTopicName",
"arn:aws:kafka:us-east-1:111122223333:group/mskClusterName/cluster-uuid/mskGroupName"
]
}
]
}
```
------
Para obter mais informações, consulte : [Configuração de permissões do Lambda para mapeamentos da origem do evento do Amazon MSK](with-msk-permissions.md). Ao redigir sua política:
+ Substitua *us-east-1* e *111122223333* por Região da AWS e por Conta da AWS de seu cluster Amazon MSK.
+ Em *mskClusterName*, forneça o nome do seu cluster do Amazon MSK.
+ Em *cluster-uuid*, forneça o UUID no ARN para seu cluster do Amazon MSK.
+ Em *mskTopicName*, forneça o nome do seu tópico do Kafka.
+ Em *mskGroupName*, forneça o nome do seu grupo de consumidores do Kafka.
1. Identifique as permissões do Amazon MSK, do Amazon EC2 e do CloudWatch necessárias para que o Lambda descubra e conecte seu cluster Amazon MSK e registre esses eventos no log.
A política gerenciada `AWSLambdaMSKExecutionRole` define de forma permissiva as permissões necessárias. Use-a nas etapas a seguir.
Em um ambiente de produção, avalie `AWSLambdaMSKExecutionRole` para restringir sua política de perfil de execução com base no princípio do privilégio mínimo e, em seguida, escreva uma política para seu perfil que substitua essa política gerenciada.
Para obter detalhes sobre a linguagem de políticas do IAM, consulte a [Documentação do IAM](https://docs.aws.amazon.com//iam/).
Agora que você escreveu seu documento de política, crie uma política do IAM para poder anexá-la ao seu perfil. É possível fazer isso usando o console com o procedimento a seguir.
**Para criar uma política do IAM a partir do seu documento de política**
1. Faça login no Console de gerenciamento da AWS e abra o console do IAM em [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/).
1. No painel de navegação à esquerda, escolha **Políticas**.
1. Escolha **Criar política**.
1. Na seção **Editor de políticas**, escolha a opção **JSON**.
1. Cole *clusterAuthPolicy*.
1. Quando terminar de adicionar as permissões à política, escolha **Avançar**.
1. Na página **Revisar e criar**, digite um **nome de política** e uma **descrição** (opcional) para a política que você está criando. Revise **Permissões definidas nessa política** para ver as permissões que são concedidas pela política.
1. Escolha **Criar política** para salvar sua nova política.
Para obter mais informações, consulte [Criar políticas do IAM](https://docs.aws.amazon.com//IAM/latest/UserGuide/access_policies_create.html) na documentação do IAM.
Agora que você tem as políticas apropriadas do IAM, crie uma perfil e anexe-o a ela. É possível fazer isso usando o console com o procedimento a seguir.
**Como criar uma função de execução no console do IAM**
1. Abra a página [Roles (Funções)](https://console.aws.amazon.com/iam/home#/roles) no console do IAM.
1. Selecione **Create role** (Criar função).
1. Em **Tipo de entidade confiável**, selecione **Serviço da AWS**.
1. Em **Use case** (Caso de uso), escolha **Lambda**.
1. Escolha **Próximo**.
1. Selecione as políticas a seguir:
+ *clusterAuthPolicy*
+ `AWSLambdaMSKExecutionRole`
1. Escolha **Próximo**.
1. Em **Nome do perfil**, insira *lambdaAuthRole* e, em seguida, escolha **Criar perfil**.
Para obter mais informações, consulte [Definir permissões de uma função do Lambda com um perfil de execução](lambda-intro-execution-role.md).
## Crie uma função do Lambda para ler do seu tópico do Amazon MSK
Crie uma função do Lambda configurada para usar seu perfil do IAM. É possível criar sua função do Lambda usando o console.
**Para criar uma função do Lambda usando sua configuração de autenticação**
1. Abra o console do Lambda e selecione **Criar função** no cabeçalho.
1. Selecione **Criar do zero**.
1. Em **Nome da função**, forneça um nome apropriado de sua escolha.
1. Em **Runtime**, escolha a versão **Mais recente com suporte** do `Node.js` para usar o código fornecido neste tutorial.
1. Escolha **Alterar perfil de execução padrão**.
1. Selecione **Usar perfil existente**.
1. Em **Perfil existente**, selecione *lambdaAuthRole*.
Em um ambiente de produção, geralmente é necessário adicionar mais políticas ao perfil de execução da sua função do Lambda para processar de forma significativa seus eventos do Amazon MSK. Para obter mais informações sobre como adicionar políticas ao seu perfil, consulte [Adicionar ou remover permissões de identidade](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html#add-policies-console) na documentação do IAM.
## Criar um mapeamento da origem do evento para sua função do Lambda
Seu mapeamento de origem de eventos do Amazon MSK fornece ao serviço do Lambda as informações necessárias para invocar seu Lambda quando eventos apropriados do Amazon MSK ocorrem. É possível criar um mapeamento do Amazon MSK usando o console. Crie um acionador do Lambda e, em seguida, o mapeamento de origem de eventos será configurado automaticamente.
**Para criar um acionador do Lambda (e mapeamento de origem de eventos)**
1. Navegue até a página de visão geral da sua função do Lambda.
1. Na seção de visão geral da função, selecione **Adicionar acionador** no canto inferior esquerdo.
1. No menu suspenso **Selecionar uma origem**, selecione **Amazon MSK**.
1. Não defina a **autenticação**.
1. Em **Cluster do MSK**, selecione o nome do seu cluster.
1. Em **Tamanho do lote**, insira 1. Essa etapa facilita o teste desse recurso, mas não é um valor ideal em ambientes de produção.
1. Em **Nome do tópico**, forneça o nome do seu tópico do Kafka.
1. Para **ID do grupo de consumidores**, forneça o ID do seu grupo de consumidores do Kafka.
## Atualize a função do Lambda para ler seus dados de streaming
O Lambda fornece informações sobre eventos do Kafka por meio do parâmetro de método de evento. Para obter um exemplo de estrutura de um evento do Amazon MSK, consulte [Evento de exemplo](with-msk.md#msk-sample-event). Depois de entender como interpretar os eventos do Amazon MSK encaminhados pelo Lambda, você pode alterar o código da função do Lambda para usar as informações fornecidas por eles.
Forneça o seguinte código à sua função do Lambda para registrar em log o conteúdo de um evento do Amazon MSK do Lambda para fins de teste:
------
#### [ .NET ]
**SDK para .NET**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumir um evento do Amazon MSK com o Lambda usando .NET.
```
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.KafkaEvents;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace MSKLambda;
public class Function
{
/// The event for the Lambda function handler to process.
/// The ILambdaContext that provides methods for logging and describing the Lambda environment.
///
public void FunctionHandler(KafkaEvent evnt, ILambdaContext context)
{
foreach (var record in evnt.Records)
{
Console.WriteLine("Key:" + record.Key);
foreach (var eventRecord in record.Value)
{
var valueBytes = eventRecord.Value.ToArray();
var valueText = Encoding.UTF8.GetString(valueBytes);
Console.WriteLine("Message:" + valueText);
}
}
}
}
```
------
#### [ Go ]
**SDK para Go V2**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumo de um evento do Amazon MSK com o Lambda usando Go.
```
package main
import (
"encoding/base64"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(event events.KafkaEvent) {
for key, records := range event.Records {
fmt.Println("Key:", key)
for _, record := range records {
fmt.Println("Record:", record)
decodedValue, _ := base64.StdEncoding.DecodeString(record.Value)
message := string(decodedValue)
fmt.Println("Message:", message)
}
}
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**SDK para Java 2.x**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumir um evento do Amazon MSK com o Lambda usando Java.
```
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent.KafkaEventRecord;
import java.util.Base64;
import java.util.Map;
public class Example implements RequestHandler {
@Override
public Void handleRequest(KafkaEvent event, Context context) {
for (Map.Entry> entry : event.getRecords().entrySet()) {
String key = entry.getKey();
System.out.println("Key: " + key);
for (KafkaEventRecord record : entry.getValue()) {
System.out.println("Record: " + record);
byte[] value = Base64.getDecoder().decode(record.getValue());
String message = new String(value);
System.out.println("Message: " + message);
}
}
return null;
}
}
```
------
#### [ JavaScript ]
**SDK para JavaScript (v3)**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumir um evento do Amazon MSK com o Lambda usando JavaScript.
```
exports.handler = async (event) => {
// Iterate through keys
for (let key in event.records) {
console.log('Key: ', key)
// Iterate through records
event.records[key].map((record) => {
console.log('Record: ', record)
// Decode base64
const msg = Buffer.from(record.value, 'base64').toString()
console.log('Message:', msg)
})
}
}
```
Consumir um evento do Amazon MSK com o Lambda usando TypeScript.
```
import { MSKEvent, Context } from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "msk-handler-sample",
});
export const handler = async (
event: MSKEvent,
context: Context
): Promise => {
for (const [topic, topicRecords] of Object.entries(event.records)) {
logger.info(`Processing key: ${topic}`);
// Process each record in the partition
for (const record of topicRecords) {
try {
// Decode the message value from base64
const decodedMessage = Buffer.from(record.value, 'base64').toString();
logger.info({
message: decodedMessage
});
}
catch (error) {
logger.error('Error processing event', { error });
throw error;
}
};
}
}
```
------
#### [ PHP ]
**SDK para PHP**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumo de um evento do Amazon MSK com o Lambda usando PHP.
```
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handle(mixed $event, Context $context): void
{
$kafkaEvent = new KafkaEvent($event);
$this->logger->info("Processing records");
$records = $kafkaEvent->getRecords();
foreach ($records as $record) {
try {
$key = $record->getKey();
$this->logger->info("Key: $key");
$values = $record->getValue();
$this->logger->info(json_encode($values));
foreach ($values as $value) {
$this->logger->info("Value: $value");
}
} catch (Exception $e) {
$this->logger->error($e->getMessage());
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**SDK para Python (Boto3).**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumir um evento do Amazon MSK com o Lambda usando Python.
```
import base64
def lambda_handler(event, context):
# Iterate through keys
for key in event['records']:
print('Key:', key)
# Iterate through records
for record in event['records'][key]:
print('Record:', record)
# Decode base64
msg = base64.b64decode(record['value']).decode('utf-8')
print('Message:', msg)
```
------
#### [ Ruby ]
**SDK para Ruby**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumir um evento do Amazon MSK com o Lambda usando Ruby.
```
require 'base64'
def lambda_handler(event:, context:)
# Iterate through keys
event['records'].each do |key, records|
puts "Key: #{key}"
# Iterate through records
records.each do |record|
puts "Record: #{record}"
# Decode base64
msg = Base64.decode64(record['value'])
puts "Message: #{msg}"
end
end
end
```
------
#### [ Rust ]
**SDK para Rust**
Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos [Exemplos sem servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda).
Consumo de um evento do Amazon MSK com o Lambda usando o Rust.
```
use aws_lambda_events::event::kafka::KafkaEvent;
use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
use base64::prelude::*;
use serde_json::{Value};
use tracing::{info};
/// Pre-Requisites:
/// 1. Install Cargo Lambda - see https://www.cargo-lambda.info/guide/getting-started.html
/// 2. Add packages tracing, tracing-subscriber, serde_json, base64
///
/// This is the main body for the function.
/// Write your code inside it.
/// There are some code example in the following URLs:
/// - https://github.com/awslabs/aws-lambda-rust-runtime/tree/main/examples
/// - https://github.com/aws-samples/serverless-rust-demo/
async fn function_handler(event: LambdaEvent) -> Result {
let payload = event.payload.records;
for (_name, records) in payload.iter() {
for record in records {
let record_text = record.value.as_ref().ok_or("Value is None")?;
info!("Record: {}", &record_text);
// perform Base64 decoding
let record_bytes = BASE64_STANDARD.decode(record_text)?;
let message = std::str::from_utf8(&record_bytes)?;
info!("Message: {}", message);
}
}
Ok(().into())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
// required to enable CloudWatch error logging by the runtime
tracing::init_default_subscriber();
info!("Setup CW subscriber!");
run(service_fn(function_handler)).await
}
```
------
É possível fornecer código de função para o Lambda usando o console.
**Para atualizar o código da função usando o editor de código do console**
1. Abra a [página Funções](https://console.aws.amazon.com/lambda/home#/functions) do console do Lambda e selecione a função.
1. Selecione a guia **Código**.
1. No painel **Código-fonte**, selecione o arquivo de código-fonte e edite-o no editor de código integrado.
1. Na seção **DEPLOY**, escolha **Implantar** para atualizar o código da função:
![\[\]](http://docs.aws.amazon.com/pt_br/lambda/latest/dg/images/getting-started-tutorial/deploy-console.png)
## Teste sua função do Lambda para verificar se ela está conectada ao seu tópico do Amazon MSK
Agora você pode verificar se seu Lambda está sendo invocado pela origem do evento inspecionando os logs de eventos do CloudWatch.
**Para verificar se sua função do Lambda está sendo invocada**
1. Use seu host de administração do Kafka para gerar eventos do Kafka usando a CLI do `kafka-console-producer`. Para obter mais informações, consulte [Escrever alguns eventos no tópico na](https://kafka.apache.org/documentation/#quickstart_send) documentação do Kafka. Envie eventos suficientes para preencher o lote definido pelo tamanho do lote para seu mapeamento de origem de eventos definido na etapa anterior, caso contrário, o Lambda aguardará a invocação de mais informações.
1. Se sua função for executada, o Lambda escreverá o que aconteceu com o CloudWatch. No console, navegue até a página de detalhes da função do Lambda.
1. Selecione a guia **Configuration** (Configuração).
1. Na barra lateral, selecione **Ferramentas de monitoramento e operações**.
1. Identifique o **Grupo de logs do CloudWatch** em **Configuração de log**. O grupo de logs deve começar com `/aws/lambda`. Escolha o link para o grupo de logs.
1. No console do CloudWatch, inspecione os **Eventos de logs** em busca dos eventos de log que o Lambda enviou para o fluxo de logs. Identifique se há eventos de log contendo a mensagem do seu evento do Kafka, como na imagem a seguir. Se houver, você conectou com êxito uma função do Lambda ao Amazon MSK com um mapeamento de origem de eventos do Lambda.
![\[\]](http://docs.aws.amazon.com/pt_br/lambda/latest/dg/images/msk_tut_log.png)