SelfManagedKafka - AWS Serverless Application Model
SyntaxePropriétésExemples

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

SelfManagedKafka

L'objet décrivant un type de source d'événement SelfManagedKafka. Pour plus d'informations, consultez la section Utilisation AWS Lambda avec Apache Kafka autogéré dans le Guide du AWS Lambda développeur.

AWS Serverless Application Model (AWS SAM) génère une AWS::Lambda::EventSourceMappingressource lorsque ce type d'événement est défini.

Pour utiliser Schema Registry, vous devez définir des autorisations de rôle IAM spécifiques pour votre fonction. Voir Configuration complète avec des rôles IAM pour un exemple de configuration requise.

Syntaxe

Pour déclarer cette entité dans votre AWS SAM modèle, utilisez la syntaxe suivante.

Propriétés

BatchSize

Nombre maximal d’enregistrements dans chaque lot que Lambda extrait de votre flux et envoie à votre fonction.

Type : entier

Obligatoire : non

Par défaut : 100

AWS CloudFormation compatibilité : cette propriété est transmise directement à la BatchSize propriété d'une AWS::Lambda::EventSourceMapping ressource.

Minimum : 1

Maximum : 10000

ConsumerGroupId

Chaîne qui configure la façon dont les événements seront lus à partir des rubriques Kafka.

Type : chaîne

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la SelfManagedKafkaConfiguration propriété d'une AWS::Lambda::EventSourceMapping ressource.

DestinationConfig

Objet de configuration qui spécifie la destination d'un événement après son traitement par Lambda.

Utilisez cette propriété pour spécifier la destination des invocations échouées de la source d'événements Kafka autogérée.

Type : DestinationConfig

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la DestinationConfig propriété d'une AWS::Lambda::EventSourceMapping ressource.

Enabled

Désactive le mappage de source d'événement pour suspendre l'interrogation et l'appel.

Type : valeur booléenne

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la Enabled propriété d'une AWS::Lambda::EventSourceMapping ressource.

FilterCriteria

Objet qui définit les critères permettant de déterminer si Lambda doit traiter un événement. Pour de plus amples informations, veuillez consulter AWS Lambda le filtrage d’événements dans le AWS Lambda Manuel du développeur.

Type : FilterCriteria

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la FilterCriteria propriété d'une AWS::Lambda::EventSourceMapping ressource.

KafkaBootstrapServers

La liste des serveurs d’amorçage pour vos agents Kafka. Inclure le port, par exemple broker.example.com:xxxx

Type : liste

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est unique AWS SAM et n'a pas d' AWS CloudFormation équivalent.

KmsKeyArn

Le nom de ressource Amazon (ARN) de la clé permettant de chiffrer les informations relatives à cet événement.

Type : chaîne

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la KmsKeyArn propriété d'une AWS::Lambda::EventSourceMapping ressource.

ProvisionedPollerConfig

Configuration pour augmenter le nombre de sondeurs utilisés pour calculer les mappages de sources d'événements. Cette configuration permet un minimum de 1 sondeur et un maximum de 20 sondeurs. Pour un exemple, reportez-vous à ProvisionedPollerConfig exemple

Type : ProvisionedPollerConfig

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la ProvisionedPollerConfig propriété d'une AWS::Lambda::EventSourceMapping ressource.

SchemaRegistryConfig

Configuration pour l'utilisation d'un registre de schémas avec la source d'événements Kafka autogérée.

Note

Cette fonctionnalité doit ProvisionedPollerConfig être configurée.

Type : SchemaRegistryConfig

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la SelfManagedKafkaEventSourceConfig propriété d'une AWS::Lambda::EventSourceMapping ressource.

SourceAccessConfigurations

Tableau du protocole d’authentification, composants VPC ou hôte virtuel pour sécuriser et définir votre source d’événement.

Valeurs valides : BASIC_AUTH | CLIENT_CERTIFICATE_TLS_AUTH | SASL_SCRAM_256_AUTH | SASL_SCRAM_512_AUTH | SERVER_ROOT_CA_CERTIFICATE

Type : liste de propriétés SourceAccessConfiguration

Obligatoire : oui

AWS CloudFormation compatibilité : cette propriété fait partie de la SelfManagedKafkaEventSourceConfigpropriété d'une AWS::Lambda::EventSourceMapping ressource.

StartingPosition

Position de début de la lecture dans le flux.

  • AT_TIMESTAMP : spécifier l'heure à partir de laquelle la lecture des enregistrements doit commencer.

  • LATEST : lire uniquement les nouveaux enregistrements.

  • TRIM_HORIZON : traiter tous les enregistrements disponibles.

Valeurs valides : AT_TIMESTAMP | LATEST | TRIM_HORIZON

Type : chaîne

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la StartingPosition propriété d'une AWS::Lambda::EventSourceMapping ressource.

StartingPositionTimestamp

L'heure à partir de laquelle commencer la lecture, en secondes au format horaire Unix. Définissez StartingPositionTimestamp lorsque StartingPosition est défini sur AT_TIMESTAMP.

Type : double

Obligatoire : non

AWS CloudFormation compatibilité : cette propriété est transmise directement à la StartingPositionTimestamp propriété d'une AWS::Lambda::EventSourceMapping ressource.

Topics

Nom de la rubrique Kafka.

Type : liste

Obligatoire : oui

AWS CloudFormation compatibilité : cette propriété est transmise directement à la Topics propriété d'une AWS::Lambda::EventSourceMapping ressource.

Exemples

Configuration complète avec des rôles IAM

L'exemple suivant montre une configuration complète, y compris la configuration de rôle IAM requise pour utiliser 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 exemple

ProvisionedPollerConfig:
  MinimumPollers: 1
  MaximumPollers: 20

Source d’événements Kafka autogéré

Voici un exemple de type de source d'événement 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

Source d'événements Kafka autogérée avec registre de schémas AWS Glue

Voici un exemple de type de source d'SelfManagedKafkaévénement configuré avec 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

Source d'événements Kafka autogérée avec registre de schémas Confluent

Voici un exemple de type de source d'SelfManagedKafkaévénement configuré avec 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