AWSCDK::LambdaEventSources

31 types

AWS Lambda Event Sources

An event source mapping is an AWS Lambda resource that reads from an event source and invokes a Lambda function. You can use event source mappings to process items from a stream or queue in services that don't invoke Lambda functions directly. Lambda provides event source mappings for the following services. Read more about lambda event sources here.

This module includes classes that allow using various AWS services as event sources for AWS Lambda via the high-level lambda.addEventSource(source) API.

NOTE: In most cases, it is also possible to use the resource APIs to invoke an AWS Lambda function. This library provides a uniform API for all Lambda event sources regardless of the underlying mechanism they use.

The following code sets up a lambda function with an SQS queue event source -

require 'aws-cdk-lib'

fn = nil # AWSCDK::Lambda::Function

queue = AWSCDK::SQS::Queue.new(self, "MyQueue")
event_source = AWSCDK::LambdaEventSources::SQSEventSource.new(queue)
fn.add_event_source(event_source)

event_source_id = event_source.event_source_mapping_id
event_source_mapping_arn = event_source.event_source_mapping_arn

The event_source_id property contains the event source id. This will be a token that will resolve to the final value at the time of deployment.

The event_source_mapping_arn property contains the event source mapping ARN. This will be a token that will resolve to the final value at the time of deployment.

SQS

Amazon Simple Queue Service (Amazon SQS) allows you to build asynchronous workflows. For more information about Amazon SQS, see Amazon Simple Queue Service. You can configure AWS Lambda to poll for these messages as they arrive and then pass the event to a Lambda function invocation. To view a sample event, see Amazon SQS Event.

To set up Amazon Simple Queue Service as an event source for AWS Lambda, you first create or update an Amazon SQS queue and select custom values for the queue parameters. The following parameters will impact Amazon SQS's polling behavior:

visibilityTimeout: May impact the period between retries.
batchSize: Determines how many records are buffered before invoking your lambda function.
maxBatchingWindow: The maximum amount of time to gather records before invoking the lambda. This increases the likelihood of a full batch at the cost of delayed processing.
maxConcurrency: The maximum concurrency setting limits the number of concurrent instances of the function that an Amazon SQS event source can invoke.
provisionedPollerConfig: Controls the number of pollers that can be provisioned to read from the queue. Valid ranges: minimum_pollers 2–200 (default: 2), maximum_pollers 2–2000 (default: 200).
enabled: If the SQS event source mapping should be enabled. The default is true.

require 'aws-cdk-lib'
fn = nil # AWSCDK::Lambda::Function


queue = AWSCDK::SQS::Queue.new(self, "MyQueue", {
    visibility_timeout: AWSCDK::Duration.seconds(30),
})

fn.add_event_source(AWSCDK::LambdaEventSources::SQSEventSource.new(queue, {
    batch_size: 10,
     # default
    max_batching_window: AWSCDK::Duration.minutes(5),
    report_batch_item_failures: true,
}))

S3

You can write Lambda functions to process S3 bucket events, such as the object-created or object-deleted events. For example, when a user uploads a photo to a bucket, you might want Amazon S3 to invoke your Lambda function so that it reads the image and creates a thumbnail for the photo.

You can use the bucket notification configuration feature in Amazon S3 to configure the event source mapping, identifying the bucket events that you want Amazon S3 to publish and which Lambda function to invoke.

require 'aws-cdk-lib'
fn = nil # AWSCDK::Lambda::Function


bucket = AWSCDK::S3::Bucket.new(self, "mybucket")

fn.add_event_source(AWSCDK::LambdaEventSources::S3EventSource.new(bucket, {
    events: [AWSCDK::S3::EventType::OBJECT_CREATED, AWSCDK::S3::EventType::OBJECT_REMOVED],
    filters: [{prefix: "subdir/"}],
}))

In the example above, S3EventSource is accepting Bucket type as parameter. However, Functions like from_bucket_name and from_bucket_arn will return IBucket and is not compliant with S3EventSource. If this is the case, please consider using S3EventSourceV2 instead, this class accepts IBucket.

