SelfManagedKafka

El objeto que describe un tipo de fuente de evento SelfManagedKafka. Para obtener más información, consulte Uso AWS Lambda con Apache Kafka autogestionado en la Guía para AWS Lambda desarrolladores.

AWS Serverless Application Model (AWS SAM) genera un AWS::Lambda::EventSourceMappingrecurso cuando se establece este tipo de evento.

Para utilizar Schema Registry, debe definir permisos de rol de IAM específicos para su función. Consulte Configuración completa con funciones de IAM para ver un ejemplo de la configuración requerida.

Sintaxis

Para declarar esta entidad en la AWS SAM plantilla, utilice la siguiente sintaxis.

YAML


  BatchSize: Integer
  ConsumerGroupId: String
  DestinationConfig: DestinationConfig
  Enabled: Boolean
  FilterCriteria: FilterCriteria
  KafkaBootstrapServers: List
  KmsKeyArn: String
  ProvisionedPollerConfig: ProvisionedPollerConfig
  SchemaRegistryConfig: SchemaRegistryConfig
  SourceAccessConfigurations: SourceAccessConfigurations
  StartingPosition: String
  StartingPositionTimestamp: Double
  Topics: List

Propiedades

BatchSize

El número máximo de registros en cada lote que Lambda extrae del flujo o la cola y envía a su función.

Tipo: entero

Obligatorio: no

Valor predeterminado: 100

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la BatchSize propiedad de un AWS::Lambda::EventSourceMapping recurso.

Mínimo: 1

Máximo: 10000

ConsumerGroupId

Cadena que configura cómo se leerán los eventos de los temas de Kafka.

Tipo: cadena

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la SelfManagedKafkaConfiguration propiedad de un AWS::Lambda::EventSourceMapping recurso.

DestinationConfig

Un objeto de configuración que especifica el destino de un evento después de que Lambda lo procese.

Utilice esta propiedad para especificar el destino de las invocaciones fallidas desde el origen de eventos autoadministrado de Kafka.

Tipo: DestinationConfig

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la DestinationConfig propiedad de un AWS::Lambda::EventSourceMapping recurso.

Enabled

Deshabilita el mapeo de origen de eventos para pausar el sondeo y la invocación.

Tipo: Booleano

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la Enabled propiedad de un AWS::Lambda::EventSourceMapping recurso.

FilterCriteria

Un objeto que define los criterios que determinan si Lambda debe procesar un evento o no. Para obtener más información, consulta Filtrado de eventos de AWS Lambda en la Guía para desarrolladores de AWS Lambda .

Tipo: FilterCriteria

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la FilterCriteria propiedad de un AWS::Lambda::EventSourceMapping recurso.

KafkaBootstrapServers

La lista de servidores de arranque para sus agentes Kafka: Incluya el puerto, por ejemplo broker.example.com:xxxx

Tipo: lista

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad es exclusiva AWS SAM y no tiene un AWS CloudFormation equivalente.

KmsKeyArn

El nombre de recurso de Amazon (ARN) de la clave de que se utilizará para cifrar la información de este evento.

Tipo: cadena

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la KmsKeyArn propiedad de un AWS::Lambda::EventSourceMapping recurso.

ProvisionedPollerConfig

Configuración para aumentar la cantidad de sondeos utilizados para calcular las asignaciones de fuentes de eventos. Esta configuración permite un mínimo de 1 sondeador y un máximo de 20 sondeos. Para ver un ejemplo, consulte ProvisionedPollerConfig ejemplo

Tipo: ProvisionedPollerConfig

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la ProvisionedPollerConfig propiedad de un AWS::Lambda::EventSourceMapping recurso.

SchemaRegistryConfig

Configuración para utilizar un registro de esquemas con la fuente de eventos de Kafka autogestionada.

nota

Es necesario ProvisionedPollerConfig configurar esta función.

Tipo: SchemaRegistryConfig

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la SelfManagedKafkaEventSourceConfig propiedad de un AWS::Lambda::EventSourceMapping recurso.

