MSK

L'oggetto che descrive un tipo di sorgente di MSK eventi. Per ulteriori informazioni, consulta Using AWS Lambda with Amazon MSK nella AWS Lambda Developer Guide.

AWS Serverless Application Model (AWS SAM) genera una AWS::Lambda::EventSourceMappingrisorsa quando viene impostato questo tipo di evento.

Per utilizzare Schema Registry, devi definire autorizzazioni di ruolo IAM specifiche per la tua funzione. Vedi Configurazione completa con ruoli IAM per un esempio della configurazione richiesta.

Sintassi

Per dichiarare questa entità nel tuo AWS SAM modello, usa la seguente sintassi.

YAML


  ConsumerGroupId: String
  DestinationConfig: DestinationConfig
  FilterCriteria: FilterCriteria
  KmsKeyArn: String 
  MaximumBatchingWindowInSeconds: Integer
  ProvisionedPollerConfig: ProvisionedPollerConfig
  SchemaRegistryConfig: SchemaRegistryConfig
  SourceAccessConfigurations: SourceAccessConfigurations
  StartingPosition: String
  StartingPositionTimestamp: Double
  Stream: String
  Topics: List

Proprietà

ConsumerGroupId

Una stringa che configura il modo in cui gli eventi verranno letti dagli argomenti di Kafka.

Tipo: stringa

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla AmazonManagedKafkaConfiguration proprietà di una risorsa. AWS::Lambda::EventSourceMapping

DestinationConfig

Oggetto di configurazione che specifica la destinazione di un evento dopo che Lambda lo elabora.

Utilizza questa proprietà per specificare la destinazione delle chiamate non riuscite dall'origine dell'evento Amazon MSK.

Tipo: DestinationConfig

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla DestinationConfig proprietà di una risorsa. AWS::Lambda::EventSourceMapping

FilterCriteria

Un oggetto che definisce i criteri che determinano se Lambda deve elaborare un evento. Per ulteriori informazioni, consulta il filtro AWS Lambda degli eventi nella Guida per gli AWS Lambda sviluppatori.

Tipo: FilterCriteria

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla FilterCriteria proprietà di una AWS::Lambda::EventSourceMapping risorsa.

KmsKeyArn

L'Amazon Resource Name (ARN) della chiave per crittografare le informazioni relative a questo evento.

Tipo: stringa

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla KmsKeyArn proprietà di una AWS::Lambda::EventSourceMapping risorsa.

MaximumBatchingWindowInSeconds

Il tempo massimo in secondi per la raccolta dei record prima di richiamare la funzione.

Tipo: integer

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla MaximumBatchingWindowInSeconds proprietà di una AWS::Lambda::EventSourceMapping risorsa.

ProvisionedPollerConfig

Configurazione per aumentare la quantità di poller utilizzati per calcolare le mappature delle sorgenti degli eventi. Questa configurazione consente un minimo di 1 poller e un massimo di 20 poller. Per un esempio, fare riferimento a. ProvisionedPollerConfig esempio

Tipo: ProvisionedPollerConfig

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla ProvisionedPollerConfig proprietà di una AWS::Lambda::EventSourceMapping risorsa.

SchemaRegistryConfig

Configurazione per l'utilizzo di un registro degli schemi con l'origine degli eventi Kafka.

Nota

Questa funzionalità deve essere ProvisionedPollerConfig configurata.

Tipo: SchemaRegistryConfig

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla AmazonManagedKafkaEventSourceConfig proprietà di una AWS::Lambda::EventSourceMapping risorsa.

SourceAccessConfigurations

Un array del protocollo di autenticazione, dei componenti VPC o dell'host virtuale per proteggere e definire l'origine eventi.

Valori validi: CLIENT_CERTIFICATE_TLS_AUTH

Tipo: elenco di SourceAccessConfiguration

Required: No

AWS CloudFormation compatibilità: questa proprietà fa parte della AmazonManagedKafkaEventSourceConfigproprietà di una risorsa. AWS::Lambda::EventSourceMapping

StartingPosition

La posizione in un flusso da cui iniziare la lettura.

AT_TIMESTAMP— Specificare l'ora a partire dalla quale iniziare a leggere i record.
LATEST— Leggi solo i nuovi record.
TRIM_HORIZON— Elabora tutti i record disponibili.

Valori validi: AT_TIMESTAMP | LATEST | TRIM_HORIZON

Tipo: stringa

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla StartingPosition proprietà di una AWS::Lambda::EventSourceMapping risorsa.