require 'aws-cdk-lib'
fn = nil # AWSCDK::Lambda::Function


bucket = AWSCDK::S3::Bucket.from_bucket_name(self, "Bucket", "amzn-s3-demo-bucket")

fn.add_event_source(AWSCDK::LambdaEventSources::S3EventSourceV2.new(bucket, {
    events: [AWSCDK::S3::EventType::OBJECT_CREATED, AWSCDK::S3::EventType::OBJECT_REMOVED],
    filters: [{prefix: "subdir/"}],
}))

SNS

You can write Lambda functions to process Amazon Simple Notification Service notifications. When a message is published to an Amazon SNS topic, the service can invoke your Lambda function by passing the message payload as a parameter. Your Lambda function code can then process the event, for example publish the message to other Amazon SNS topics, or send the message to other AWS services.

This also enables you to trigger a Lambda function in response to Amazon CloudWatch alarms and other AWS services that use Amazon SNS.

For an example event, see Appendix: Message and JSON Formats and Amazon SNS Sample Event. For an example use case, see Using AWS Lambda with Amazon SNS from Different Accounts.

require 'aws-cdk-lib'

topic = nil # AWSCDK::SNS::Topic

fn = nil # AWSCDK::Lambda::Function

dead_letter_queue = AWSCDK::SQS::Queue.new(self, "deadLetterQueue")
fn.add_event_source(AWSCDK::LambdaEventSources::SNSEventSource.new(topic, {
    filter_policy: {},
    dead_letter_queue: dead_letter_queue,
}))

When a user calls the SNS Publish API on a topic that your Lambda function is subscribed to, Amazon SNS will call Lambda to invoke your function asynchronously. Lambda will then return a delivery status. If there was an error calling Lambda, Amazon SNS will retry invoking the Lambda function up to three times. After three tries, if Amazon SNS still could not successfully invoke the Lambda function, then Amazon SNS will send a delivery status failure message to CloudWatch.

DynamoDB Streams

You can write Lambda functions to process change events from a DynamoDB Table. An event is emitted to a DynamoDB stream (if configured) whenever a write (Put, Delete, Update) operation is performed against the table. See Using AWS Lambda with Amazon DynamoDB for more information about configuring Lambda function event sources with DynamoDB.

To process events with a Lambda function, first create or update a DynamoDB table and enable a stream specification. Then, create a DynamoEventSource and add it to your Lambda function. The following parameters will impact Amazon DynamoDB's polling behavior:

batchSize: Determines how many records are buffered before invoking your lambda function - could impact your function's memory usage (if too high) and ability to keep up with incoming data velocity (if too low).
bisectBatchOnError: If a batch encounters an error, this will cause the batch to be split in two and have each new smaller batch retried, allowing the records in error to be isolated.
reportBatchItemFailures: Allow functions to return partially successful responses for a batch of records.
maxBatchingWindow: The maximum amount of time to gather records before invoking the lambda. This increases the likelihood of a full batch at the cost of delayed processing.
maxRecordAge: The maximum age of a record that will be sent to the function for processing. Records that exceed the max age will be treated as failures.
onFailure: In the event a record fails after all retries or if the record age has exceeded the configured value, the record will be sent to S3 bucket, SQS queue or SNS topic that is specified here
parallelizationFactor: The number of batches to concurrently process on each shard.
retryAttempts: The maximum number of times a record should be retried in the event of failure.
startingPosition: Will determine where to being consumption, either at the most recent ('LATEST') record or the oldest record ('TRIM_HORIZON'). 'TRIM_HORIZON' will ensure you process all available data, while 'LATEST' will ignore all records that arrived prior to attaching the event source.
tumblingWindow: The duration in seconds of a processing window when using streams.
enabled: If the DynamoDB Streams event source mapping should be enabled. The default is true.
filters: Filters to apply before sending a change event from a DynamoDB table to a Lambda function. Events that are filtered out are not sent to the Lambda function.

require 'aws-cdk-lib'

table = nil # AWSCDK::DynamoDB::Table

fn = nil # AWSCDK::Lambda::Function