SourceAccessConfigurations

Una matriz del protocolo de autenticación, los componentes de VPC o el host virtual para proteger y definir su origen de eventos.

Valores válidos: BASIC_AUTH | CLIENT_CERTIFICATE_TLS_AUTH | SASL_SCRAM_256_AUTH | SASL_SCRAM_512_AUTH | SERVER_ROOT_CA_CERTIFICATE

Tipo: lista de SourceAccessConfiguration

Obligatorio: sí

AWS CloudFormation compatibilidad: esta propiedad forma parte de la SelfManagedKafkaEventSourceConfigpropiedad de un AWS::Lambda::EventSourceMapping recurso.

StartingPosition

La posición en el flujo donde comienza la lectura.

AT_TIMESTAMP: especifique el tiempo a partir del cual comenzar la lectura de registros.
LATEST: registros nuevos de solo lectura.
TRIM_HORIZON: procese todos los registros disponibles.

Valores válidos: AT_TIMESTAMP | LATEST | TRIM_HORIZON

Tipo: cadena

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la StartingPosition propiedad de un AWS::Lambda::EventSourceMapping recurso.

StartingPositionTimestamp

El tiempo a partir del cual comenzar la lectura, en segundos de tiempo Unix. Defina StartingPositionTimestamp cuando StartingPosition se especifica como AT_TIMESTAMP.

Tipo: Doble

Obligatorio: no

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la StartingPositionTimestamp propiedad de un AWS::Lambda::EventSourceMapping recurso.

Topics

El nombre del tema de Kafka.

Tipo: lista

Obligatorio: sí

AWS CloudFormation compatibilidad: esta propiedad se pasa directamente a la Topics propiedad de un AWS::Lambda::EventSourceMapping recurso.

Ejemplos

Configuración completa con funciones de IAM

El siguiente ejemplo muestra una configuración completa que incluye la configuración del rol de IAM necesaria para usar Schema Registry:


Parameters:
  PreCreatedSubnetOne:
    Type: String
  PreCreatedSubnetTwo:
    Type: String
 
Resources:
  MyLambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Action: [sts:AssumeRole]
          Effect: Allow
          Principal:
            Service: [lambda.amazonaws.com]
      Policies:
      - PolicyName: KafkaAuthPolicy
        PolicyDocument:
          Statement:
          - Action: [secretsmanager:GetSecretValue, kms:Decrypt]
            Effect: "Allow"            
            Resource: ['arn:aws:secretsmanager:us-west-2:123456789012:secret:kafkaSecret-******',
                        'arn:aws:kms:us-west-2:123456789012:key/keyId']
      - PolicyName: ENIPolicy
        PolicyDocument:
          Statement:
          - Action: [ec2:CreateNetworkInterface,
              ec2:DescribeNetworkInterfaces, ec2:DescribeVpcs, ec2:DeleteNetworkInterface,
              ec2:DescribeSubnets, ec2:DescribeSecurityGroups]
            Effect: Allow
            Resource: '*'      
      - PolicyName: SchemaRegistryPolicy
        PolicyDocument:
          Statement:
          - Action: [glue:GetRegistry]
            Effect: Allow
            Resource: 'arn:aws:glue:{region}:{account-id}:registry/{registry-name}'
      - PolicyName: SchemaVersionsPolicy
        PolicyDocument:
          Statement:
          - Action: [glue:GetSchemaVersions]
            Effect: Allow
            Resource: '*'          
      ManagedPolicyArns:
      - !Sub arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Tags:
      - {Value: SAM, Key: lambda:createdBy}
 
  MyKafkaProcessor:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: nodejs18.x
      Handler: index.handler
      CodeUri: ${codeuri}
      Role:
        Fn::GetAtt: [MyLambdaExecutionRole, Arn]
      Events:
        SelfManagedKafkaEvent:
          Type: SelfManagedKafka
          Properties:
            KafkaBootstrapServers:
              - my-kafka-broker-1:9092
              - my-kafka-broker-2:9092
            Topics:
              - SchemaRegistryTestTopic
            StartingPosition: LATEST
            SourceAccessConfigurations:
              - Type: VPC_SUBNET
                URI: subnet:subnet-12345678
              - Type: VPC_SECURITY_GROUP
                URI: security_group:sg-12345678
              - Type: BASIC_AUTH
                URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
            ProvisionedPollerConfig:
              MinimumPollers: 1
            SchemaRegistryConfig:
              AccessConfigs:
              - Type: BASIC_AUTH
                URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
              SchemaValidationConfigs:
              - Attribute: KEY
              EventRecordFormat: JSON
              SchemaRegistryURI: !Sub arn:${AWS::Partition}:glue:us-west-2:123456789012:registry/myregistry

