SelfManagedKafka

Objek yang menggambarkan tipe sumber peristiwa SelfManagedKafka. Untuk informasi selengkapnya, lihat Menggunakan AWS Lambda Apache Kafka yang dikelola sendiri di Panduan Pengembang.AWS Lambda

AWS Serverless Application Model (AWS SAM) menghasilkan AWS::Lambda::EventSourceMappingsumber daya saat jenis acara ini disetel.

Untuk menggunakan Schema Registry, Anda perlu menentukan izin peran IAM tertentu untuk fungsi Anda. Lihat Penyiapan lengkap dengan peran IAM untuk contoh konfigurasi yang diperlukan.

Sintaksis

Untuk mendeklarasikan entitas ini di AWS SAM template Anda, gunakan sintaks berikut.

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

Properti

BatchSize

Jumlah maksimum rekaman di setiap batch yang Lambda tarik dari aliran Anda dan dikirim ke fungsi Anda.

Tipe: Integer

Wajib: Tidak

Default: 100

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke BatchSize properti AWS::Lambda::EventSourceMapping sumber daya.

Minimal: 1

Maksimum: 10000

ConsumerGroupId

String yang mengonfigurasi bagaimana acara akan dibaca dari topik Kafka.

Tipe: String

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke SelfManagedKafkaConfiguration properti AWS::Lambda::EventSourceMapping sumber daya.

DestinationConfig

Objek konfigurasi yang menentukan tujuan dari peristiwa setelah Lambda memprosesnya.

Gunakan properti ini untuk menentukan tujuan pemanggilan gagal dari sumber acara Kafka yang dikelola sendiri.

Jenis: DestinationConfig

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke DestinationConfig properti AWS::Lambda::EventSourceMapping sumber daya.

Enabled

Menonaktifkan pemetaan sumber peristiwa untuk menjeda polling dan pemanggilan.

Tipe: Boolean

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke Enabled properti AWS::Lambda::EventSourceMapping sumber daya.

FilterCriteria

Objek yang mendefinisikan kriteria untuk menentukan apakah Lambda harus memproses suatu peristiwa. Untuk informasi selengkapnya, lihat pemfilteran AWS Lambda acara di Panduan AWS Lambda Pengembang.

Jenis: FilterCriteria

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke FilterCriteria properti AWS::Lambda::EventSourceMapping sumber daya.

KafkaBootstrapServers

Daftar server bootstrap untuk broker Kafka Anda. Sertakan port, misalnya broker.example.com:xxxx

Tipe: Daftar

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini unik AWS SAM dan tidak memiliki AWS CloudFormation padanan.

KmsKeyArn

Nama Sumber Daya Amazon (ARN) dari kunci untuk mengenkripsi informasi yang terkait dengan acara ini.

Tipe: String

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke KmsKeyArn properti AWS::Lambda::EventSourceMapping sumber daya.

ProvisionedPollerConfig

Konfigurasi untuk meningkatkan jumlah poller yang digunakan untuk menghitung pemetaan sumber peristiwa. Konfigurasi ini memungkinkan minimal 1 poller dan maksimal 20 poller. Sebagai contoh, lihat ProvisionedPollerConfig contoh

Jenis: ProvisionedPollerConfig

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke ProvisionedPollerConfig properti AWS::Lambda::EventSourceMapping sumber daya.

SchemaRegistryConfig

Konfigurasi untuk menggunakan registri skema dengan sumber acara Kafka yang dikelola sendiri.

catatan

Fitur ini ProvisionedPollerConfig perlu dikonfigurasi.

Jenis: SchemaRegistryConfig

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke SelfManagedKafkaEventSourceConfig properti AWS::Lambda::EventSourceMapping sumber daya.

SourceAccessConfigurations

Susunan protokol autentikasi, komponen VPC, atau host virtual untuk mengamankan dan menentukan sumber peristiwa Anda.

Nilai yang valid: BASIC_AUTH | CLIENT_CERTIFICATE_TLS_AUTH | SASL_SCRAM_256_AUTH | SASL_SCRAM_512_AUTH | SERVER_ROOT_CA_CERTIFICATE

Jenis: Daftar SourceAccessConfiguration

Wajib: Ya

AWS CloudFormation kompatibilitas: Properti ini adalah bagian dari SelfManagedKafkaEventSourceConfigproperti AWS::Lambda::EventSourceMapping sumber daya.

StartingPosition

Posisi dalam pengaliran tempat untuk mulai membaca.

AT_TIMESTAMP— Tentukan waktu untuk mulai membaca catatan.
LATEST— Baca hanya catatan baru.
TRIM_HORIZON— Memproses semua catatan yang tersedia.

Nilai yang valid: AT_TIMESTAMP | LATEST | TRIM_HORIZON

Tipe: String

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke StartingPosition properti AWS::Lambda::EventSourceMapping sumber daya.

StartingPositionTimestamp

Waktu untuk mulai membaca, dalam detik waktu Unix. Tentukan StartingPositionTimestamp kapan StartingPosition ditentukan sebagaiAT_TIMESTAMP.

Tipe: Ganda

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke StartingPositionTimestamp properti AWS::Lambda::EventSourceMapping sumber daya.

Topics

Nama topik Kafka.

Tipe: Daftar

Wajib: Ya

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke Topics properti AWS::Lambda::EventSourceMapping sumber daya.

Contoh

Penyiapan lengkap dengan peran IAM

Contoh berikut menunjukkan penyiapan lengkap termasuk konfigurasi peran IAM yang diperlukan untuk menggunakan 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 contoh


ProvisionedPollerConfig:
  MinimumPollers: 1
  MaximumPollers: 20

Sumber acara Kafka yang dikelola sendiri

Berikut adalah contoh peristiwa dari tipe sumber peristiwa 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

Sumber Acara Kafka yang dikelola sendiri dengan AWS Glue Schema Registry

Berikut ini adalah contoh dari jenis sumber SelfManagedKafka peristiwa dikonfigurasi dengan 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

Sumber Acara Kafka yang dikelola sendiri dengan Registri Skema Konfluen

Berikut ini adalah contoh dari jenis sumber SelfManagedKafka peristiwa dikonfigurasi dengan 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

Awas Javascript dinonaktifkan atau tidak tersedia di browser Anda.

Untuk menggunakan Dokumentasi AWS, Javascript harus diaktifkan. Lihat halaman Bantuan browser Anda untuk petunjuk.

Konvensi Dokumen

ScheduleV2

SNS