dead_letter_queue = AWSCDK::SQS::Queue.new(self, "deadLetterQueue")
fn.add_event_source(AWSCDK::LambdaEventSources::DynamoEventSource.new(table, {
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    batch_size: 5,
    bisect_batch_on_error: true,
    on_failure: AWSCDK::LambdaEventSources::SQSDlq.new(dead_letter_queue),
    retry_attempts: 10,
}))

The following code sets up a Lambda function with a DynamoDB event source. A filter is applied to only send DynamoDB events to the Lambda function when the id column is a boolean that equals true.

require 'aws-cdk-lib'

table = nil # AWSCDK::DynamoDB::Table

fn = nil # AWSCDK::Lambda::Function

fn.add_event_source(AWSCDK::LambdaEventSources::DynamoEventSource.new(table, {
    starting_position: AWSCDK::Lambda::StartingPosition::LATEST,
    filters: [
        AWSCDK::Lambda::FilterCriteria.filter({
            event_name: AWSCDK::Lambda::FilterRule.is_equal("INSERT"),
            dynamodb: {
                NewImage: {
                    id: {BOOL: AWSCDK::Lambda::FilterRule.is_equal(true)},
                },
            },
        }),
    ],
}))

Kinesis

You can write Lambda functions to process streaming data in Amazon Kinesis Streams. For more information about Amazon Kinesis, see Amazon Kinesis Service. To learn more about configuring Lambda function event sources with kinesis and view a sample event, see Amazon Kinesis Event.

To set up Amazon Kinesis as an event source for AWS Lambda, you first create or update an Amazon Kinesis stream and select custom values for the event source parameters. The following parameters will impact Amazon Kinesis's polling behavior:

batchSize: Determines how many records are buffered before invoking your lambda function - could impact your function's memory usage (if too high) and ability to keep up with incoming data velocity (if too low).
bisectBatchOnError: If a batch encounters an error, this will cause the batch to be split in two and have each new smaller batch retried, allowing the records in error to be isolated.
reportBatchItemFailures: Allow functions to return partially successful responses for a batch of records.
maxBatchingWindow: The maximum amount of time to gather records before invoking the lambda. This increases the likelihood of a full batch at the cost of possibly delaying processing.
maxRecordAge: The maximum age of a record that will be sent to the function for processing. Records that exceed the max age will be treated as failures.
onFailure: In the event a record fails and consumes all retries, the record will be sent to S3 bucket, SQS queue or SNS topic that is specified here
parallelizationFactor: The number of batches to concurrently process on each shard.
retryAttempts: The maximum number of times a record should be retried in the event of failure.
startingPosition: Will determine where to begin consumption. 'LATEST' will start at the most recent record and ignore all records that arrived prior to attaching the event source, 'TRIM_HORIZON' will start at the oldest record and ensure you process all available data, while 'AT_TIMESTAMP' will start reading records from a specified time stamp.
startingPositionTimestamp: The time stamp from which to start reading. Used in conjunction with startingPosition when set to 'AT_TIMESTAMP'.
tumblingWindow: The duration in seconds of a processing window when using streams.
enabled: If the event source mapping should be enabled. The default is true.

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


stream = AWSCDK::Kinesis::Stream.new(self, "MyStream")
my_function.add_event_source(AWSCDK::LambdaEventSources::KinesisEventSource.new(stream, {
    batch_size: 100,
     # default
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
}))

To use a dedicated-throughput consumer with enhanced fan-out

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


stream = AWSCDK::Kinesis::Stream.new(self, "MyStream")
stream_consumer = AWSCDK::Kinesis::StreamConsumer.new(self, "MyStreamConsumer", {
    stream: stream,
    stream_consumer_name: "MyStreamConsumer",
})
my_function.add_event_source(AWSCDK::LambdaEventSources::KinesisConsumerEventSource.new(stream_consumer, {
    batch_size: 100,
     # default
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
}))

Kafka

You can write Lambda functions to process data either from Amazon MSK or a self-managed Kafka cluster. The following parameters will impact to the polling behavior:

startingPosition: Will determine where to begin consumption. 'LATEST' will start at the most recent record and ignore all records that arrived prior to attaching the event source, 'TRIM_HORIZON' will start at the oldest record and ensure you process all available data, while 'AT_TIMESTAMP' will start reading records from a specified time stamp.
startingPositionTimestamp: The time stamp from which to start reading. Used in conjunction with startingPosition when set to 'AT_TIMESTAMP'.
batchSize: Determines how many records are buffered before invoking your lambda function - could impact your function's memory usage (if too high) and ability to keep up with incoming data velocity (if too low).
maxBatchingWindow: The maximum amount of time to gather records before invoking the lambda. This increases the likelihood of a full batch at the cost of possibly delaying processing.
onFailure: In the event a record fails and consumes all retries, the record will be sent to SQS queue or SNS topic that is specified here
enabled: If the Kafka event source mapping should be enabled. The default is true.
bisectBatchOnError: If a batch encounters an error, this will cause the batch to be split in two and have each new smaller batch retried, allowing the records in error to be isolated. Available in provisioned mode only.
reportBatchItemFailures: Allow functions to return partially successful responses for a batch of records. Available in provisioned mode only.
retryAttempts: The maximum number of times a record should be retried in the event of failure. Available in provisioned mode only.
maxRecordAge: The maximum age of a record that will be sent to the function for processing. Records that exceed the max age will be treated as failures. Available in provisioned mode only.
provisionedPollerConfig: Controls the number of pollers that can be provisioned to read from the event source. Valid ranges: minimum_pollers 1–200 (default: 1), maximum_pollers 1–2000 (default: 200).

The following code sets up Amazon MSK as an event source for a lambda function. Credentials will need to be configured to access the MSK cluster, as described in Username/Password authentication.

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


# Your MSK cluster arn
cluster_arn = "arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4"

# The Kafka topic you want to subscribe to
topic = "some-cool-topic"

# The secret that allows access to your MSK cluster
secret = AWSCDK::SecretsManager::Secret.new(self, "Secret", {secret_name: "AmazonMSK_KafkaSecret"})
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    secret: secret,
    batch_size: 100,
     # default
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    bisect_batch_on_error: true,
    report_batch_item_failures: true,
    retry_attempts: 3,
    max_record_age: AWSCDK::Duration.hours(24),
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 3,
    },
}))

The following code sets up a self managed Kafka cluster as an event source. Username and password based authentication will need to be set up as described in Managing access and permissions.

require 'aws-cdk-lib'

# The secret that allows access to your self hosted Kafka cluster
secret = nil # AWSCDK::SecretsManager::Secret

my_function = nil # AWSCDK::Lambda::Function


# The list of Kafka brokers
bootstrap_servers = ["kafka-broker:9092"]

# The Kafka topic you want to subscribe to
topic = "some-cool-topic"

# (Optional) The consumer group id to use when connecting to the Kafka broker. If omitted the UUID of the event source mapping will be used.
consumer_group_id = "my-consumer-group-id"
my_function.add_event_source(AWSCDK::LambdaEventSources::SelfManagedKafkaEventSource.new({
    bootstrap_servers: bootstrap_servers,
    topic: topic,
    consumer_group_id: consumer_group_id,
    secret: secret,
    batch_size: 100,
     # default
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    bisect_batch_on_error: true,
    report_batch_item_failures: true,
    retry_attempts: 3,
    max_record_age: AWSCDK::Duration.hours(24),
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 3,
    },
}))

If your self managed Kafka cluster is only reachable via VPC also configure vpc vpc_subnets and security_group.

You can specify event filtering for managed and self managed Kafka clusters using the filters property:

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


# Your MSK cluster arn
cluster_arn = "arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4"

# The Kafka topic you want to subscribe to
topic = "some-cool-topic"
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    filters: [
        AWSCDK::Lambda::FilterCriteria.filter({
            string_equals: AWSCDK::Lambda::FilterRule.is_equal("test"),
        }),
    ],
}))

By default, Lambda will encrypt Filter Criteria using AWS managed keys. But if you want to use a self managed KMS key to encrypt the filters, You can specify the self managed key using the filter_encryption property.

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