ProvisionedPollerConfig ejemplo


ProvisionedPollerConfig:
  MinimumPollers: 1
  MaximumPollers: 20

Origen de eventos Kafka autoadministrado

A continuación, se muestra un ejemplo de un tipo de fuente de evento SelfManagedKafka.

YAML


Events:
  SelfManagedKafkaEvent:
    Type: SelfManagedKafka
    Properties:
      BatchSize: 1000
      Enabled: true
      KafkaBootstrapServers:
        - abc.xyz.com:xxxx
      SourceAccessConfigurations: 
        -  Type: BASIC_AUTH
           URI: arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
      Topics:
        - MyKafkaTopic

Fuente de eventos de Kafka autogestionada con Schema Registry AWS Glue

El siguiente es un ejemplo de un tipo de fuente de SelfManagedKafka eventos configurado con AWS Glue Schema Registry.


Events:
  SelfManagedKafkaEvent:
    Type: SelfManagedKafka
    Properties:
      KafkaBootstrapServers:
        - abc.xyz.com:9092
      Topics:
        - SchemaRegistryTestTopic
      StartingPosition: LATEST
      ProvisionedPollerConfig:
        MinimumPollers: 1
      SchemaRegistryConfig:
        SchemaRegistryURI: !Sub arn:${AWS::Partition}:glue:us-west-2:123456789012:registry/myregistry
        EventRecordFormat: JSON
        SchemaValidationConfigs:
          - Attribute: KEY
          - Attribute: VALUE
      SourceAccessConfigurations:
        - Type: VPC_SUBNET
          URI: subnet:subnet-12345678
        - Type: VPC_SECURITY_GROUP
          URI: security_group:sg-12345678

Fuente de eventos de Kafka autogestionada con registro de esquemas confluentes

El siguiente es un ejemplo de un tipo de fuente de SelfManagedKafka eventos configurado con Confluent Schema Registry.


Events:
  SelfManagedKafkaEvent:
    Type: SelfManagedKafka
    Properties:
      KafkaBootstrapServers:
        - abc.xyz.com:9092
      Topics:
        - SchemaRegistryTestTopic
      StartingPosition: LATEST
      ProvisionedPollerConfig:
        MinimumPollers: 1
      SchemaRegistryConfig:
        SchemaRegistryURI: https://my-schema-registry.confluent.cloud
        AccessConfigs:
          - Type: BASIC_AUTH
            URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-secret
        EventRecordFormat: JSON
        SchemaValidationConfigs:
          - Attribute: KEY
          - Attribute: VALUE
      SourceAccessConfigurations:
        - Type: VPC_SUBNET
          URI: subnet:subnet-12345678
        - Type: VPC_SECURITY_GROUP
          URI: security_group:sg-12345678
        - Type: BASIC_AUTH
          URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:kafka-secret

Aviso JavaScript está desactivado o no está disponible en su navegador.

Para utilizar la documentación de AWS, debe estar habilitado JavaScript. Para obtener más información, consulte las páginas de ayuda de su navegador.

Convenciones del documento

ScheduleV2

SNS