ConfluentSchemaRegistryProps Struct
Properties for confluent schema registry configuration.
Example
ts
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 } }
})
}));
go
// 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(),
},
},
}),
}))
Properties
authenticationTypeKafkaSchemaRegistryAccessConfigTypeauthentication_typeKafkaSchemaRegistryAccessConfigTypeauthenticationTypeKafkaSchemaRegistryAccessConfigTypeAuthenticationTypeKafkaSchemaRegistryAccessConfigTypeAuthenticationTypeKafkaSchemaRegistryAccessConfigType= nonerequired#The type of authentication for schema registry credentials.
eventRecordFormatEventRecordFormatevent_record_formatEventRecordFormateventRecordFormatEventRecordFormatEventRecordFormatEventRecordFormatEventRecordFormatEventRecordFormat= nonerequired#The record format that Lambda delivers to your function after schema validation.
notes
- Choose JSON to have Lambda deliver the record to your function as a standard JSON object.
- Choose SOURCE to have Lambda deliver the record to your function in its original source format. Lambda removes all schema metadata, such as the schema ID, before sending the record to your function.
schemaRegistryUristringschema_registry_uristrschemaRegistryUriStringSchemaRegistryUristringSchemaRegistryUri*string= nonerequired#The URI for your schema registry.
schemaValidationConfigsKafkaSchemaValidationConfig[]schema_validation_configstyping.List[KafkaSchemaValidationConfig]schemaValidationConfigsList<KafkaSchemaValidationConfig>SchemaValidationConfigsKafkaSchemaValidationConfig[]SchemaValidationConfigs*[]KafkaSchemaValidationConfig= nonerequired#An array of schema validation configuration objects, which tell Lambda the message attributes you want to validate and filter using your schema registry.