ConfluentSchemaRegistry Class

Confluent schema registry configuration for a Lambda event source.

Source: aws-lambda-event-sources/lib/confluent-schema-registry.ts
Type last changed: v2.202.0 · 20 Jun 2025

Example

import { ManagedKafkaEventSource, ConfluentSchemaRegistry } from 'aws-cdk-lib/aws-lambda-event-sources';
import { Secret } from 'aws-cdk-lib/aws-secretsmanager';

// Your MSK cluster arn
declare const clusterArn: string;

// The Kafka topic you want to subscribe to
const topic = 'some-cool-topic';

const secret = new Secret(this, 'Secret', { secretName: 'AmazonMSK_KafkaSecret' });

declare const myFunction: lambda.Function;
myFunction.addEventSource(new ManagedKafkaEventSource({
  clusterArn,
  topic,
  startingPosition: lambda.StartingPosition.TRIM_HORIZON,
  provisionedPollerConfig: {
    minimumPollers: 1,
    maximumPollers: 3,
  },
  schemaRegistryConfig: new ConfluentSchemaRegistry({
    schemaRegistryUri: 'https://example.com',
    eventRecordFormat: lambda.EventRecordFormat.JSON,
    authenticationType: lambda.KafkaSchemaRegistryAccessConfigType.BASIC_AUTH,
    secret: secret,
    schemaValidationConfigs: [{ attribute: lambda.KafkaSchemaValidationAttribute.KEY }],
  }),
}));

python

from aws_cdk.aws_lambda_event_sources import ProvisionedPollerConfig
from aws_cdk.aws_lambda_event_sources import ManagedKafkaEventSource, ConfluentSchemaRegistry
from aws_cdk.aws_secretsmanager import Secret

# Your MSK cluster arn
# cluster_arn: str

# my_function: lambda.Function


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

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

java

import software.amazon.awscdk.services.lambda.eventsources.ManagedKafkaEventSource;
import software.amazon.awscdk.services.lambda.eventsources.ConfluentSchemaRegistry;
import software.amazon.awscdk.services.secretsmanager.Secret;

// Your MSK cluster arn
String clusterArn;

Function myFunction;


// The Kafka topic you want to subscribe to
String topic = "some-cool-topic";

Secret secret = Secret.Builder.create(this, "Secret").secretName("AmazonMSK_KafkaSecret").build();
myFunction.addEventSource(ManagedKafkaEventSource.Builder.create()
        .clusterArn(clusterArn)
        .topic(topic)
        .startingPosition(StartingPosition.TRIM_HORIZON)
        .provisionedPollerConfig(ProvisionedPollerConfig.builder()
                .minimumPollers(1)
                .maximumPollers(3)
                .build())
        .schemaRegistryConfig(ConfluentSchemaRegistry.Builder.create()
                .schemaRegistryUri("https://example.com")
                .eventRecordFormat(EventRecordFormat.JSON)
                .authenticationType(KafkaSchemaRegistryAccessConfigType.BASIC_AUTH)
                .secret(secret)
                .schemaValidationConfigs(List.of(KafkaSchemaValidationConfig.builder().attribute(KafkaSchemaValidationAttribute.KEY).build()))
                .build())
        .build());

csharp

// Install: dotnet add package Amazon.CDK.Lib
using Amazon.CDK.AWS.Lambda.EventSources;
using Amazon.CDK.AWS.SecretsManager;

// Your MSK cluster arn
string clusterArn;

Function myFunction;


// The Kafka topic you want to subscribe to
var topic = "some-cool-topic";

var secret = new Secret(this, "Secret", new SecretProps { SecretName = "AmazonMSK_KafkaSecret" });
myFunction.AddEventSource(new ManagedKafkaEventSource(new ManagedKafkaEventSourceProps {
    ClusterArn = clusterArn,
    Topic = topic,
    StartingPosition = StartingPosition.TRIM_HORIZON,
    ProvisionedPollerConfig = new ProvisionedPollerConfig {
        MinimumPollers = 1,
        MaximumPollers = 3
    },
    SchemaRegistryConfig = new ConfluentSchemaRegistry(new ConfluentSchemaRegistryProps {
        SchemaRegistryUri = "https://example.com",
        EventRecordFormat = EventRecordFormat.JSON,
        AuthenticationType = KafkaSchemaRegistryAccessConfigType.BASIC_AUTH,
        Secret = secret,
        SchemaValidationConfigs = new [] { new KafkaSchemaValidationConfig { Attribute = KafkaSchemaValidationAttribute.KEY } }
    })
}));

// Install: go get github.com/aws/aws-cdk-go/awscdk/v2
import "github.com/aws/aws-cdk-go/awscdk"
import "github.com/aws/aws-cdk-go/awscdk"

// Your MSK cluster arn
var clusterArn string

var myFunction Function


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

secret := awscdk.NewSecret(this, jsii.String("Secret"), &SecretProps{
	SecretName: jsii.String("AmazonMSK_KafkaSecret"),
})
myFunction.AddEventSource(awscdk.NewManagedKafkaEventSource(&ManagedKafkaEventSourceProps{
	ClusterArn: jsii.String(ClusterArn),
	Topic: jsii.String(Topic),
	StartingPosition: lambda.StartingPosition_TRIM_HORIZON,
	ProvisionedPollerConfig: &ProvisionedPollerConfig{
		MinimumPollers: jsii.Number(1),
		MaximumPollers: jsii.Number(3),
	},
	SchemaRegistryConfig: awscdk.NewConfluentSchemaRegistry(&ConfluentSchemaRegistryProps{
		SchemaRegistryUri: jsii.String("https://example.com"),
		EventRecordFormat: lambda.EventRecordFormat_JSON(),
		AuthenticationType: lambda.KafkaSchemaRegistryAccessConfigType_BASIC_AUTH(),
		Secret: secret,
		SchemaValidationConfigs: []KafkaSchemaValidationConfig{
			&KafkaSchemaValidationConfig{
				Attribute: lambda.KafkaSchemaValidationAttribute_KEY(),
			},
		},
	}),
}))

Initializer

new ConfluentSchemaRegistry(props: ConfluentSchemaRegistryProps)

python

ConfluentSchemaRegistry(props: ConfluentSchemaRegistryProps)

java

new ConfluentSchemaRegistry(ConfluentSchemaRegistryProps props)

csharp

new ConfluentSchemaRegistry(ConfluentSchemaRegistryProps props)

NewConfluentSchemaRegistry(Props ConfluentSchemaRegistryProps)

Methods

bind(_target: IEventSourceMapping, targetHandler: IFunction): KafkaSchemaRegistryConfigdef bind(self, _target: IEventSourceMapping, target_handler: IFunction) -> KafkaSchemaRegistryConfigpublic KafkaSchemaRegistryConfig bind(IEventSourceMapping _target, IFunction targetHandler)public KafkaSchemaRegistryConfig Bind(IEventSourceMapping _target, IFunction targetHandler)func Bind(_target IEventSourceMapping, TargetHandler IFunction) KafkaSchemaRegistryConfig#

Returns a schema registry configuration.

ConfluentSchemaRegistry Class ​

Example ​

Initializer ​

Methods ​

ConfluentSchemaRegistry Class

Example

Initializer

Methods