# Your MSK cluster arn
cluster_arn = "arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4"

# The Kafka topic you want to subscribe to
topic = "some-cool-topic"

# Your self managed KMS key
my_key = AWSCDK::KMS::Key.from_key_arn(self, "SourceBucketEncryptionKey", "arn:aws:kms:us-east-1:123456789012:key/<key-id>")
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    filters: [
        AWSCDK::Lambda::FilterCriteria.filter({
            string_equals: AWSCDK::Lambda::FilterRule.is_equal("test"),
        }),
    ],
    filter_encryption: my_key,
}))

Failure Destinations

You can specify failure destinations for records that fail processing. Kafka event sources support Kafka Topic Destinations, S3 Bucket Destinations, SQS Queue and SNS topic:

Kafka Topic Destination

For Kafka event sources, you can send failed records to another Kafka topic using KafkaDlq:

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


# Your MSK cluster arn
cluster_arn = "arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4"

# The Kafka topic you want to subscribe to
topic = "some-cool-topic"

# Create a Kafka DLQ destination
kafka_dlq = AWSCDK::LambdaEventSources::KafkaDlq.new("failure-topic")

my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    on_failure: kafka_dlq,
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 1,
    },
}))

The same approach works with self-managed Kafka:

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


bootstrap_servers = ["kafka-broker:9092"]
topic = "some-cool-topic"

my_function.add_event_source(AWSCDK::LambdaEventSources::SelfManagedKafkaEventSource.new({
    bootstrap_servers: bootstrap_servers,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    on_failure: AWSCDK::LambdaEventSources::KafkaDlq.new("error-topic"),
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 1,
    },
}))

S3 Bucket Destination

You can also specify an S3 bucket as an "on failure" destination:

require 'aws-cdk-lib'

bucket = nil # AWSCDK::S3::IBucket
my_function = nil # AWSCDK::Lambda::Function


# Your MSK cluster arn
cluster_arn = "arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4"

# The Kafka topic you want to subscribe to
topic = "some-cool-topic"

s3_on_failure_destination = AWSCDK::LambdaEventSources::S3OnFailureDestination.new(bucket)

my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    on_failure: s3_on_failure_destination,
}))

Kafka Observability Features

AWS Lambda provides enhanced observability for Kafka event sources through logging and metrics configuration.

Important: Observability features (LogLevel and MetricsConfig) are only available when using provisioned mode.

Logging

You can configure the verbosity of logs generated by the polling infrastructure. This is particularly useful for troubleshooting connection issues, monitoring polling behavior, and understanding the internal operations of your event source mapping.

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


# Your MSK cluster arn
cluster_arn = "arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4"

# Configure INFO level logging for production monitoring
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: "production-events",
    starting_position: AWSCDK::Lambda::StartingPosition::LATEST,
    # Provisioned mode is required for observability features
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 5,
    },
    log_level: AWSCDK::Lambda::EventSourceMappingLogLevel::INFO,
}))

Metrics Configuration

Enhanced metrics provide detailed insights into your Kafka event source performance. Metrics include event processing rates, error counts, and Kafka-specific metrics like consumer lag.

require 'aws-cdk-lib'

my_function = nil # AWSCDK::Lambda::Function


# Your MSK cluster arn
cluster_arn = "arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4"

# Enable basic event and error metrics
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: "basic-monitoring",
    starting_position: AWSCDK::Lambda::StartingPosition::LATEST,
    # Provisioned mode is required for observability features
    provisioned_poller_config: {
        minimum_pollers: 2,
        maximum_pollers: 10,
    },
    metrics_config: {
        metrics: [
            AWSCDK::Lambda::MetricType::EVENT_COUNT,
            AWSCDK::Lambda::MetricType::ERROR_COUNT,
        ],
    },
}))

Set configuration for provisioned pollers that read from the event source. When specified, allows control over the minimum and maximum number of pollers that can be provisioned to process events from the source.

require 'aws-cdk-lib'

# Your MSK cluster arn
cluster_arn = nil

my_function = nil # AWSCDK::Lambda::Function


