As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
# Fontes da Amazon EventBridge Pipes
EventBridge O Pipes recebe dados de eventos de várias fontes, aplica filtros e enriquecimentos opcionais a esses dados e os envia para um destino.
Se uma fonte impõe a ordem aos eventos enviados ao EventBridge Pipes, essa ordem é mantida durante todo o processo até o destino.
Os seguintes AWS serviços podem ser especificados como fontes para EventBridge Pipes:
+ [Fluxo do Amazon DynamoDB](eb-pipes-dynamodb.md)
+ [Fluxo do Amazon Kinesis](eb-pipes-kinesis.md)
+ [Agente do Amazon MQ](eb-pipes-mq.md)
+ [Fluxo do Amazon MSK](eb-pipes-msk.md)
+ [Fila do Amazon SQS](eb-pipes-sqs.md)
+ [Fluxo do Apache Kafka](eb-pipes-kafka.md)
Ao especificar um fluxo do Apache Kafka como origem do pipe, você pode especificar um fluxo do Apache Kafka que você gerencia ou que é gerenciado por um provedor terceirizado, como:
+ [https://www.confluent.io/](https://www.confluent.io/)
+ [https://www.cloudkarafka.com/](https://www.cloudkarafka.com/)
+ [https://redpanda.com/](https://redpanda.com/)
# Fluxo do Amazon DynamoDB como origem para o EventBridge Pipes
É possível usar o EventBridge Pipes para receber registros em um fluxo do DynamoDB. Opcionalmente, é possível filtrar ou aprimorar esses registros antes de enviá-los para um destino para processamento. Há configurações específicas para o Amazon DynamoDB Streams que podem ser escolhidas ao configurar o pipe. O EventBridge Pipes mantém a ordem dos registros do fluxo de dados ao enviar esses dados para o destino.
**Importante**
A desativação de um fluxo do DynamoDB que é a origem de um pipe faz com que este pipe se torne inutilizável, mesmo que você habilite o fluxo novamente. Isto pode acontecer porque:
Não é possível interromper, iniciar ou atualizar um pipe cuja origem está desativada.
Não é possível atualizar um pipe com uma nova origem após a criação. Ao reativar um fluxo do DynamoDB, esse fluxo recebe um novo nome do recurso da Amazon (ARN) e não está mais associado ao seu pipe.
Se reativar o fluxo do DynamoDB, precisará criar um novo pipe usando o novo ARN do fluxo.
**Evento de exemplo**
O exemplo de evento a seguir mostra as informações recebidas pelo pipe. É possível usar esse evento para criar e filtrar seus padrões de eventos ou para definir a transformação de entrada. Nem todos os campos podem ser filtrados. Para mais informações sobre quais campos podem ser filtrados, consulte [Filtragem de eventos no Amazon Pipes EventBridge](eb-pipes-event-filtering.md).
```
[
{
"eventID": "1",
"eventVersion": "1.0",
"dynamodb": {
"Keys": {
"Id": {
"N": "101"
}
},
"NewImage": {
"Message": {
"S": "New item!"
},
"Id": {
"N": "101"
}
},
"StreamViewType": "NEW_AND_OLD_IMAGES",
"SequenceNumber": "111",
"SizeBytes": 26
},
"awsRegion": "us-west-2",
"eventName": "INSERT",
"eventSourceARN": "arn:aws:dynamodb:us-east-1:111122223333:table/EventSourceTable",
"eventSource": "aws:dynamodb"
},
{
"eventID": "2",
"eventVersion": "1.0",
"dynamodb": {
"OldImage": {
"Message": {
"S": "New item!"
},
"Id": {
"N": "101"
}
},
"SequenceNumber": "222",
"Keys": {
"Id": {
"N": "101"
}
},
"SizeBytes": 59,
"NewImage": {
"Message": {
"S": "This item has changed"
},
"Id": {
"N": "101"
}
},
"StreamViewType": "NEW_AND_OLD_IMAGES"
},
"awsRegion": "us-west-2",
"eventName": "MODIFY",
"eventSourceARN": "arn:aws:dynamodb:us-east-1:111122223333:table/EventSourceTable",
"eventSource": "aws:dynamodb"
}
]
```
## Fluxos de sondagem e agrupamento em lotes
O EventBridge sonda os fragmentos em seu fluxo do DynamoDB em busca de registros a uma taxa básica de quatro vezes por segundo. Quando os registros estão disponíveis, o EventBridge processa o evento e aguarda o resultado. Se o processamento tiver êxito, o EventBridge continua a sondagem até que ela receba mais registros.
Por padrão, o EventBridge invoca seu pipe assim que os registros estão disponíveis. Se o lote que o EventBridge lê da origem tiver apenas um registro nele, apenas um evento será processado. Para evitar invocar a função com um número pequeno de registros, você pode instruir à origem dos eventos para fazer o buffer dos registros por até cinco minutos, configurando uma janela de processamento de lotes. Antes de processar os eventos, o EventBridge continua a ler registros da origem de eventos até coletar um lote inteiro, até que a janela de processamento de lotes expire ou até que o lote atinja o limite de carga útil de 6 MB.
**Importante**
O pipe entregará os registros de stream do DynamoDB para o Amazon SQS pelo menos uma vez. Para garantir que nenhum registro seja perdido, sugerimos definir uma política de repetição com uma idade máxima menor que o período de retenção do stream do DynamoDB. Geralmente, os streams do DynamoDB retêm eventos por 24 horas.
Você também pode aumentar a simultaneidade processando vários lotes de cada fragmento em paralelo. O EventBridge pode processar até 10 lotes em cada fragmento simultaneamente. Se aumentar o número de lotes simultâneos por fragmento, o EventBridge ainda garantirá o processamento por ordem no nível da chave de partição.
Configure a configuração `ParallelizationFactor` para processar um fragmento de um fluxo de dados do Kinesis ou do DynamoDB com mais de uma execução do pipe simultaneamente. É possível especificar o número de lotes simultâneos que o Lambda pesquisa de um fragmento por meio de um fator de paralelização de 1 (padrão) a 10. Por exemplo, ao definir `ParallelizationFactor` como 2, você poderá ter no máximo 200 execuções simultâneas do EventBridge Pipe para processar 100 fragmentos de dados do Kinesis. Isso ajuda a aumentar o throughput de processamento quando o volume de dados é volátil e o valor de `IteratorAge` é alto. Observe que o fator de paralelização não funcionará se você estiver usando a agregação do Kinesis.
## Posição inicial de sondagem e fluxo
Esteja ciente de que a origem de fluxo durante a criação e as atualizações de pipes é, finamente, consistente.
+ Durante a criação do pipe, pode levar alguns minutos para a sondagem de eventos do fluxo iniciar.
+ Durante as atualizações de pipe para a configuração de sondagem de origem, pode levar alguns minutos para interromper e reiniciar a sondagem de eventos do fluxo.
Este comportamento significa que, se especificar `LATEST` como posição inicial do fluxo, o mapeamento da origem do evento poderá perder eventos durante a criação ou as atualizações. Para garantir que nenhum evento seja perdido, especifique a posição inicial do fluxo como `TRIM_HORIZON`.
## Gerar relatórios de falhas de itens de lote
Quando o EventBridge consome e processa dados de streaming de uma origem, ele definirá checkpoints por padrão no número mais elevado na sequência de um lote, mas somente quando o lote tiver êxito total. Para evitar o reprocessamento de todas as mensagens processadas com êxito em um lote com falha, é possível configurar o enriquecimento ou o destino para retornar um objeto indicando quais mensagens tiveram êxito e quais falharam. Isso se chama resposta parcial em lote.
Para obter mais informações, consulte [Lote com falha parcial](eb-pipes-batching-concurrency.md#pipes-partial-batch-failure).
### Condições de sucesso e falha
Se retornar qualquer um dos seguintes, o EventBridge tratará um lote como um êxito total:
+ Uma lista de `batchItemFailure` vazia
+ Uma lista de `batchItemFailure` nula
+ Uma vazia `EventResponse`
+ Uma nula `EventResponse`
Se retornar qualquer um dos seguintes, o EventBridge tratará um lote como uma falha total:
+ Uma string vazia `itemIdentifier`
+ Uma nula `itemIdentifier`
+ Um `itemIdentifier` com um nome de chave inválido
O EventBridge faz novas tentativas após falhas com base na sua estratégia de repetição.
# Stream do Amazon Kinesis como fonte para Pipes EventBridge
Você pode usar o EventBridge Pipes para receber registros em um stream de dados do Kinesis. Opcionalmente, é possível filtrar ou aprimorar esses registros antes de enviá-los para um dos destinos disponíveis para processamento. Há configurações específicas do Kinesis que podem ser escolhidas ao configurar um pipe. EventBridge O Pipes mantém a ordem dos registros do fluxo de dados ao enviar esses dados para o destino.
Uma transmissão de dados do Kinesis é um conjunto de [fragmentos](https://docs.aws.amazon.com/kinesis/latest/dev/key-concepts.html#shard). Cada fragmento contém uma sequência de registros de dados. Um **consumidor** é um aplicativo que processa os dados de um stream de dados do Kinesis. [Você pode mapear um EventBridge Pipe para um consumidor de taxa de transferência compartilhada (iterador padrão) ou para um consumidor de taxa de transferência dedicada com distribuição aprimorada.](https://docs.aws.amazon.com/kinesis/latest/dev/enhanced-consumers.html)
Para iteradores padrão, EventBridge usa o protocolo HTTP para pesquisar registros em cada fragmento em seu stream do Kinesis. O pipe compartilha o throughput de leitura com outros consumidores do fragmento.
Para minimizar a latência e maximizar o throughput de leitura, é possível criar um consumidor de fluxo de dados com distribuição avançada. Os consumidores de fluxo obtêm uma conexão dedicada para cada estilhaço que não afeta outros aplicativos que fazem leitura do fluxo. O throughput dedicado pode ajudar se você tiver muitos aplicativos lendo os mesmos dados, ou se você estiver reprocessando um fluxo com grandes registros. O Kinesis envia registros para mais de HTTP/2. EventBridge Para obter detalhes sobre fluxos de dados do Kinesis, consulte [Como ler dados do Amazon Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/dev/building-consumers.html).
**Evento de exemplo**
O exemplo de evento a seguir mostra as informações recebidas pelo pipe. É possível usar esse evento para criar e filtrar seus padrões de eventos ou para definir a transformação de entrada. Nem todos os campos podem ser filtrados. Para mais informações sobre quais campos podem ser filtrados, consulte [Filtragem de eventos no Amazon Pipes EventBridge](eb-pipes-event-filtering.md).
```
[
{
"kinesisSchemaVersion": "1.0",
"partitionKey": "1",
"sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
"data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
"approximateArrivalTimestamp": 1545084650.987,
"eventSource": "aws:kinesis",
"eventVersion": "1.0",
"eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
"eventName": "aws:kinesis:record",
"invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
"awsRegion": "us-east-2",
"eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
},
{
"kinesisSchemaVersion": "1.0",
"partitionKey": "1",
"sequenceNumber": "49590338271490256608559692540925702759324208523137515618",
"data": "VGhpcyBpcyBvbmx5IGEgdGVzdC4=",
"approximateArrivalTimestamp": 1545084711.166,
"eventSource": "aws:kinesis",
"eventVersion": "1.0",
"eventID": "shardId-000000000006:49590338271490256608559692540925702759324208523137515618",
"eventName": "aws:kinesis:record",
"invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
"awsRegion": "us-east-2",
"eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
}
]
```
## Fluxos de sondagem e agrupamento em lotes
EventBridge pesquisa fragmentos em seu stream do Kinesis para obter registros a uma taxa básica de uma vez por segundo. Quando os registros estão disponíveis, EventBridge processa o evento e aguarda o resultado. Se o processamento for bem-sucedido, EventBridge retomará a pesquisa até receber mais registros.
Por padrão, EventBridge invoca seu pipe assim que os registros estiverem disponíveis. Se o lote EventBridge lido da fonte tiver somente um registro, somente um evento será processado. Para evitar invocar a função com um número pequeno de registros, você pode instruir à origem dos eventos para fazer o buffer dos registros por até cinco minutos, configurando uma janela de processamento de lotes. Antes de processar os eventos, EventBridge continua lendo os registros da fonte até reunir um lote completo, a janela de agrupamento expirar ou o lote atingir o limite de carga útil de 6 MB.
Você também pode aumentar a simultaneidade processando vários lotes de cada fragmento em paralelo. EventBridge pode processar até 10 lotes em cada fragmento simultaneamente. Se você aumentar o número de lotes simultâneos por fragmento, EventBridge ainda garante o processamento em ordem no nível da chave de partição.
Configure a configuração `ParallelizationFactor` para processar um fragmento de um fluxo de dados do Kinesis ou do DynamoDB com mais de uma execução do pipe simultaneamente. Você pode especificar o número de lotes simultâneos que são EventBridge pesquisados a partir de um fragmento por meio de um fator de paralelização de 1 (padrão) a 10. Por exemplo, quando você define como `ParallelizationFactor` 2, você pode ter 200 execuções simultâneas de EventBridge Pipe no máximo para processar 100 fragmentos de dados do Kinesis. Isso ajuda a aumentar o throughput de processamento quando o volume de dados é volátil e o valor de `IteratorAge` é alto. Observe que o fator de paralelização não funcionará se você estiver usando a agregação do Kinesis.
## Posição inicial de sondagem e fluxo
Esteja ciente de que a origem de fluxo durante a criação e as atualizações de pipes é, finamente, consistente.
+ Durante a criação do pipe, pode levar alguns minutos para a sondagem de eventos do fluxo iniciar.
+ Durante as atualizações de pipe para a configuração de sondagem de origem, pode levar alguns minutos para interromper e reiniciar a sondagem de eventos do fluxo.
Este comportamento significa que, se especificar `LATEST` como posição inicial do fluxo, o mapeamento da origem do evento poderá perder eventos durante a criação ou as atualizações. Para garantir que nenhum evento seja perdido, especifique a posição inicial do fluxo como `TRIM_HORIZON` ou `AT_TIMESTAMP`.
## Gerar relatórios de falhas de itens de lote
Quando EventBridge consome e processa dados de streaming de uma fonte, por padrão, ele aponta para o número de sequência mais alto de um lote, mas somente quando o lote é totalmente bem-sucedido. Para evitar o reprocessamento de todas as mensagens processadas com êxito em um lote com falha, é possível configurar o enriquecimento ou o destino para retornar um objeto indicando quais mensagens tiveram êxito e quais falharam. Isso se chama resposta parcial em lote.
Para obter mais informações, consulte [Lote com falha parcial](eb-pipes-batching-concurrency.md#pipes-partial-batch-failure).
### Condições de sucesso e falha
Se você retornar qualquer um dos itens a seguir, EventBridge tratará um lote como um sucesso total:
+ Uma lista de `batchItemFailure` vazia
+ Uma lista de `batchItemFailure` nula
+ Uma `EventResponse` vazia
+ Uma `EventResponse` nula
Se você retornar qualquer um dos itens a seguir, EventBridge tratará um lote como uma falha completa:
+ Uma string `itemIdentifier` vazia
+ Uma `itemIdentifier` nula
+ Um `itemIdentifier` com um nome de chave inválido
EventBridge repete as falhas com base em sua estratégia de repetição.
# Agente de mensagens Amazon MQ como fonte no Pipes EventBridge
Você pode usar o EventBridge Pipes para receber registros de um agente de mensagens do Amazon MQ. Opcionalmente, é possível filtrar ou aprimorar esses registros antes de enviá-los para um dos destinos disponíveis para processamento. Há configurações específicas para o Amazon MQ que você pode escolher ao configurar um canal. EventBridge O Pipes mantém a ordem dos registros do agente de mensagens ao enviar esses dados para o destino.
O Amazon MQ é um serviço gerenciado de agente de mensagem para o [Apache ActiveMQ](https://activemq.apache.org/) e o [RabbitMQ](https://www.rabbitmq.com/). Um agente de mensagens habilita aplicações de software e componentes para se comunicarem usando diferentes linguagens de programação, sistemas operacionais e protocolos de mensagens formais com destinos de eventos de tópicos ou filas.
O Amazon MQ também pode gerenciar instâncias do Amazon Elastic Compute Cloud (Amazon EC2) em seu nome instalando agentes do ActiveMQ ou RabbitMQ. Depois que um agente é instalado, ele fornece diferentes topologias de rede e outras necessidades de infraestrutura para suas instâncias.
A origem do Amazon MQ tem as seguintes restrições de configuração:
+ **Conta cruzada** — EventBridge não suporta processamento entre contas. Você não pode usar EventBridge para processar registros de um agente de mensagens do Amazon MQ que esteja em uma conta diferente AWS .
+ **Autenticação** [— Para o ActiveMQ, somente o ActiveMQ é suportado. SimpleAuthenticationPlugin](https://activemq.apache.org/security#simple-authentication-plugin) Para RabbitMQ, somente o[PLAIN](https://www.rabbitmq.com/access-control.html#mechanisms)Mecanismo de autenticação é compatível. Para gerenciar credenciais, use AWS Secrets Manager. Para obter mais informações sobre a autenticação do ActiveMQ, consulte [Como integrar agentes do ActiveMQ com o LDAP](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/security-authentication-authorization.html) no Guia do desenvolvedor do Amazon MQ.
+ **Cota de conexão**: os agentes têm um número máximo de conexões permitidas por cada protocolo de nível de transmissão de dados. Essa cota é baseada no tipo de instância do agente. Para obter mais informações, consulte a seção [Agentes](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/amazon-mq-limits.html#broker-limits) do **\$1Cotas no Amazon MQ\$1** no Guia do desenvolvedor do Amazon MQ.
+ **Conectividade**: é possível criar agentes em uma nuvem privada virtual (VPC) pública ou privada. Para uso privado VPCs, seu canal precisa acessar a VPC para receber mensagens.
+ **Destinos de eventos**: somente destinos de fila são compatíveis. No entanto, é possível usar um tópico virtual, que se comporta como um tópico internamente e como uma fila externamente enquanto interage com seus pipes. Para obter mais informações, consulte [Destinos virtuais](https://activemq.apache.org/virtual-destinations) no site do Apache ActiveMQ e [Hosts virtuais](https://www.rabbitmq.com/vhosts.html) nio site do the RabbitMQ.
+ **Topologia de rede**: para o ActiveMQ, somente um agente de instância única ou em espera é aceito por pipe. No RabbitMQ, apenas um agente de instância única ou implantação de cluster é aceito por cada pipe. Os agentes de instância única requerem um endpoint de failover. Para obter mais informações sobre esses modos de implantação do agente, consulte [Arquitetura do agente ativo do MQ](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/amazon-mq-broker-architecture.html) e [Arquitetura do agente de MQ do Rabbit](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/rabbitmq-broker-architecture.html) noGuia do desenvolvedor do Amazon MQ.
+ **Protocolos**: os protocolos compatíveis dependem do tipo de integração do Amazon MQ.
+ Para integrações com o ActiveMQ EventBridge , usa OpenWire/Java o protocolo Message Service (JMS) para consumir mensagens. O consumo de mensagens não é suportado em nenhum outro protocolo. EventBridge suporta somente as `BytesMessage` operações `TextMessage` e dentro do protocolo JMS. Para obter mais informações sobre o OpenWire protocolo, consulte o [OpenWire](https://activemq.apache.org/openwire.html)site do Apache ActiveMQ.
+ Para integrações com o RabbitMQ, EventBridge usa o protocolo AMQP 0-9-1 para consumir mensagens. Nenhum outro protocolo é compatível para o consumo de mensagens. Para obter mais informações sobre a implementação do protocolo AMQP 0-9-1 pelo RabbitMQ, consulte[AMQP 0-9-1 Guia de referência completo](https://www.rabbitmq.com/amqp-0-9-1-reference.html)no site do RabbitMQ.
EventBridge suporta automaticamente as versões mais recentes do ActiveMQ e do RabbitMQ suportadas pelo Amazon MQ. Para obter as versões compatíveis mais recentes, consulte [Notas de versão do Amazon MQ](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/amazon-mq-release-notes.html) no Guia do desenvolvedor do Amazon MQ.
**nota**
Por padrão, o Amazon MQ tem uma janela de manutenção semanal para agentes. Durante essa janela de tempo, os agentes não estão disponíveis. Para corretores sem espera, não EventBridge processará mensagens até que a janela termine.
**Eventos de exemplo**
O exemplo de evento a seguir mostra as informações recebidas pelo pipe. É possível usar esse evento para criar e filtrar seus padrões de eventos ou para definir a transformação de entrada. Nem todos os campos podem ser filtrados. Para mais informações sobre quais campos podem ser filtrados, consulte [Filtragem de eventos no Amazon Pipes EventBridge](eb-pipes-event-filtering.md).
**ActiveMQ**
```
[
{
"eventSource": "aws:amq",
"eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",
"messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1---mq---us-west-2.amazonaws.com.rproxy.govskope.ca-37557-1234520418293-4:1:1:1:1",
"messageType": "jms/text-message",
"data": "QUJDOkFBQUE=",
"connectionId": "myJMSCoID",
"redelivered": false,
"destination": {
"physicalname": "testQueue"
},
"timestamp": 1598827811958,
"brokerInTime": 1598827811958,
"brokerOutTime": 1598827811959
},
{
"eventSource": "aws:amq",
"eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",
"messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1---mq---us-west-2.amazonaws.com.rproxy.govskope.ca-37557-1234520418293-4:1:1:1:1",
"messageType": "jms/bytes-message",
"data": "3DTOOW7crj51prgVLQaGQ82S48k=",
"connectionId": "myJMSCoID1",
"persistent": false,
"destination": {
"physicalname": "testQueue"
},
"timestamp": 1598827811958,
"brokerInTime": 1598827811958,
"brokerOutTime": 1598827811959
}
]
```
**RabbitMQ**
```
[
{
"eventSource": "aws:rmq",
"eventSourceArn": "arn:aws:mq:us-west-2:111122223333:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8",
"eventSourceKey": "pizzaQueue::/",
"basicProperties": {
"contentType": "text/plain",
"contentEncoding": null,
"headers": {
"header1": {
"bytes": [
118,
97,
108,
117,
101,
49
]
},
"header2": {
"bytes": [
118,
97,
108,
117,
101,
50
]
},
"numberInHeader": 10
},
"deliveryMode": 1,
"priority": 34,
"correlationId": null,
"replyTo": null,
"expiration": "60000",
"messageId": null,
"timestamp": "Jan 1, 1970, 12:33:41 AM",
"type": null,
"userId": "AIDACKCEVSQ6C2EXAMPLE",
"appId": null,
"clusterId": null,
"bodySize": 80
},
"redelivered": false,
"data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ=="
}
]
```
## Grupo de consumidores
Para interagir com o Amazon MQ, EventBridge crie um grupo de consumidores que pode ler seus corretores do Amazon MQ. O grupo de consumidores é criado com o mesmo ID que um UUID de pipes.
Para fontes do Amazon MQ, agrupa registros em EventBridge lotes e os envia para sua função em uma única carga útil. Para controlar o comportamento, você pode configurar a janela e o tamanho do lote. EventBridge extrai mensagens até que ocorra uma das seguintes situações:
+ Os registros processados atingem o tamanho máximo da carga útil de 6 MB.
+ A janela de processamento de lotes expira.
+ O número de registros atinge o tamanho total do lote.
EventBridge converte seu lote em uma única carga útil e, em seguida, invoca sua função. As mensagens não são persistentes nem desserializadas. Em vez disso, o grupo de consumidores os recupera como um BLOB de bytes. Em seguida, o base64 os codifica em uma carga útil JSON. Se o pipe retornar um erro para qualquer uma das mensagens em um lote, EventBridge tente novamente o lote inteiro de mensagens até que o processamento seja bem-sucedido ou as mensagens expirem.
## Configuração de rede
Por padrão, os agentes do Amazon MQ são criados com o a sinalização `PubliclyAccessible` definida como falsa. Somente quando `PubliclyAccessible` é definido como verdadeiro que o agente recebe um endereço IP público. Para ter acesso total ao seu pipe, seu agente deve usar um endpoint público ou fornecer acesso à VPC.
Se seu agente do Amazon MQ não estiver acessível ao público, EventBridge deverá ter acesso aos recursos da Amazon Virtual Private Cloud (Amazon VPC) associados ao seu corretor.
+ Para acessar a VPC de seus corretores do Amazon MQ EventBridge , você pode usar o acesso de saída à Internet para as sub-redes de sua fonte. Para sub-redes públicas, esse deve ser um[ gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) gerenciado. Para sub-redes privadas, pode ser um gateway NAT ou o seu próprio NAT. Certifique-se de que o NAT tenha um endereço IP público e possa se conectar à Internet.
+ EventBridge O Pipes também suporta a entrega de eventos [AWS PrivateLink](https://aws.amazon.com/privatelink/), permitindo que você envie eventos de uma fonte de eventos localizada em um Amazon Virtual Private Cloud (Amazon VPC) para um destino do Pipes sem atravessar a Internet pública. Você pode usar o Pipes para pesquisar Amazon Managed Streaming for Apache Kafka (Amazon MSK), Apache Kafka autogerenciado e Amazon MQ fontes residentes em uma sub-rede privada sem a necessidade de implantar um gateway de internet, configurar regras de firewall ou configurar servidores proxy.
Para configurar um endpoint da VPC, consulte [Create a VPC endpoint](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint-aws) no *Guia do usuário do AWS PrivateLink *. Para o nome do serviço, selecione `com.amazonaws.region.pipes-data`.
Configure os grupos de segurança da Amazon VPC com as seguintes regras (no mínimo):
+ Regras de entrada: permitir todo tráfego na porta do agente do Amazon MQ para os grupos de segurança especificados para sua origem.
+ Regras de saída: permitir todo o tráfego na porta 443 para todos os destinos. Permita todo tráfego na porta do agente do Amazon MQ para os grupos de segurança especificados para sua origem.
As portas do agente incluem:
+ 9092 para texto simples
+ 9094 para TLS
+ 9096 para SASL
+ 9098 para o IAM
**nota**
Sua configuração da Amazon VPC pode ser detectada pela [API do Amazon MQ](https://docs.aws.amazon.com/amazon-mq/latest/api-reference/resources.html). Não é necessário defini-la durante a configuração.
# Amazon Managed Streaming for Apache: tópico Kafka como fonte no Pipes EventBridge
[Você pode usar o EventBridge Pipes para receber registros de um tópico do [Amazon Managed Streaming for Apache Kafka (Amazon MSK), inclusive do tipo de cluster Amazon](https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html) MSK Serverless.](https://aws.amazon.com/msk/features/msk-serverless/) Opcionalmente, é possível filtrar ou aprimorar esses registros antes de enviá-los para um dos destinos disponíveis para processamento. Há configurações específicas para o Amazon MSK que você pode escolher ao configurar um canal. EventBridge O Pipes mantém a ordem dos registros do agente de mensagens ao enviar esses dados para o destino.
O Amazon MSK é um serviço totalmente gerenciado que pode ser usado na criação e a execução de aplicações que usam o Apache Kafka para processar dados de streaming. O Amazon MSK simplifica a configuração, a escalabilidade e o gerenciamento de clusters que executam o Apache Kafka. Com o Amazon MSK, você pode configurar seu aplicativo para várias zonas de disponibilidade e para segurança com AWS Identity and Access Management (IAM). O Amazon MSK é compatível com várias versões de código aberto do Kafka.
O Amazon MSK como fonte opera de forma semelhante ao uso do Amazon Simple Queue Service (Amazon SQS) ou do Amazon Kinesis. EventBridgepesquisa internamente novas mensagens da origem e, em seguida, invoca o alvo de forma síncrona. EventBridge lê as mensagens em lotes e as fornece à sua função como uma carga útil de eventos. O tamanho máximo do lote é configurável. (O valor padrão é de 100 mensagens.)
Para fontes baseadas no Apache Kafka, EventBridge suporta parâmetros de controle de processamento, como janelas de lote e tamanho do lote.
EventBridge lê as mensagens sequencialmente para cada partição. Depois de EventBridge processar cada lote, ele confirma os deslocamentos das mensagens nesse lote. Se o destino do canal retornar um erro para qualquer uma das mensagens em um lote, EventBridge tente novamente o lote inteiro de mensagens até que o processamento seja bem-sucedido ou as mensagens expirem.
EventBridge envia o lote de mensagens no evento quando invoca o destino. 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.
**Eventos de exemplo**
O exemplo de evento a seguir mostra as informações recebidas pelo pipe. É possível usar esse evento para criar e filtrar seus padrões de eventos ou para definir a transformação de entrada. Nem todos os campos podem ser filtrados. Para mais informações sobre quais campos podem ser filtrados, consulte [Filtragem de eventos no Amazon Pipes EventBridge](eb-pipes-event-filtering.md).
```
[
{
"eventSource": "aws:kafka",
"eventSourceArn": "arn:aws:kafka:sa-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
"eventSourceKey": "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
]
}
]
}
]
```
## Posição inicial de sondagem e fluxo
Esteja ciente de que a origem de fluxo durante a criação e as atualizações de pipes é, finamente, consistente.
+ Durante a criação do pipe, pode levar alguns minutos para a sondagem de eventos do fluxo iniciar.
+ Durante as atualizações de pipe para a configuração de sondagem de origem, pode levar alguns minutos para interromper e reiniciar a sondagem de eventos do fluxo.
Este comportamento significa que, se especificar `LATEST` como posição inicial do fluxo, o mapeamento da origem do evento poderá perder eventos durante a criação ou as atualizações. Para garantir que nenhum evento seja perdido, especifique a posição inicial do fluxo como `TRIM_HORIZON`.
## Autenticação de cluster do MSK
EventBridge precisa de permissão para acessar o cluster Amazon MSK, recuperar registros e realizar outras tarefas. O Amazon MSK oferece suporte a várias opções para controlar o acesso do cliente ao cluster do MSK. Para obter mais informações quando cada método de autenticação é usado, consulte [Como EventBridge escolhe um corretor de bootstrap](#pipes-msk-bootstrap-brokers).
**Topics**
+ [Acesso não autenticado](#pipes-msk-permissions-none)
+ [Autenticação SASL/SCRAM](#pipes-msk-permissions-add-secret)
+ [Autenticação baseada em função do IAM](#pipes-msk-permissions-iam-policy)
+ [Autenticação TLS mútua](#pipes-msk-permissions-mTLS)
+ [Configurar o segredo de mTLS](#smaa-auth-secret)
+ [Como EventBridge escolhe um corretor de bootstrap](#pipes-msk-bootstrap-brokers)
### Acesso não autenticado
É recomendado usar somente o acesso não autenticado para desenvolvimento. O acesso não autenticado só funcionará se a autenticação baseada em perfis do IAM estiver desativada para o cluster.
### Autenticação SASL/SCRAM
O Amazon MSK oferece suporte à autenticação Simple Authentication e ao Security Layer/Salted Challenge Response Authentication Mechanism (SASL/SCRAM) com criptografia Transport Layer Security (TLS). EventBridge Para se conectar ao cluster, você armazena as credenciais de autenticação (credenciais de login) em um segredo. AWS Secrets Manager
Para obter mais informações sobre o uso do Secrets Manager, consulte [Autenticação de nome de usuário e senha com o AWS Secrets Manager](https://docs.aws.amazon.com/msk/latest/developerguide/msk-password.html), no *Guia do Desenvolvedor do Amazon Managed Streaming para Apache Kafka*.
O Amazon MSK não oferece suporte à SASL/PLAIN autenticação.
### Autenticação baseada em função do IAM
Use o IAM para autenticar a identidade dos clientes que se conectam ao cluster do MSK. Se a autenticação do IAM estiver ativa no seu cluster MSK e você não fornecer um segredo para a autenticação, EventBridge automaticamente usará a autenticação do IAM como padrão. Para criar e implantar políticas baseadas em funções ou usuários do IAM, use o console ou a API do IAM. Para obter mais informações, consulte [IAM access control](https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html) (Controle de acesso do IAM) no *Guia do desenvolvedor do Amazon Managed Streaming for Apache Kafka*.
EventBridge Para permitir a conexão com o cluster MSK, a leitura de registros e a execução de outras ações necessárias, adicione as seguintes permissões à função de execução de seus pipes.
------
#### [ 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:123456789012:cluster/cluster-name/cluster-uuid",
"arn:aws:kafka:us-east-1:123456789012:topic/cluster-name/cluster-uuid/topic-name",
"arn:aws:kafka:us-east-1:123456789012:group/cluster-name/cluster-uuid/consumer-group-id"
]
}
]
}
```
------
É possível definir o escopo dessas permissões para abranger clusters, tópicos e grupos específicos. Para obter mais informações, consulte [Amazon MSK Kafka actions](https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html#kafka-actions) (Ações do Amazon MSK Kafka) no *Guia do desenvolvedor Amazon Managed Streaming for Apache Kafka*.
### Autenticação TLS mútua
O TLS mútuo (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, e o servidor envia um certificado ao cliente para que o cliente verifique o servidor.
Para o Amazon MSK, EventBridge atua como cliente. Você configura um certificado de cliente (como um segredo no Secrets Manager) para se autenticar EventBridge com os corretores em seu cluster MSK. O certificado do servidor deve ser assinado por uma autoridade de certificação (CA) no armazenamento de confiança da . O cluster MSK envia um certificado de servidor para EventBridge autenticar os corretores com. EventBridge O certificado do servidor deve ser assinado por uma CA que esteja no armazenamento AWS confiável.
O Amazon MSK não oferece suporte a certificados de servidor autoassinados, porque todos os corretores do Amazon MSK usam certificados [públicos assinados](https://docs.aws.amazon.com/msk/latest/developerguide/msk-encryption.html) pelo [Amazon Trust Services CAs](https://www.amazontrust.com/repository/), que EventBridge é confiável por padrão.
Para obter mais informações sobre o mTLS para o Amazon MSK, consulte [Mutual TLS Authentication](https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html) (Autenticação TLS mútua) no *Guia do desenvolvedor Amazon Managed Streaming for Apache Kafka*.
### 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**
EventBridge suporta [PBES1](https://datatracker.ietf.org/doc/html/rfc2898/#section-6.1)(mas não PBES2) os algoritmos de criptografia de chave privada.
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-----"
}
```
### Como EventBridge escolhe um corretor de bootstrap
EventBridge escolhe um [agente 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 em seu cluster e se você fornece um segredo para autenticação. Se você fornecer um segredo para mTLS ou SASL/SCRAM, escolhe EventBridge automaticamente esse método de autenticação. Se você não fornecer um segredo, EventBridge escolha o método de autenticação mais forte que esteja ativo no seu cluster. A seguir está a ordem de prioridade na qual EventBridge seleciona um corretor, da autenticação mais forte para a mais fraca:
+ mTLs (segredo fornecido para mTLs)
+ SASL/SCRAM (secret provided for 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 não EventBridge conseguir se conectar ao tipo de corretor mais seguro, ele não tentará se conectar a um tipo de corretor diferente (mais fraco). Se você quiser EventBridge escolher um tipo de agente mais fraco, desative todos os métodos de autenticação mais fortes em seu cluster.
## Configuração de rede
EventBridge deve ter acesso aos recursos da Amazon Virtual Private Cloud (Amazon VPC) associados ao seu cluster Amazon MSK.
+ Para acessar a VPC do seu cluster Amazon MSK, EventBridge você pode usar o acesso de saída à Internet para as sub-redes da sua fonte. Para sub-redes privadas, pode ser um gateway NAT ou o seu próprio NAT. Certifique-se de que o NAT tenha um endereço IP público e possa se conectar à Internet. Para sub-redes públicas, você deve usar endpoints da VPC (explicado abaixo).
+ EventBridge O Pipes também suporta a entrega de eventos [AWS PrivateLink](https://aws.amazon.com/privatelink/), permitindo que você envie eventos de uma fonte de eventos localizada em um Amazon Virtual Private Cloud (Amazon VPC) para um destino do Pipes sem atravessar a Internet pública. Você pode usar o Pipes para pesquisar Amazon Managed Streaming for Apache Kafka (Amazon MSK), Apache Kafka autogerenciado e Amazon MQ fontes residentes em uma sub-rede privada sem a necessidade de implantar um gateway de internet, configurar regras de firewall ou configurar servidores proxy. Você também pode usar endpoints da VPC para atender a entregas provenientes de clusters do Kafka em sub-redes públicas.
Para configurar um endpoint da VPC, consulte [Create a VPC endpoint](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint-aws) no *Guia do usuário do AWS PrivateLink *. Para o nome do serviço, selecione `com.amazonaws.region.pipes-data`.
Configure os grupos de segurança da Amazon VPC com as seguintes regras (no mínimo):
+ Regras de entrada: permitir todo tráfego na porta do agente do Amazon MSK para os grupos de segurança especificados para sua origem.
+ Regras de saída: permitir todo o tráfego na porta 443 para todos os destinos. Permita todo tráfego na porta do agente do Amazon MSK para os grupos de segurança especificados para sua origem.
As portas do agente incluem:
+ 9092 para texto simples
+ 9094 para TLS
+ 9096 para SASL
+ 9098 para o IAM
**nota**
Sua configuração da Amazon VPC pode ser detectada pela [API do Amazon MSK](https://docs.aws.amazon.com/msk/1.0/apireference/resources.html). Não é necessário defini-la durante a configuração.
## ID de grupo de consumidores personalizável
Ao configurar o Apache Kafka como uma origem de eventos, é possível especificar um ID de grupo de consumidores. Este ID de grupo de consumidores é um identificador existente para o grupo de consumidores do Apache Kafka no qual deseja que a função do Lambda ingresse. Você pode usar esse recurso para migrar qualquer configuração contínua de processamento de registros do Apache Kafka de outros consumidores para o. EventBridge
Se você especificar um ID de grupo de consumidores e houver outros pesquisadores ativos dentro desse grupo de consumidores, o Apache Kafka distribuirá mensagens entre todos os consumidores. Em outras palavras, EventBridge não recebe todas as mensagens do tópico Apache Kafka. Se você quiser EventBridge lidar com todas as mensagens do tópico, desative qualquer outra pesquisa nesse grupo de consumidores.
Além disso, se você especificar um ID de grupo de consumidores e o Apache Kafka encontrar um grupo de consumidores válido existente com o mesmo ID, EventBridge ignorará o `StartingPosition` parâmetro do seu canal. Em vez disso, EventBridge começa a processar os registros de acordo com a compensação comprometida do grupo de consumidores. Se você especificar um ID de grupo de consumidores e o Apache Kafka não conseguir encontrar um grupo de consumidores existente, então EventBridge configura sua fonte com o especificado. `StartingPosition`
O ID do grupo de consumidores que você especificar deverá ser exclusivo entre todas as origens de eventos do Apache Kafka. Após criar um pipe com o ID do grupo de consumidores especificado, não poderá atualizar este valor.
## Ajuste de escala automático da origem do Amazon MSK
Quando você cria inicialmente uma fonte do Amazon MSK, EventBridge aloca um consumidor para processar todas as partições no tópico Apache Kafka. Cada consumidor conta com vários processadores em execução em paralelo para lidar com um aumento de workloads. Além disso, aumenta ou diminui EventBridge automaticamente o número de consumidores, com base na carga de trabalho. Para preservar a ordenação de mensagens em cada partição, o número máximo de consumidores é um consumidor por partição no tópico.
Em intervalos de um minuto, EventBridge avalia o atraso de compensação do consumidor de todas as partições no tópico. Se o atraso for muito alto, a partição está recebendo mensagens mais rápido do que EventBridge pode processá-las. Se necessário, EventBridge adiciona ou remove consumidores do tópico. O processo de escalabilidade de adicionar ou remover consumidores ocorre em até três minutos após a avaliação.
Se sua meta estiver sobrecarregada, EventBridge reduz o número de consumidores. Essa ação reduz a workload no pipe, reduzindo o número de mensagens que os consumidores podem recuperar e enviar para o pipe.
# Fluxos do Apache Kafka como origem no EventBridge Pipes
O Apache Kafka é uma plataforma de streaming de eventos de código aberto que é compatível com workloads, como pipelines de dados e análises de streaming. É possível usar o [Amazon Managed Streaming for Apache Kafka](eb-pipes-msk.md) (Amazon MSK) ou um cluster do Apache Kafka autogerenciado. Na terminologia da AWS, um cluster *autogerenciado* se refere a qualquer cluster do Apache Kafka não hospedado pela AWS. Isso inclui tanto os clusters que você gerencia quanto os hospedados por um provedor terceirizado, como [https://www.confluent.io/](https://www.confluent.io/), [https://www.cloudkarafka.com/](https://www.cloudkarafka.com/) ou [https://redpanda.com/](https://redpanda.com/).
Para obter mais informações sobre outras opções da AWS para hospedagem de cluster, consulte [Best Practices for Running Apache Kafka on AWS](https://aws.amazon.com/blogs/big-data/best-practices-for-running-apache-kafka-on-aws/) no *AWS Big Data Blog*.
O Apache Kafka como origem opera de forma semelhante ao uso do Amazon Simple Queue Service (Amazon SQS) ou do Amazon Kinesis. O EventBridge pesquisa internamente por novas mensagens da origem do evento e invoca, de forma síncrona, o destino. O EventBridge lê as mensagens em lotes e fornece estas para a sua função como uma carga útil de eventos. O tamanho máximo do lote é configurável. (O valor padrão é de 100 mensagens.)
Para origens de eventos baseadas em Apache Kafka, o EventBridge oferece suporte a parâmetros de controle de processamento, como janelas de processamento de lotes e tamanho do lote.
O EventBridge envia o lote de mensagens no parâmetro de evento ao invocar seu pipe. O payload do evento contém uma matriz de mensagens. Cada item da matriz contém detalhes do tópico do Apache Kafka e do identificador de partição do Apache Kafka, juntamente com um carimbo de data/hora e uma mensagem codificada em base64.
**Eventos de exemplo**
O exemplo de evento a seguir mostra as informações recebidas pelo pipe. É possível usar esse evento para criar e filtrar seus padrões de eventos ou para definir a transformação de entrada. Nem todos os campos podem ser filtrados. Para mais informações sobre quais campos podem ser filtrados, consulte [Filtragem de eventos no Amazon Pipes EventBridge](eb-pipes-event-filtering.md).
```
[
{
"eventSource": "SelfManagedKafka",
"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",
"eventSourceKey": "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
]
}
]
}
]
```
## Autenticação de clusters do Apache Kafka
O EventBridge Pipes é compatível com vários métodos para autenticar com seu cluster Apache Kafka autogerenciado. Configure o cluster Apache Kafka para utilizar um dos métodos de autenticação compatíveis. Para obter mais informações sobre a segurança do Apache Kafka, consulte a seção [Segurança da documentação do Apache Kafka](http://kafka.apache.org/documentation.html#security).
### Acesso à VPC
Se estiver usando um ambiente do Apache Kafka autogerenciado em que apenas os usuários do Apache Kafka de sua VPC têm acesso a seus agentes do Apache Kafka, será necessário configurar a Amazon Virtual Private Cloud (Amazon VPC) na origem do Apache Kafka.
### Autenticação SASL/SCRAM
O EventBridge Pipes é compatível com autenticação Simple Authentication e Security Layer/Salted Challenge Response Authentication Mechanism (SASL/SCRAM) com criptografia Transport Layer Security (TLS). O EventBridge Pipes envia as credenciais criptografadas para autenticar com o cluster. Para obter mais informações sobre a autenticação do SASL/SCRAM, consulte [RFC 5802](https://tools.ietf.org/html/rfc5802).
O EventBridge Pipes é compatível com autenticação SASL/PLAIN com criptografia TLS. Com a autenticação SASL/PLAIN, o EventBridge envia credenciais como texto simples (não criptografado) ao servidor.
Para a autenticação SASL, armazene as credenciais de login como um segredo no AWS Secrets Manager.
### Autenticação TLS mútua
O TLS mútuo (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, e o servidor envia um certificado ao cliente para que o cliente verifique o servidor.
No Apache Kafka autogerenciado, o EventBridge atua como cliente. É configurado um certificado de cliente (como um segredo no Secrets Manager) para autenticar o EventBridge Pipes com os agentes do Apache Kafka. O certificado do servidor deve ser assinado por uma autoridade de certificação (CA) no armazenamento de confiança da .
O cluster do Apache Kafka envia um certificado de servidor ao EventBridge Pipes para autenticar os agentes do Apache Kafka com o EventBridge Pipes. O certificado do servidor pode ser um certificado CA público ou um certificado de CA privado/autoassinado. O certificado CA público deve ser assinado por uma autoridade de certificação (CA) que esteja no repositório de confiança do EventBridge Pipes. Para um certificado CA privado/autoassinado, configure o certificado CA raiz do servidor (como um segredo no Secrets Manager). O EventBridge Pipes usa o certificado raiz para verificar os agentes do Apache Kafka.
Para obter mais informações sobre o mTLS, consulte [Introdução à autenticação TLS mútua para o Amazon MSK como origem](https://aws.amazon.com/blogs/compute/introducing-mutual-tls-authentication-for-amazon-msk-as-an-event-source).
### Configurar o segredo do certificado do cliente
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 EventBridge Pipes é compatível com 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 de 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-----"
}
```
### Configurar o segredo do certificado CA raiz do servidor
Este segredo é criado se seus agentes do Apache Kafka usarem criptografia TLS com certificados assinados por uma CA privada. É possível usar criptografia TLS para autenticação de VPC, SASL/SCRAM, SASL/PLAIN ou mTLS.
O segredo do certificado CA raiz do servidor requer um campo que contenha o certificado CA raiz do agente do Apache Kafka no formato PEM. O exemplo a seguir exibe a estrutura do segredo.
```
{
"certificate": "-----BEGIN CERTIFICATE-----
MIID7zCCAtegAwIBAgIBADANBgkqhkiG9w0BAQsFADCBmDELMAkGA1UEBhMCVVMx
EDAOBgNVBAgTB0FyaXpvbmExEzARBgNVBAcTClNjb3R0c2RhbGUxJTAjBgNVBAoT
HFN0YXJmaWVsZCBUZWNobm9sb2dpZXMsIEluYy4xOzA5BgNVBAMTMlN0YXJmaWVs
ZCBTZXJ2aWNlcyBSb290IENlcnRpZmljYXRlIEF1dG...
-----END CERTIFICATE-----"
```
## Configuração de rede
Se estiver usando um ambiente do Apache Kafka autogerenciado que usa conectividade de VPC privada, o EventBridge deverá ter acesso aos recursos da Amazon Virtual Private Cloud (Amazon VPC) associados aos seus agentes do Apache Kafka.
+ Para acessar a VPC do seu cluster do Apache Kafka, o EventBridge pode usar o acesso de saída à internet para as sub-redes da sua origem. Para sub-redes privadas, pode ser um gateway NAT ou o seu próprio NAT. Certifique-se de que o NAT tenha um endereço IP público e possa se conectar à Internet. Para sub-redes públicas, você deve usar endpoints da VPC (explicado abaixo).
+ O EventBridge Pipes também aceita a entrega de eventos por meio do [AWS PrivateLink](https://aws.amazon.com/privatelink/), permitindo que você envie eventos de uma origem de eventos localizada em uma Amazon Virtual Private Cloud (Amazon VPC) a um destino do Pipes sem atravessar a internet pública. Você pode usar o Pipes para sondar de origens do Amazon Managed Streaming for Apache Kafka (Amazon MSK), do Apache Kafka autogerenciado e do Amazon MQ residentes em uma sub-rede privada sem a necessidade de implantar um gateway da internet, configurar regras de firewall ou configurar servidores proxy. Você também pode usar endpoints da VPC para atender a entregas provenientes de clusters do Kafka em sub-redes públicas.
Para configurar um endpoint da VPC, consulte [Create a VPC endpoint](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint-aws) no *Guia do usuário do AWS PrivateLink*. Para o nome do serviço, selecione `com.amazonaws.region.pipes-data`.
Configure os grupos de segurança da Amazon VPC com as seguintes regras (no mínimo):
+ Regras de entrada: permitir todo tráfego na porta do agente do Apache Kafka para os grupos de segurança especificados para sua origem.
+ Regras de saída: permitir todo o tráfego na porta 443 para todos os destinos. Permita todo tráfego na porta do agente do Apache Kafka para os grupos de segurança especificados para sua origem.
As portas do agente incluem:
+ 9092 para texto simples
+ 9094 para TLS
+ 9096 para SASL
+ 9098 para o IAM
## Ajuste de escala automático do consumidor com origens do Apache Kafka
Ao criar inicialmente uma origem de eventos do Apache Kafka, o EventBridge aloca um consumidor para processar todas as partições no tópico do Apache Kafka. Cada consumidor conta com vários processadores em execução em paralelo para lidar com um aumento de workloads. O EventBridge aumenta ou reduz a escala verticalmente automaticamente do número de consumidores com base na workload. Para preservar a ordenação de mensagens em cada partição, o número máximo de consumidores é um consumidor por partição no tópico.
A cada um minuto, o EventBridge avalia o atraso de compensação do consumidor de todas as partições do tópico. Se o atraso for muito alto, a partição está recebendo mensagens mais rápido do que o EventBridge pode processá-las. Se necessário, o EventBridge adiciona ou remove os consumidores do tópico. O processo de escalabilidade de adicionar ou remover consumidores ocorre em até três minutos após a avaliação.
Se seu destino estiver sobrecarregado, o EventBridge reduz o número de consumidores. Essa ação reduz a workload na função, reduzindo o número de mensagens que os consumidores podem recuperar e enviar para a função.
# Amazon Simple Queue Service como origem no EventBridge Pipes
É possível usar o EventBridge Pipes para receber registros de um agente de mensagens do Amazon SQS. Opcionalmente, é possível filtrar ou aprimorar esses registros antes de enviá-los para um destino disponível para processamento.
É possível usar um pipe para processar mensagens em uma fila do Amazon Simple Queue Service (Amazon SQS). O EventBridge Pipes é compatível com [filas padrão](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues.html) e [filas de primeiro a chegar, primeiro a sair (FIFO)](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html). Com o Amazon SQS, você pode descarregar tarefas de um componente do aplicativo enviando-as a uma fila e processando-as de forma assíncrona.
O EventBridge sonda a fila e invoca seu pipe de forma síncrona com um evento que contém mensagens da fila. O EventBridge lê mensagens em lotes e invoca seu pipe uma vez para cada lote. Quando sua função processa com êxito um lote, o EventBridge exclui suas mensagens da fila.
Por padrão, o EventBridge pesquisará até dez mensagens em sua fila simultaneamente e envia esse lote para seu pipe. Para evitar invocar a função com um número pequeno de registros, você pode instruir à origem dos eventos para fazer o buffer dos registros por até cinco minutos, configurando uma janela de lote. Antes de invocar o pipe, o EventBridge continua a pesquisar mensagens da fila padrão do Amazon SQS até que uma dessas coisas ocorra:
+ A janela de lotes expira.
+ A cota de tamanho da carga útil de invocação foi atingida.
+ O tamanho máximo do lote configurado foi atingido.
**nota**
Se estiver usando uma janela em lote e sua fila do Amazon SQS contiver um volume baixo de tráfego, o EventBridge poderá esperar até 20 segundos antes de invocar sua função. Isto será válido mesmo se definir uma janela de lote inferior a 20 segundos. Para as filas FIFO, os registros contêm atributos adicionais relacionados a desduplicação e sequenciamento.
Quando o EventBridge faz a leitura de um lote, as mensagens permanecem na fila, mas são ocultadas durante o [tempo limite de visibilidade da fila](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html). Se seu pipe processar com êxito um lote, o EventBridge excluirá as mensagens da fila. Por padrão, se seu pipe se deparar com um erro durante o processamento de um lote, todas as mensagens naquele lote ficarão visíveis na fila novamente. Por conta disso, seu pipe deve ter a capacidade de processar a mesma mensagem várias vezes sem causar efeitos colaterais não intencionais. É possível modificar esse comportamento de reprocessamento incluindo falhas de item de lote na resposta do pipe. O exemplo a seguir mostra um evento para um lote de duas mensagens.
**Eventos de exemplo**
O exemplo de evento a seguir mostra as informações recebidas pelo pipe. É possível usar esse evento para criar e filtrar seus padrões de eventos ou para definir a transformação de entrada. Nem todos os campos podem ser filtrados. Para mais informações sobre quais campos podem ser filtrados, consulte [Filtragem de eventos no Amazon Pipes EventBridge](eb-pipes-event-filtering.md).
**Fila padrão**
```
[
{
"messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
"receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
"body": "Test message.",
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1545082649183",
"SenderId": "AIDAIENQZJOLO23YVJ4VO",
"ApproximateFirstReceiveTimestamp": "1545082649185"
},
"messageAttributes": {},
"md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
"awsRegion": "us-east-2"
},
{
"messageId": "2e1424d4-f796-459a-8184-9c92662be6da",
"receiptHandle": "AQEBzWwaftRI0KuVm4tP+/7q1rGgNqicHq...",
"body": "Test message.",
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1545082650636",
"SenderId": "AIDAIENQZJOLO23YVJ4VO",
"ApproximateFirstReceiveTimestamp": "1545082650649"
},
"messageAttributes": {},
"md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
"awsRegion": "us-east-2"
}
]
```
**Fila FIFO**
```
[
{
"messageId": "11d6ee51-4cc7-4302-9e22-7cd8afdaadf5",
"receiptHandle": "AQEBBX8nesZEXmkhsmZeyIE8iQAMig7qw...",
"body": "Test message.",
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1573251510774",
"SequenceNumber": "18849496460467696128",
"MessageGroupId": "1",
"SenderId": "AIDAIO23YVJENQZJOL4VO",
"MessageDeduplicationId": "1",
"ApproximateFirstReceiveTimestamp": "1573251510774"
},
"messageAttributes": {},
"md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:fifo.fifo",
"awsRegion": "us-east-2"
}
]
```
## Escalabilidade e processamento
Para filas padrão, o EventBridge usa a [sondagem longa](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-short-and-long-polling.html) para sondar uma fila até que ela se torne ativa. Quando mensagens estão disponíveis, o EventBridge faz a leitura de até cinco lotes e os envia ao seu pipe. Se ainda houver mensagens disponíveis, o EventBridge aumentará o número de processos de leitura de lotes em até 300 instâncias a mais por minuto. O número máximo de lotes que um pipe pode processar simultaneamente é mil.
Em filas FIFO, o EventBridge envia mensagens para o seu lote na ordem em que as recebe. Ao enviar uma mensagem para uma fila do FIFO, você especifica um[ID do grupo de mensagens](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/using-messagegroupid-property.html). O Amazon SQS facilita a entrega de mensagens no mesmo grupo para o EventBridge, em ordem. O EventBridge classifica as mensagens recebidas em grupos e envia apenas um lote de cada vez para um grupo. Se seu pipe retornar um erro, ele fará todas as tentativas possíveis nas mensagens afetadas antes que o EventBridge receba mensagens adicionais do mesmo grupo.
## Como configurar uma fila a ser usada com o EventBridge Pipes
[Crie uma fila do Amazon SQS](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-configure-create-queue.html) para servir como uma origem para seu pipe. Depois, configure a fila para permitir que a sua função do Lambda processe cada lote de eventos e que o EventBridge tentar novamente em resposta a erros de controle de utilização à medida que aumenta a escala verticalmente.
Para permitir que o pipe tenha tempo para processar cada lote de registros, defina o tempo limite de visibilidade da fila de origem para pelo menos seis vezes o runtime combinado do enriquecimento do pipe e dos componentes de destino. O tempo extra permite que o EventBridge repita o processo se seu pope for limitado durante o processamento de um lote anterior.
Se a sua função não conseguir processar uma mensagem várias vezes, o Amazon SQS poderá enviá-la para uma [fila de mensagens não entregues](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-dead-letter-queues.html). Quando seu pipe retorna um erro, o EventBridge a mantém na fila. Depois que o tempo limite de visibilidade ocorrer, o EventBridge recebe a mensagem novamente. Para enviar mensagens para uma segunda fila após vários recebimentos, configure uma fila de mensagens mortas na fila de origem.
**nota**
Certifique-se de configurar a fila de mensagens não entregues na fila de origem, e não no pipe. A fila de mensagens não entregues configurada em uma função é usada para a fila de invocação assíncrona da função, e não para filas de origem de evento.
Se seu pipe retornar um erro ou não puder ser invocada porque está na simultaneidade máxima, o processamento poderá ter êxito com tentativas adicionais. Para que as mensagens tenham mais chances de serem processadas antes de serem enviadas para a fila de mensagens não entregues, defina a `maxReceiveCount` na política de redirecionamento da fila de origem como pelo menos **5**.
## Gerar relatórios de falhas de itens de lote
Quando o EventBridge consome e processa dados de streaming de uma origem, ele definirá checkpoints por padrão no número mais elevado na sequência de um lote, mas somente quando o lote tiver êxito total. Para evitar o reprocessamento de todas as mensagens processadas com êxito em um lote com falha, é possível configurar o enriquecimento ou o destino para retornar um objeto indicando quais mensagens tiveram êxito e quais falharam. Isso se chama resposta parcial em lote.
Para obter mais informações, consulte [Lote com falha parcial](eb-pipes-batching-concurrency.md#pipes-partial-batch-failure).
### Condições de sucesso e falha
Se retornar qualquer um dos seguintes, o EventBridge tratará um lote como um êxito total:
+ Uma lista de `batchItemFailure` vazia
+ Uma lista de `batchItemFailure` nula
+ Uma vazia `EventResponse`
+ Uma nula `EventResponse`
Se retornar qualquer um dos seguintes, o EventBridge tratará um lote como uma falha total:
+ Uma string vazia `itemIdentifier`
+ Uma nula `itemIdentifier`
+ Um `itemIdentifier` com um nome de chave inválido
O EventBridge faz novas tentativas após falhas com base na sua estratégia de repetição.