# Creación de una asignación de orígenes de eventos de Lambda para un origen de eventos de Apache Kafka autoadministrado
Para crear una asignación de orígenes de eventos, puede usar la consola de Lambda, la [AWS Command Line Interface (CLI)](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) o un [AWS SDK](https://aws.amazon.com/getting-started/tools-sdks/).
Los siguientes pasos de la consola agregan un clúster de Apache Kafka autoadministrado como desencadenador para su función de Lambda. Internamente, esto crea un recurso de asignación de orígenes de eventos.
## Requisitos previos
+ Un clúster de Apache Kafka autoadministrado. Lambda es compatible con Apache Kafka 0.10.1.0 y versiones posteriores.
+ Un [rol de ejecución](lambda-intro-execution-role.md) con permiso para acceder a los recursos de AWS que utiliza el clúster de Kafka autoadministrado.
## Agregar un clúster de Kafka autoadministrado (consola)
Siga estos pasos para agregar su clúster Apache Kafka autoadministrado y un tema Kafka como desencadenador de su función de Lambda.
**Para agregar un desencadenador de Apache Kafka a su función de Lambda (consola)**
1. .Abra la página de [Functions (Funciones)](https://console.aws.amazon.com/lambda/home#/functions) en la consola de Lambda.
1. Elija el nombre de su función de Lambda.
1. En **Descripción general de la función**, elija **Agregar desencadenador**.
1. En **Configuración del desencadenador**, haga lo siguiente:
1. Elija el tipo de desencadenador **Apache Kafka**.
1. Para los **servidores de Bootstrap**, ingrese la dirección de host y par de puertos de un broker de Kafka en su clúster y, a continuación, elija **Add (Agregar)**. Repita para cada broker de Kafka en el clúster.
1. Para el **nombre del tema**, escriba el nombre del tema Kafka utilizado para almacenar registros en el clúster.
1. Si configura el modo aprovisionado, ingrese un valor para **Sondeadores de eventos mínimos**, un valor para **Sondeadores de eventos máximos** y un valor opcional para PollerGroupName para especificar la agrupación de varias ESM dentro de la misma VPC de origen de eventos.
1. (Opcional) Para **Tamaño del lote**, introduzca el número máximo de registros que se recibirán en un solo lote.
1. Para el **periodo de lotes**, ingrese la cantidad máxima de segundos que Lambda emplea a fin de recopilar registros antes de invocar la función.
1. (Opcional) Para el **ID del grupo de consumidores**, ingrese el ID de un grupo de consumidores de Kafka al que unirse.
1. (Opcional) En **Posición inicial**, elija **Última** para empezar a leer el flujo desde el registro más reciente, **Horizonte de supresión** para comenzar por el registro más antiguo disponible o **En la marca de tiempo** para especificar una marca de tiempo desde la cual comenzar a leer.
1. (Opcional) Para **VPC**, elija Amazon VPC para su clúster de Kafka. A continuación, elija **VPC subnets** (Subredes de VPC) y **VPC security groups** (Grupos de seguridad de VPC).
Esta configuración es obligatoria si solo los usuarios de la VPC acceden a los agentes.
1. (Opcional) Para **Authentication** (Autenticación), elija **Add** (Agregar) y, a continuación, haga lo siguiente:
1. Elija el protocolo de acceso o autenticación de los agentes de Kafka en su clúster.
+ Si su agente de Kafka utiliza autenticación SASL/PLAIN, elija **BASIC\$1AUTH**.
+ Si su agente utiliza autenticación de SASL/SCRAM, elija uno de los protocolos de **SASL\$1SCRAM**.
+ Si configura la autenticación de mTLS, elija el protocolo **CLIENT\$1CERTIFICATE\$1TLS\$1AUTH**.
1. Para la autenticación de SASL/SCRAM o mTLS, elija la clave secreta de Secrets Manager que contiene las credenciales del clúster de Kafka.
1. (Opcional) Para **Encryption** (Cifrado), elija el secreto de Secrets Manager que contiene el certificado de entidad de certificación raíz que los agentes de Kafka utilizan para el cifrado con TLS, si los agentes de Kafka utilizan certificados firmados por una entidad de certificación privada.
Esta configuración se aplica al cifrado con TLS para SASL/SCRAM o SASL/PLAIN y a la autenticación con mTLS.
1. Para crear el desencadenador en un estado deshabilitado para la prueba (recomendado), desactive **Activar desencadenador**. O bien, para habilitar el desencadenador de inmediato, seleccione**Activar desencadenador**.
1. Para crear el desencadenador, elija **Add** (Añadir).
## Agregar un clúster Kafka autoadministrado (AWS CLI)
Utilice los siguientes AWS CLI comandos de ejemplo para crear y ver un desencadenador Apache Kafka autoadminstrado para su función de Lambda.
### Uso de SASL/SCRAM
Si los usuarios de Kafka acceden a los agentes de Kafka a través de Internet, especifique el secreto de Secrets Manager que creó para la autenticación con SASL/SCRAM. En el siguiente ejemplo se utiliza el comando [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) de la AWS CLI para asignar una función de Lambda llamada `my-kafka-function` a un tema de Kafka llamado `AWSKafkaTopic`.
```
aws lambda create-event-source-mapping \
--topics AWSKafkaTopic \
--source-access-configuration Type=SASL_SCRAM_512_AUTH,URI=arn:aws:secretsmanager:us-east-1:111122223333:secret:MyBrokerSecretName \
--function-name arn:aws:lambda:us-east-1:111122223333:function:my-kafka-function \
--self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc3.xyz.com:9092", "abc2.xyz.com:9092"]}}'
```
### Uso de una VPC
Si solo los usuarios de Kafka de su VPC acceden a sus agentes de Kafka, debe especificar su VPC, subredes y grupo de seguridad de VPC. En el siguiente ejemplo se utiliza el comando [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) de la AWS CLI para asignar una función de Lambda llamada `my-kafka-function` a un tema de Kafka llamado `AWSKafkaTopic`.
```
aws lambda create-event-source-mapping \
--topics AWSKafkaTopic \
--source-access-configuration '[{"Type": "VPC_SUBNET", "URI": "subnet:subnet-0011001100"}, {"Type": "VPC_SUBNET", "URI": "subnet:subnet-0022002200"}, {"Type": "VPC_SECURITY_GROUP", "URI": "security_group:sg-0123456789"}]' \
--function-name arn:aws:lambda:us-east-1:111122223333:function:my-kafka-function \
--self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc3.xyz.com:9092", "abc2.xyz.com:9092"]}}'
```
### Visualización del estado mediante la AWS CLI
En el siguiente ejemplo se utiliza el comando [get-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/get-event-source-mapping.html) de la AWS CLI para describir el estado de la asignación de orígenes de eventos que ha creado.
```
aws lambda get-event-source-mapping
--uuid dh38738e-992b-343a-1077-3478934hjkfd7
```