# The Kafka topic you want to subscribe to
topic = "some-cool-topic"
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 3,
    },
}))

You can reduce costs by sharing provisioned pollers across multiple Kafka event sources using the poller_group_name property. This is particularly useful when you have multiple Kafka topics that don't require dedicated polling capacity.

require 'aws-cdk-lib'

cluster_arn = nil
orders_function = nil # AWSCDK::Lambda::Function


# Orders processing function
orders_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: "orders-topic",
    starting_position: AWSCDK::Lambda::StartingPosition::LATEST,
    provisioned_poller_config: {
        minimum_pollers: 2,
        maximum_pollers: 10,
        poller_group_name: "shared-kafka-pollers",
    },
}))

Set a confluent or self-managed schema registry to de-serialize events from the event source.

Note: This will also work for SelfManagedKafkaEventSource.

require 'aws-cdk-lib'

# Your MSK cluster arn
cluster_arn = nil

my_function = nil # AWSCDK::Lambda::Function


# The Kafka topic you want to subscribe to
topic = "some-cool-topic"

secret = AWSCDK::SecretsManager::Secret.new(self, "Secret", {secret_name: "AmazonMSK_KafkaSecret"})
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 3,
    },
    schema_registry_config: AWSCDK::LambdaEventSources::ConfluentSchemaRegistry.new({
        schema_registry_uri: "https://example.com",
        event_record_format: AWSCDK::Lambda::EventRecordFormat.JSON,
        authentication_type: AWSCDK::Lambda::KafkaSchemaRegistryAccessConfigType.BASIC_AUTH,
        secret: secret,
        schema_validation_configs: [{attribute: AWSCDK::Lambda::KafkaSchemaValidationAttribute.KEY}],
    }),
}))

Set Glue schema registry to de-serialize events from the event source.

Note: This will also work for SelfManagedKafkaEventSource.

require 'aws-cdk-lib'

# Your MSK cluster arn
cluster_arn = nil

my_function = nil # AWSCDK::Lambda::Function


# The Kafka topic you want to subscribe to
topic = "some-cool-topic"

# Your Glue Schema Registry
glue_registry = AWSCDK::Glue::CfnRegistry.new(self, "Registry", {
    name: "schema-registry",
    description: "Schema registry for event source",
})
my_function.add_event_source(AWSCDK::LambdaEventSources::ManagedKafkaEventSource.new({
    cluster_arn: cluster_arn,
    topic: topic,
    starting_position: AWSCDK::Lambda::StartingPosition::TRIM_HORIZON,
    provisioned_poller_config: {
        minimum_pollers: 1,
        maximum_pollers: 3,
    },
    schema_registry_config: AWSCDK::LambdaEventSources::GlueSchemaRegistry.new({
        schema_registry: glue_registry,
        event_record_format: AWSCDK::Lambda::EventRecordFormat.JSON,
        schema_validation_configs: [{attribute: AWSCDK::Lambda::KafkaSchemaValidationAttribute.KEY}],
    }),
}))

Roadmap

Eventually, this module will support all the event sources described under Supported Event Sources in the AWS Lambda Developer Guide.

API Reference

Classes 17

Interfaces 13

BaseStreamEventSourcePropsThe set of properties for streaming event sources shared by Dynamo, Kinesis and Kafka. ConfluentSchemaRegistryPropsProperties for confluent schema registry configuration. DynamoEventSourceProps GlueSchemaRegistryPropsProperties for glue schema registry configuration. KafkaEventSourcePropsProperties for a Kafka event source. KinesisEventSourceProps ManagedKafkaEventSourcePropsProperties for a MSK event source. ProvisionedPollerConfig(Amazon MSK and self-managed Apache Kafka only) The provisioned mode configuration for the S3EventSourceProps SelfManagedKafkaEventSourcePropsProperties for a self managed Kafka cluster event source. SNSEventSourcePropsProperties forwarded to the Lambda Subscription. SQSEventSourceProps StreamEventSourcePropsThe set of properties for streaming event sources shared by Dynamo and Kinesis.

Enums 1

AuthenticationMethodThe authentication method to use with SelfManagedKafkaEventSource.