StartingPositionTimestamp

L'ora da cui iniziare a leggere, in secondi di tempo Unix. Definisce StartingPositionTimestamp quando StartingPosition è specificato comeAT_TIMESTAMP.

Tipo: double

Required: No

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla StartingPositionTimestamp proprietà di una AWS::Lambda::EventSourceMapping risorsa.

Stream

L'Amazon Resource Name (ARN) del flusso di dati o di un consumatore di streaming.

Tipo: stringa

Campo obbligatorio: sì

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla EventSourceArn proprietà di una AWS::Lambda::EventSourceMapping risorsa.

Topics

Il nome dell'argomento Kafka.

Tipo: Elenco

Campo obbligatorio: sì

AWS CloudFormation compatibilità: questa proprietà viene passata direttamente alla Topics proprietà di una AWS::Lambda::EventSourceMapping risorsa.

Esempi

Configurazione completa con ruoli IAM

L'esempio seguente mostra una configurazione completa che include la configurazione del ruolo IAM richiesta per l'utilizzo di Schema Registry:


Parameters:
  PreCreatedSubnetOne:
    Type: String
  PreCreatedSubnetTwo:
    Type: String
  MskClusterName4:
    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: KafkaClusterPermissions
        PolicyDocument:
          Statement:
          - Action: [kafka:DescribeClusterV2, kafka:GetBootstrapBrokers]
            Effect: Allow
            Resource: 'arn:aws:kafka:us-east-1:123456789012:cluster/*'
      - 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}
 
  MyMskCluster:
    Type: AWS::MSK::Cluster
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
        - Ref: PreCreatedSubnetOne
        - Ref: PreCreatedSubnetTwo
        InstanceType: kafka.t3.small
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 1
      ClusterName:
        Ref: MskClusterName4
      KafkaVersion: 3.8.x
      NumberOfBrokerNodes: 2
 
  MyMskStreamProcessor:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: nodejs18.x
      Handler: index.handler
      CodeUri: ${codeuri}
      Role:
        Fn::GetAtt: [MyLambdaExecutionRole, Arn]
      Events:
        MyMskEvent:
          Type: MSK
          Properties:
            StartingPosition: LATEST
            Stream:
              Ref: MyMskCluster
            SourceAccessConfigurations:
            - Type: SASL_SCRAM_512_AUTH 
              URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
            Topics:
            - SchemaRegistryTestTopic
            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 esempio


ProvisionedPollerConfig:
  MinimumPollers: 1
  MaximumPollers: 20

Esempio di Amazon MSK per un cluster esistente

Di seguito è riportato un esempio di un tipo di origine di MSK eventi per un cluster Amazon MSK che esiste già in un Account AWS.

YAML


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream: arn:aws:kafka:us-east-1:012345678012:cluster/exampleClusterName/abcdefab-1234-abcd-5678-cdef0123ab01-2
      Topics:
        - MyTopic

Esempio di Amazon MSK per un cluster dichiarato nello stesso modello

Di seguito è riportato un esempio di un tipo di origine di MSK eventi per un cluster Amazon MSK dichiarato nello stesso file modello.

YAML


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream:
        Ref: MyMskCluster   # This must be the name of an MSK cluster declared in the same template file
      Topics:
        - MyTopic

Fonte di eventi MSK con registro dello schema

Di seguito è riportato un esempio di un tipo di origine di MSK eventi configurato con un registro dello schema.


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream:
        Ref: MyMskCluster
      Topics:
        - SchemaRegistryTestTopic
      ProvisionedPollerConfig:
        MinimumPollers: 1
      SchemaRegistryConfig:
        SchemaRegistryURI: !Sub arn:${AWS::Partition}:glue:us-west-2:123456789012:registry/myregistry
        EventRecordFormat: JSON
        SchemaValidationConfigs:
          - Attribute: KEY
          - Attribute: VALUE

Fonte di eventi MSK con registro degli schemi Confluent

Di seguito è riportato un esempio di un tipo di origine di MSK eventi configurato con un Confluent Schema Registry.


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream:
        Ref: MyMskCluster
      Topics:
        - SchemaRegistryTestTopic
      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

Avvertimento JavaScript è disabilitato o non è disponibile nel tuo browser.

Per usare la documentazione AWS, JavaScript deve essere abilitato. Consulta le pagine della guida del browser per le istruzioni.

Convenzioni dei documenti