Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà. # Utilizzo di registri di schema con sorgenti di eventi Kafka in Lambda I registri degli schemi ti aiutano a definire e gestire gli schemi dei flussi di dati. Uno schema definisce la struttura e il formato di un registro di dati. Nel contesto delle mappature delle sorgenti degli eventi Kafka, puoi configurare un registro degli schemi per convalidare la struttura e il formato dei messaggi Kafka rispetto a schemi predefiniti prima che raggiungano la tua funzione Lambda. Ciò aggiunge un livello di governance dei dati all'applicazione e consente di gestire in modo efficiente i formati di dati, garantire la conformità degli schemi e ottimizzare i costi tramite il filtro degli eventi. Questa funzionalità funziona con tutti i linguaggi di programmazione, ma considera questi punti importanti: + Powertools for Lambda fornisce un supporto specifico per Java, Python e TypeScript, mantenendo la coerenza con i modelli di sviluppo esistenti di Kafka e consentendo l'accesso diretto agli oggetti aziendali senza codice di deserializzazione personalizzato + Questa funzionalità è disponibile solo per le mappature delle sorgenti degli eventi utilizzando la modalità provisioned. Il registro degli schemi non supporta le mappature delle origini degli eventi in modalità su richiesta. Se si utilizza la modalità provisioning e si dispone di un registro degli schemi configurato, non è possibile passare alla modalità su richiesta a meno che non si rimuova prima la configurazione del registro degli schemi. Per ulteriori informazioni, consulta [Modalità provisioning](invocation-eventsourcemapping.md#invocation-eventsourcemapping-provisioned-mode) + È possibile configurare un solo registro di schemi per Event Source Mapping (ESM). L'utilizzo di un registro di schemi con la fonte di eventi Kafka potrebbe aumentare l'utilizzo della Lambda Event Poller Unit (EPU), che è una dimensione di determinazione dei prezzi per la modalità Provisioned. **Topics** + [Opzioni del registro dello schema](#services-consume-kafka-events-options) + [In che modo Lambda esegue la convalida dello schema per i messaggi Kafka](#services-consume-kafka-events-how) + [Configurazione di un registro di schemi Kafka](#services-consume-kafka-events-config) + [Filtraggio per Avro e Protobuf](#services-consume-kafka-events-filtering) + [Formati di payload e comportamento di deserializzazione](#services-consume-kafka-events-payload) + [Lavorare con dati deserializzati nelle funzioni Lambda](#services-consume-kafka-events-payload-examples) + [Metodi di autenticazione per il registro degli schemi](#services-consume-kafka-events-auth) + [Gestione degli errori e risoluzione dei problemi relativi al registro dello schema](#services-consume-kafka-events-troubleshooting) ## Opzioni del registro dello schema Lambda supporta le seguenti opzioni di registro degli schemi: + [AWS Glue Registro degli schemi ](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html) + [Registro degli schemi di Confluent Cloud ](https://docs.confluent.io/platform/current/schema-registry/index.html) + [Self-managed Registro degli schemi confluenti ](https://docs.confluent.io/platform/current/schema-registry/index.html) Il registro degli schemi supporta la convalida dei messaggi nei seguenti formati di dati: + Apache Avro + Buffer di protocollo (Protobuf) + Schema JSON () JSON-SE Per utilizzare un registro degli schemi, assicuratevi innanzitutto che la mappatura dell'origine degli eventi sia in modalità provisioning. Quando si utilizza un registro di schemi, Lambda aggiunge i metadati sullo schema al payload. Per ulteriori informazioni, consulta Formati di [ payload e comportamento di deserializzazione. ](#services-consume-kafka-events-payload) ## In che modo Lambda esegue la convalida dello schema per i messaggi Kafka Quando si configura un registro degli schemi, Lambda esegue i seguenti passaggi per ogni messaggio Kafka: 1. Lambda esegue il polling del record Kafka dal cluster. 1. Lambda convalida gli attributi dei messaggi selezionati nel record rispetto a uno schema specifico nel registro degli schemi. + Se lo schema associato al messaggio non viene trovato nel registro, Lambda invia il messaggio a un DLQ con il codice del motivo. `SCHEMA_NOT_FOUND` 1. Lambda deserializza il messaggio in base alla configurazione del registro dello schema per convalidare il messaggio. Se il filtro degli eventi è configurato, Lambda esegue quindi il filtraggio in base ai criteri di filtro configurati. + Se la deserializzazione fallisce, Lambda invia il messaggio a un DLQ con il codice del motivo. `DESERIALIZATION_ERROR` Se non è configurato alcun DLQ, Lambda elimina il messaggio. 1. Se il messaggio viene convalidato dal registro degli schemi e non viene filtrato in base ai criteri di filtro, Lambda richiama la funzione con il messaggio. Questa funzionalità ha lo scopo di convalidare i messaggi già prodotti utilizzando client Kafka integrati con un registro di schemi. Consigliamo di configurare i produttori Kafka in modo che funzionino con il registro degli schemi per creare messaggi formattati correttamente. ## Configurazione di un registro di schemi Kafka I seguenti passaggi della console aggiungono una configurazione del registro dello schema Kafka alla mappatura dell'origine degli eventi. **Per aggiungere una configurazione del registro dello schema Kafka alla mappatura della sorgente degli eventi (console)** 1. Apri la [pagina Funzione](https://console.aws.amazon.com/lambda/home#/functions) della console Lambda. 1. Scegliere **Configuration (Configurazione)**. 1. Scegli Triggers. ** ** 1. Seleziona la mappatura della sorgente degli eventi Kafka per la quale desideri configurare un registro degli schemi e scegli Modifica. ** ** 1. In ** Event poller configuration**, scegli ** Configura il registro degli schemi. ** La mappatura dell'origine degli eventi deve essere in modalità provisioning per visualizzare questa opzione. 1. Per l'URI del registro di ** schema**, inserisci l'ARN del tuo registro degli schemi o l'URL HTTPS del tuo registro di AWS Glue schema Confluent Cloud o del registro degli schemi di Confluent. Self-Managed 1. I seguenti passaggi di configurazione indicano a Lambda come accedere al registro degli schemi. Per ulteriori informazioni, consulta [Metodi di autenticazione per il registro degli schemi](#services-consume-kafka-events-auth). + Per il tipo di configurazione di ** Access**, scegli il tipo di autenticazione utilizzato da Lambda per accedere al registro degli schemi. + Per ** l'URI di configurazione di Access**, inserisci l'ARN del segreto di Secrets Manager per autenticarti con il registro dello schema, se applicabile. Assicurati che il ruolo di [ esecuzione della tua funzione ](with-msk-permissions.md) contenga le autorizzazioni corrette. 1. Il ** campo ** Encryption si applica solo se il registro dello schema è firmato da un'autorità di certificazione (CA) privata o da un'autorità di certificazione (CA) non presente nel trust store Lambda. Se applicabile, fornisci la chiave segreta contenente il certificato CA privato utilizzato dal registro dello schema per la crittografia TLS. 1. Per il formato dei record ** degli eventi**, scegli come desideri che Lambda fornisca i record alla tua funzione dopo la convalida dello schema. Per ulteriori informazioni, consulta gli [esempi di formati di payload](#services-consume-kafka-events-payload). + Se scegli ** JSON**, Lambda fornisce gli attributi selezionati nell'attributo di convalida dello schema riportato di seguito in formato JSON standard. Per gli attributi che non selezioni, Lambda li fornisce così come sono. + Se scegli ** SOURCE**, Lambda fornisce gli attributi selezionati nell'attributo di convalida dello schema riportato di seguito nel formato sorgente originale. 1. Per l'attributo di convalida ** dello schema**, seleziona gli attributi del messaggio che desideri che Lambda convalidi e deserializzi utilizzando il registro degli schemi. È necessario selezionare almeno una delle opzioni KEY o VALUE. ** ** ** ** Se hai scelto JSON come formato di registrazione degli eventi, Lambda deserializza anche gli attributi dei messaggi selezionati prima di inviarli alla tua funzione. Per ulteriori informazioni, consulta Formati di [ payload e comportamento di deserializzazione. ](#services-consume-kafka-events-payload) 1. Scegli **Save** (Salva). Puoi anche utilizzare l'API Lambda per creare o aggiornare la mappatura della sorgente degli eventi con una configurazione del registro dello schema. Gli esempi seguenti mostrano come configurare un registro di schemi AWS Glue o Confluent utilizzando il AWS CLI, che corrisponde alle operazioni [ CreateEventSourceMapping ](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) API [ UpdateEventSourceMapping ](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html) e nell'API Reference: *AWS Lambda * **Importante** Se si sta aggiornando un campo di configurazione del registro dello schema utilizzando AWS CLI o l'`update-event-source-mapping`API, è necessario aggiornare tutti i campi della configurazione del registro di schema. ------ #### [ Create Event Source Mapping ] ``` aws lambda create-event-source-mapping \ --function-name my-schema-validator-function \ --event-source-arn arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/a1b2c3d4-5678-90ab-cdef-11111EXAMPLE \ --topics my-kafka-topic \ --provisioned-poller-config MinimumPollers=1,MaximumPollers=1 \ --amazon-managed-kafka-event-source-mapping '{ "SchemaRegistryConfig" : { "SchemaRegistryURI": "https://abcd-ef123.us-west-2.aws.confluent.cloud", "AccessConfigs": [{ "Type": "BASIC_AUTH", "URI": "arn:aws:secretsmanager:us-east-1:123456789012:secret:secretName" }], "EventRecordFormat": "JSON", "SchemaValidationConfigs": [ { "Attribute": "KEY" }, { "Attribute": "VALUE" }] } }' ``` ------ #### [ Update AWS Glue Schema Registry ] ``` aws lambda update-event-source-mapping \ --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \ --amazon-managed-kafka-event-source-mapping '{ "SchemaRegistryConfig" : { "SchemaRegistryURI": "arn:aws:glue:us-east-1:123456789012:registry/registryName", "EventRecordFormat": "JSON", "SchemaValidationConfigs": [ { "Attribute": "KEY" }, { "Attribute": "VALUE" }] } }' ``` ------ #### [ Update Confluent Schema Registry with Authentication ] ``` aws lambda update-event-source-mapping \ --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \ --amazon-managed-kafka-event-source-mapping '{ "SchemaRegistryConfig" : { "SchemaRegistryURI": "https://abcd-ef123.us-west-2.aws.confluent.cloud", "AccessConfigs": [{ "Type": "BASIC_AUTH", "URI": "arn:aws:secretsmanager:us-east-1:123456789012:secret:secretName" }], "EventRecordFormat": "JSON", "SchemaValidationConfigs": [ { "Attribute": "KEY" }, { "Attribute": "VALUE" }] } }' ``` ------ #### [ Update Confluent Schema Registry without Authentication ] ``` aws lambda update-event-source-mapping \ --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \ --amazon-managed-kafka-event-source-mapping '{ "SchemaRegistryConfig" : { "SchemaRegistryURI": "https://abcd-ef123.us-west-2.aws.confluent.cloud", "EventRecordFormat": "JSON", "SchemaValidationConfigs": [ { "Attribute": "KEY" }, { "Attribute": "VALUE" }] } }' ``` ------ #### [ Remove Schema Registry Configuration ] Per rimuovere una configurazione del registro dello schema dalla mappatura dell'origine degli eventi, puoi utilizzare il comando CLI [ UpdateEventSourceMapping ](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html) nell'*AWS Lambda API Reference. * ``` aws lambda update-event-source-mapping \ --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \ --amazon-managed-kafka-event-source-mapping '{ "SchemaRegistryConfig" : {} }' ``` ------ ## Filtraggio per Avro e Protobuf Quando si utilizzano i formati Avro o Protobuf con un registro di schemi, è possibile applicare il filtro degli eventi alla funzione Lambda. I modelli di filtro vengono applicati alla classica rappresentazione JSON deserializzata dei dati dopo la convalida dello schema. Ad esempio, con uno schema Avro che definisce i dettagli del prodotto, incluso il prezzo, puoi filtrare i messaggi in base al valore del prezzo: **Nota** Quando viene deserializzato, Avro viene convertito in JSON standard, il che significa che non può essere riconvertito direttamente in un oggetto Avro. Se devi convertire in un oggetto Avro, usa invece il formato SOURCE. Per la deserializzazione di Protobuf, i nomi dei campi nel JSON risultante corrispondono a quelli definiti nello schema, anziché essere convertiti in maiuscole e minuscole come normalmente fa Protobuf. Tienilo a mente quando crei modelli di filtro. ``` aws lambda create-event-source-mapping \ --function-name myAvroFunction \ --topics myAvroTopic \ --starting-position TRIM_HORIZON \ --kafka-bootstrap-servers '["broker1:9092", "broker2:9092"]' \ --schema-registry-config '{ "SchemaRegistryURI": "arn:aws:glue:us-east-1:123456789012:registry/myAvroRegistry", "EventRecordFormat": "JSON", "SchemaValidationConfigs": [ { "Attribute": "VALUE" } ] }' \ --filter-criteria '{ "Filters": [ { "Pattern": "{ \"value\" : { \"field_1\" : [\"value1\"], \"field_2\" : [\"value2\"] } }" } ] }' ``` In questo esempio, lo schema di filtro analizza l'`value`oggetto, abbinando i messaggi `field_1` con `"value1"` e `field_2` con. `"value2"` I criteri di filtro vengono valutati rispetto ai dati deserializzati, dopo che Lambda ha convertito il messaggio dal formato Avro a JSON. Per informazioni più dettagliate sul filtro degli eventi, consulta Filtraggio degli eventi Lambda. [Controllare gli eventi che Lambda invia alla funzione](invocation-eventfiltering.md) ## Formati di payload e comportamento di deserializzazione Quando si utilizza un registro di schemi, Lambda fornisce il payload finale alla funzione in un formato simile al [ normale payload degli eventi](with-msk.md#msk-sample-event), con alcuni campi aggiuntivi. I campi aggiuntivi dipendono dal parametro. `SchemaValidationConfigs` Per ogni attributo selezionato per la convalida (chiave o valore), Lambda aggiunge i metadati dello schema corrispondenti al payload. **Nota** È necessario aggiornare [https://github.com/aws/aws-lambda-java-libs/tree/main/aws-lambda-java-events](https://github.com/aws/aws-lambda-java-libs/tree/main/aws-lambda-java-events) aws-lambda-java-events alla versione 3.16.0 o successiva per utilizzare i campi di metadati dello schema. Ad esempio, se convalidi il campo, Lambda aggiunge un campo chiamato al tuo payload. `value` `valueSchemaMetadata` Analogamente, per il `key` campo, Lambda aggiunge un campo chiamato. `keySchemaMetadata` Questi metadati contengono informazioni sul formato dei dati e sull'ID dello schema utilizzato per la convalida: ``` "valueSchemaMetadata": { "dataFormat": "AVRO", "schemaId": "a1b2c3d4-5678-90ab-cdef-EXAMPLE11111" } ``` Il `EventRecordFormat` parametro può essere impostato su `JSON` o`SOURCE`, il che determina in che modo Lambda gestisce i dati convalidati dallo schema prima di consegnarli alla funzione. Ogni opzione offre diverse funzionalità di elaborazione: + `JSON`- Lambda deserializza gli attributi convalidati in formato JSON standard, rendendo i dati pronti per l'uso diretto nei linguaggi con supporto JSON nativo. Questo formato è ideale quando non è necessario preservare il formato binario originale o lavorare con classi generate. + `SOURCE`- Lambda conserva il formato binario originale dei dati come Base64-encoded stringa, consentendo la conversione diretta in oggetti Avro o Protobuf. Questo formato è essenziale quando si lavora con linguaggi fortemente tipizzati o quando è necessario mantenere tutte le funzionalità degli schemi Avro o Protobuf. In base a queste caratteristiche di formato e a considerazioni specifiche della lingua, consigliamo i seguenti formati: **Formati consigliati basati sul linguaggio di programmazione** | Lingua | Avro | Protobuf | JSON | | --- | --- | --- | --- | | Java | SOURCE | SOURCE | SOURCE | | Python | JSON | JSON | JSON | | NodeJS | JSON | JSON | JSON | | .NET | SOURCE | SOURCE | SOURCE | | Altri | JSON | JSON | JSON | Le sezioni seguenti descrivono questi formati in dettaglio e forniscono esempi di payload per ogni formato. ### Formato JSON Se scegli `JSON` come`EventRecordFormat`, Lambda convalida e deserializza gli attributi del messaggio che hai selezionato nel campo (gli attributi). `SchemaValidationConfigs` `key` and/or `value` Lambda fornisce questi attributi selezionati come stringhe codificate in base64 della loro rappresentazione JSON standard nella funzione. **Nota** Quando viene deserializzato, Avro viene convertito in JSON standard, il che significa che non può essere riconvertito direttamente in un oggetto Avro. Se devi convertire in un oggetto Avro, usa invece il formato SOURCE. Per la deserializzazione di Protobuf, i nomi dei campi nel JSON risultante corrispondono a quelli definiti nello schema, anziché essere convertiti in maiuscole e minuscole come normalmente fa Protobuf. Tienilo a mente quando crei modelli di filtro. Di seguito viene mostrato un esempio di payload, supponendo che tu scelga `JSON` come `EventRecordFormat` ed entrambi gli attributi `key` e `value` come `SchemaValidationConfigs`: ``` { "eventSource":"aws:kafka", "eventSourceArn":"arn:aws:kafka:us-east-1:123456789012:cluster/vpc-2priv-2pub/a1b2c3d4-5678-90ab-cdef-EXAMPLE11111-1", "bootstrapServers":"b-2.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092,b-1.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092", "records":{ "mytopic-0":[ { "topic":"mytopic", "partition":0, "offset":15, "timestamp":1545084650987, "timestampType":"CREATE_TIME", "key":"abcDEFghiJKLmnoPQRstuVWXyz1234==", //Base64 encoded string of JSON "keySchemaMetadata": { "dataFormat": "AVRO", "schemaId": "a1b2c3d4-5678-90ab-cdef-EXAMPLE11111" }, "value":"abcDEFghiJKLmnoPQRstuVWXyz1234", //Base64 encoded string of JSON "valueSchemaMetadata": { "dataFormat": "AVRO", "schemaId": "a1b2c3d4-5678-90ab-cdef-EXAMPLE11111" }, "headers":[ { "headerKey":[ 104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101 ] } ] } ] } } ``` In questo esempio: + Entrambe `key` `value` sono stringhe codificate in base64 della loro rappresentazione JSON dopo la deserializzazione. + Lambda include i metadati dello schema per entrambi gli attributi in e. `keySchemaMetadata` `valueSchemaMetadata` + La tua funzione può decodificare le stringhe `key` e `value` per accedere ai dati JSON deserializzati. Il formato JSON è consigliato per i linguaggi che non sono fortemente digitati, come Python o. Node.js Questi linguaggi hanno un supporto nativo per la conversione di JSON in oggetti. ### Formato sorgente Se lo scegli `SOURCE``EventRecordFormat`, Lambda convalida comunque il record rispetto al registro dello schema, ma fornisce i dati binari originali alla tua funzione senza deserializzazione. Questi dati binari vengono forniti come stringa codificata Base64 dei dati in byte originali, con i metadati aggiunti dal produttore rimossi. Di conseguenza, è possibile convertire direttamente i dati binari grezzi in oggetti Avro e Protobuf all'interno del codice funzione. Consigliamo di utilizzare Powertools for AWS Lambda, che deserializza i dati binari grezzi e fornisce direttamente gli oggetti Avro e Protobuf. Ad esempio, se configuri Lambda per convalidare entrambi gli attributi `key` e `value` ma utilizzi il formato `SOURCE`, la tua funzione riceve un payload come questo: ``` { "eventSource": "aws:kafka", "eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/vpc-2priv-2pub/a1b2c3d4-5678-90ab-cdef-EXAMPLE11111-1", "bootstrapServers": "b-2.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092,b-1.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092", "records": { "mytopic-0": [ { "topic": "mytopic", "partition": 0, "offset": 15, "timestamp": 1545084650987, "timestampType": "CREATE_TIME", "key": "abcDEFghiJKLmnoPQRstuVWXyz1234==", // Base64 encoded string of Original byte data, producer-appended metadata removed "keySchemaMetadata": { "dataFormat": "AVRO", "schemaId": "a1b2c3d4-5678-90ab-cdef-EXAMPLE11111" }, "value": "abcDEFghiJKLmnoPQRstuVWXyz1234==", // Base64 encoded string of Original byte data, producer-appended metadata removed "valueSchemaMetadata": { "dataFormat": "AVRO", "schemaId": "a1b2c3d4-5678-90ab-cdef-EXAMPLE11111" }, "headers": [ { "headerKey": [ 104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101 ] } ] } ] } } ``` In questo esempio: + Sia `key` che `value` contengono i dati binari originali come stringhe codificate in Base64. + La tua funzione deve gestire la deserializzazione utilizzando le librerie appropriate. La `SOURCE` scelta `EventRecordFormat` è consigliata se si utilizzano Protobuf-generated oggetti, Avro-generated in particolare con funzioni Java. Questo perché Java è fortemente tipizzato e richiede deserializzatori specifici per i formati Avro e Protobuf. Nel codice funzione, puoi utilizzare la tua libreria Avro o Protobuf preferita per deserializzare i dati. ## Lavorare con dati deserializzati nelle funzioni Lambda Powertools for ti AWS Lambda aiuta a deserializzare i record Kafka nel tuo codice funzione in base al formato che usi. Questa utilità semplifica l'utilizzo dei record Kafka gestendo la conversione dei dati e fornendo oggetti pronti all'uso. Per utilizzare Powertools for AWS Lambda nella tua funzione, devi aggiungere Powertools for come livello o includerlo come AWS Lambda dipendenza quando crei la tua funzione Lambda. Per istruzioni di configurazione e ulteriori informazioni, consulta Powertools per la AWS Lambda documentazione relativa alla tua lingua preferita: + [Powertools per AWS Lambda (Java) ](https://docs.aws.amazon.com/powertools/java/latest/utilities/kafka/) + [Powertools per AWS Lambda (Python) ](https://docs.aws.amazon.com/powertools/python/latest/utilities/kafka/) + [Powertools per () AWS Lambda TypeScript ](https://docs.aws.amazon.com/powertools/typescript/latest/features/kafka/) + [Powertools per AWS Lambda (.NET) ](https://docs.aws.amazon.com/powertools/dotnet/utilities/kafka/) **Nota** Quando si lavora con l'integrazione del registro degli schemi, è possibile scegliere `SOURCE` o `JSON` formattare. Ogni opzione supporta diversi formati di serializzazione, come illustrato di seguito: | Formato | Supporta | | --- | --- | | SOURCE | Avro e Protobuf (utilizzando l'integrazione del Lambda Schema Registry) | | JSON | Dati JSON | Quando si utilizza il `JSON` formato `SOURCE` or, è possibile utilizzare Powertools per aiutare AWS a deserializzare i dati nel codice funzione. Ecco alcuni esempi di come gestire diversi formati di dati: ------ #### [ AVRO ] Esempio in Java: ``` package org.demo.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.demo.kafka.avro.AvroProduct; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.RequestHandler; import software.amazon.lambda.powertools.kafka.Deserialization; import software.amazon.lambda.powertools.kafka.DeserializationType; import software.amazon.lambda.powertools.logging.Logging; public class AvroDeserializationFunction implements RequestHandler, String> { private static final Logger LOGGER = LoggerFactory.getLogger(AvroDeserializationFunction.class); @Override @Logging @Deserialization(type = DeserializationType.KAFKA_AVRO) public String handleRequest(ConsumerRecords records, Context context) { for (ConsumerRecord consumerRecord : records) { LOGGER.info("ConsumerRecord: {}", consumerRecord); AvroProduct product = consumerRecord.value(); LOGGER.info("AvroProduct: {}", product); String key = consumerRecord.key(); LOGGER.info("Key: {}", key); } return "OK"; } } ``` Esempio di Python: ``` from aws_lambda_powertools.utilities.kafka_consumer.kafka_consumer import kafka_consumer from aws_lambda_powertools.utilities.kafka_consumer.schema_config import SchemaConfig from aws_lambda_powertools.utilities.kafka_consumer.consumer_records import ConsumerRecords from aws_lambda_powertools.utilities.typing import LambdaContext from aws_lambda_powertools import Logger logger = Logger(service="kafkaConsumerPowertools") value_schema_str = open("customer_profile.avsc", "r").read() schema_config = SchemaConfig( value_schema_type="AVRO", value_schema=value_schema_str) @kafka_consumer(schema_config=schema_config) def lambda_handler(event: ConsumerRecords, context:LambdaContext): for record in event.records: value = record.value logger.info(f"Received value: {value}") ``` TypeScript esempio: ``` import { kafkaConsumer } from '@aws-lambda-powertools/kafka'; import type { ConsumerRecords } from '@aws-lambda-powertools/kafka/types'; import { Logger } from '@aws-lambda-powertools/logger'; import type { Context } from 'aws-lambda'; const logger = new Logger(); type Value = { id: number; name: string; price: number; }; const schema = '{ "type": "record", "name": "Product", "fields": [ { "name": "id", "type": "int" }, { "name": "name", "type": "string" }, { "name": "price", "type": "double" } ] }'; export const handler = kafkaConsumer( (event: ConsumerRecords, _context: Context) => { for (const record of event.records) { logger.info(Processing record with key: ${record.key}); logger.info(Record value: ${JSON.stringify(record.value)}); // You can add more processing logic here } }, { value: { type: 'avro', schema: schema, }, } ); ``` esempio.NET: ``` using Amazon.Lambda.Core; using AWS.Lambda.Powertools.Kafka; using AWS.Lambda.Powertools.Kafka.Avro; using AWS.Lambda.Powertools.Logging; using Com.Example; // Assembly attribute to enable the Lambda function's Kafka event to be converted into a .NET class. [assembly: LambdaSerializer(typeof(PowertoolsKafkaAvroSerializer))] namespace ProtoBufClassLibrary; public class Function { public string FunctionHandler(ConsumerRecords records, ILambdaContext context) { foreach (var record in records) { Logger.LogInformation("Processing messagem from topic: {topic}", record.Topic); Logger.LogInformation("Partition: {partition}, Offset: {offset}", record.Partition, record.Offset); Logger.LogInformation("Produced at: {timestamp}", record.Timestamp); foreach (var header in record.Headers.DecodedValues()) { Logger.LogInformation($"{header.Key}: {header.Value}"); } Logger.LogInformation("Processing order for: {fullName}", record.Value.FullName); } return "Processed " + records.Count() + " records"; } } ``` ------ #### [ PROTOBUF ] Esempio Java: ``` package org.demo.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.demo.kafka.protobuf.ProtobufProduct; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.RequestHandler; import software.amazon.lambda.powertools.kafka.Deserialization; import software.amazon.lambda.powertools.kafka.DeserializationType; import software.amazon.lambda.powertools.logging.Logging; public class ProtobufDeserializationFunction implements RequestHandler, String> { private static final Logger LOGGER = LoggerFactory.getLogger(ProtobufDeserializationFunction.class); @Override @Logging @Deserialization(type = DeserializationType.KAFKA_PROTOBUF) public String handleRequest(ConsumerRecords records, Context context) { for (ConsumerRecord consumerRecord : records) { LOGGER.info("ConsumerRecord: {}", consumerRecord); ProtobufProduct product = consumerRecord.value(); LOGGER.info("ProtobufProduct: {}", product); String key = consumerRecord.key(); LOGGER.info("Key: {}", key); } return "OK"; } } ``` Esempio di Python: ``` from aws_lambda_powertools.utilities.kafka_consumer.kafka_consumer import kafka_consumer from aws_lambda_powertools.utilities.kafka_consumer.schema_config import SchemaConfig from aws_lambda_powertools.utilities.kafka_consumer.consumer_records import ConsumerRecords from aws_lambda_powertools.utilities.typing import LambdaContext from aws_lambda_powertools import Logger from user_pb2 import User # protobuf generated class logger = Logger(service="kafkaConsumerPowertools") schema_config = SchemaConfig( value_schema_type="PROTOBUF", value_schema=User) @kafka_consumer(schema_config=schema_config) def lambda_handler(event: ConsumerRecords, context:LambdaContext): for record in event.records: value = record.value logger.info(f"Received value: {value}") ``` TypeScript esempio: ``` import { kafkaConsumer } from '@aws-lambda-powertools/kafka'; import type { ConsumerRecords } from '@aws-lambda-powertools/kafka/types'; import { Logger } from '@aws-lambda-powertools/logger'; import type { Context } from 'aws-lambda'; import { Product } from './product.generated.js'; const logger = new Logger(); type Value = { id: number; name: string; price: number; }; export const handler = kafkaConsumer( (event: ConsumerRecords, _context: Context) => { for (const record of event.records) { logger.info(Processing record with key: ${record.key}); logger.info(Record value: ${JSON.stringify(record.value)}); } }, { value: { type: 'protobuf', schema: Product, }, } ); ``` esempio.NET: ``` using Amazon.Lambda.Core; using AWS.Lambda.Powertools.Kafka; using AWS.Lambda.Powertools.Kafka.Protobuf; using AWS.Lambda.Powertools.Logging; using Com.Example; // Assembly attribute to enable the Lambda function's Kafka event to be converted into a .NET class. [assembly: LambdaSerializer(typeof(PowertoolsKafkaProtobufSerializer))] namespace ProtoBufClassLibrary; public class Function { public string FunctionHandler(ConsumerRecords records, ILambdaContext context) { foreach (var record in records) { Logger.LogInformation("Processing messagem from topic: {topic}", record.Topic); Logger.LogInformation("Partition: {partition}, Offset: {offset}", record.Partition, record.Offset); Logger.LogInformation("Produced at: {timestamp}", record.Timestamp); foreach (var header in record.Headers.DecodedValues()) { Logger.LogInformation($"{header.Key}: {header.Value}"); } Logger.LogInformation("Processing order for: {fullName}", record.Value.FullName); } return "Processed " + records.Count() + " records"; } } ``` ------ #### [ JSON ] Esempio Java: ``` package org.demo.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.RequestHandler; import software.amazon.lambda.powertools.kafka.Deserialization; import software.amazon.lambda.powertools.kafka.DeserializationType; import software.amazon.lambda.powertools.logging.Logging; public class JsonDeserializationFunction implements RequestHandler, String> { private static final Logger LOGGER = LoggerFactory.getLogger(JsonDeserializationFunction.class); @Override @Logging @Deserialization(type = DeserializationType.KAFKA_JSON) public String handleRequest(ConsumerRecords consumerRecords, Context context) { for (ConsumerRecord consumerRecord : consumerRecords) { LOGGER.info("ConsumerRecord: {}", consumerRecord); Product product = consumerRecord.value(); LOGGER.info("Product: {}", product); String key = consumerRecord.key(); LOGGER.info("Key: {}", key); } return "OK"; } } ``` Esempio di Python: ``` from aws_lambda_powertools.utilities.kafka_consumer.kafka_consumer import kafka_consumer from aws_lambda_powertools.utilities.kafka_consumer.schema_config import SchemaConfig from aws_lambda_powertools.utilities.kafka_consumer.consumer_records import ConsumerRecords from aws_lambda_powertools.utilities.typing import LambdaContext from aws_lambda_powertools import Logger logger = Logger(service="kafkaConsumerPowertools") schema_config = SchemaConfig(value_schema_type="JSON") @kafka_consumer(schema_config=schema_config) def lambda_handler(event: ConsumerRecords, context:LambdaContext): for record in event.records: value = record.value logger.info(f"Received value: {value}") ``` TypeScript esempio: ``` import { kafkaConsumer } from '@aws-lambda-powertools/kafka'; import type { ConsumerRecords } from '@aws-lambda-powertools/kafka/types'; import { Logger } from '@aws-lambda-powertools/logger'; import type { Context } from 'aws-lambda'; const logger = new Logger(); type Value = { id: number; name: string; price: number; }; export const handler = kafkaConsumer( (event: ConsumerRecords, _context: Context) => { for (const record of event.records) { logger.info(Processing record with key: ${record.key}); logger.info(Record value: ${JSON.stringify(record.value)}); // You can add more processing logic here } }, { value: { type: 'json', }, } ); ``` esempio.NET: ``` using Amazon.Lambda.Core; using AWS.Lambda.Powertools.Kafka; using AWS.Lambda.Powertools.Kafka.Json; using AWS.Lambda.Powertools.Logging; using Com.Example; // Assembly attribute to enable the Lambda function's Kafka event to be converted into a .NET class. [assembly: LambdaSerializer(typeof(PowertoolsKafkaJsonSerializer))] namespace JsonClassLibrary; public class Function { public string FunctionHandler(ConsumerRecords records, ILambdaContext context) { foreach (var record in records) { Logger.LogInformation("Processing messagem from topic: {topic}", record.Topic); Logger.LogInformation("Partition: {partition}, Offset: {offset}", record.Partition, record.Offset); Logger.LogInformation("Produced at: {timestamp}", record.Timestamp); foreach (var header in record.Headers.DecodedValues()) { Logger.LogInformation($"{header.Key}: {header.Value}"); } Logger.LogInformation("Processing order for: {fullName}", record.Value.FullName); } return "Processed " + records.Count() + " records"; } } ``` ------ ## Metodi di autenticazione per il registro degli schemi Per utilizzare un registro di schemi, Lambda deve essere in grado di accedervi in modo sicuro. Se stai lavorando con un registro di AWS Glue schemi, Lambda si basa sull'autenticazione IAM. Ciò significa che il ruolo di [ esecuzione della funzione ](lambda-intro-execution-role.md) deve disporre delle seguenti autorizzazioni per accedere al registro: AWS Glue + [GetRegistry](https://docs.aws.amazon.com/glue/latest/webapi/API_GetRegistry.html)nel *AWS Glue Web API Reference * + [GetSchemaVersion](https://docs.aws.amazon.com/glue/latest/webapi/API_GetSchemaVersion.html)nel *AWS Glue Web API Reference * Esempio della policy IAM richiesta: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "glue:GetRegistry", "glue:GetSchemaVersion" ], "Resource": [ "*" ] } ] } ``` ------ **Nota** Per i registri di AWS Glue schema, se si fornisce un AWS Glue registro, Lambda restituisce un'`AccessConfigs`eccezione di convalida. Se utilizzi un registro di schemi Confluent, puoi scegliere uno dei tre metodi di autenticazione supportati per il parametro del tuo oggetto: `Type` [ KafkaSchemaRegistryAccessConfig ](https://docs.aws.amazon.com/lambda/latest/api/API_KafkaSchemaRegistryAccessConfig) + **BASIC\_AUTH**: Lambda utilizza nome utente e password o API Key e autenticazione API Secret per accedere al registro. Se scegli questa opzione, fornisci l'ARN di Secrets Manager contenente le tue credenziali nel campo URI. + **CLIENT\_CERTIFICATE\_TLS\_AUTH — Lambda utilizza l'autenticazione TLS reciproca ** con i certificati client. Per utilizzare questa opzione, Lambda deve accedere sia al certificato che alla chiave privata. Fornisci l'ARN di Secrets Manager contenente queste credenziali nel campo URI. + **NO\_AUTH**: il certificato emesso da una CA pubblica deve essere firmato da un'autorità di certificazione (CA) presente nel trust store di Lambda. Per un CA/self-signed certificato privato, si configura il certificato CA principale del server. Per utilizzare questa opzione, ometti il `AccessConfigs` parametro. Inoltre, se Lambda ha bisogno di accedere a un certificato CA privato per verificare il certificato TLS del registro degli schemi, scegli `SERVER_ROOT_CA_CERT` come valore `Type` e fornisci l'ARN di Secrets Manager al certificato nel campo URI. **Nota** Per configurare l'`SERVER_ROOT_CA_CERT`opzione nella console, fornisci l'ARN segreto contenente il certificato nel campo Encryption. ** ** La configurazione di autenticazione per il registro degli schemi è separata da qualsiasi autenticazione configurata per il cluster Kafka. È necessario configurarli entrambi separatamente, anche se utilizzano metodi di autenticazione simili. ## Gestione degli errori e risoluzione dei problemi relativi al registro dello schema Quando si utilizza un registro di schemi con la sorgente di eventi Amazon MSK, è possibile che si verifichino diversi errori. Questa sezione fornisce indicazioni sui problemi più comuni e su come risolverli. ### Errori di configurazione Questi errori si verificano durante l'impostazione della configurazione del registro dello schema. È richiesta la modalità con provisioning **Messaggio di errore**: `SchemaRegistryConfig is only available for Provisioned Mode. To configure Schema Registry, please enable Provisioned Mode by specifying MinimumPollers in ProvisionedPollerConfig.` **Risoluzione: ** abilita la modalità predisposta per la mappatura della sorgente degli eventi configurando il parametro in. `MinimumPollers` `ProvisionedPollerConfig` URL del registro dello schema non valido **Messaggio di errore**: `Malformed SchemaRegistryURI provided. Please provide a valid URI or ARN. For example, https://schema-registry.example.com:8081 or arn:aws:glue:us-east-1:123456789012:registry/ExampleRegistry.` **Risoluzione: ** fornire un URL HTTPS valido per Confluent Schema Registry o un ARN valido per Schema Registry. AWS Glue Formato di registrazione degli eventi non valido o mancante **Messaggio di errore**: `EventRecordFormat is a required field for SchemaRegistryConfig. Please provide one of supported format types: SOURCE, JSON.` **Risoluzione: ** specificate SOURCE o JSON come valore EventRecordFormat nella configurazione del registro dello schema. Attributi di convalida duplicati **Messaggio di errore**: `Duplicate KEY/VALUE Attribute in SchemaValidationConfigs. SchemaValidationConfigs must contain at most one KEY/VALUE Attribute.` **Risoluzione: ** rimuovi gli attributi KEY o VALUE duplicati dal tuo. SchemaValidationConfigs Ogni tipo di attributo può apparire una sola volta. Configurazione di convalida mancante **Messaggio di errore**: `SchemaValidationConfigs is a required field for SchemaRegistryConfig.` **Risoluzione: ** aggiungi SchemaValidationConfigs alla tua configurazione, specificando almeno un attributo di convalida (KEY o VALUE). ### Errori di accesso e autorizzazione Questi errori si verificano quando Lambda non può accedere al registro dello schema a causa di problemi di autorizzazione o autenticazione. AWS Glue Accesso allo Schema Registry negato **Messaggio di errore**: `Cannot access Glue Schema with provided role. Please ensure the provided role can perform the GetRegistry and GetSchemaVersion Actions on your schema.` **Risoluzione:** aggiungi le autorizzazioni necessarie (`glue:GetRegistry` e `glue:GetSchemaVersion`) per il ruolo di esecuzione di una funzione. Accesso al Confluent Schema Registry negato **Messaggio di errore**: `Cannot access Confluent Schema with the provided access configuration.` **Risoluzione: ** verifica che le tue credenziali di autenticazione (archiviate in Secrets Manager) siano corrette e dispongano delle autorizzazioni necessarie per accedere al registro degli schemi. Cross-account AWS Glue Registro degli schemi **Messaggio di errore**: `Cross-account Glue Schema Registry ARN not supported.` **Risoluzione: ** utilizza un registro degli AWS Glue schemi che si trova nello stesso AWS account della funzione Lambda. Cross-region AWS Glue Registro degli schemi **Messaggio di errore**: `Cross-region Glue Schema Registry ARN not supported.` **Risoluzione: ** utilizza un registro degli AWS Glue schemi che si trova nella stessa regione della funzione Lambda. Problemi di accesso segreto **Messaggio di errore**: `Lambda received InvalidRequestException from Secrets Manager.` **Risoluzione: ** verifica che il ruolo di esecuzione della tua funzione disponga dell'autorizzazione per accedere al segreto e che il segreto non sia crittografato con una AWS KMS chiave predefinita se accedi da un account diverso. ### Errori di connessione Questi errori si verificano quando Lambda non è in grado di stabilire una connessione al registro degli schemi. Problemi di connettività VPC **Messaggio di errore**: `Cannot connect to your Schema Registry. Your Kafka cluster's VPC must be able to connect to the schema registry. You can provide access by configuring AWS PrivateLink or a NAT Gateway or VPC Peering between Kafka Cluster VPC and the schema registry VPC.` **Risoluzione: ** configura la tua rete VPC per consentire le connessioni al registro dello schema utilizzando AWS PrivateLink un gateway NAT o un peering VPC. Errore di handshake TLS **Messaggio di errore**: `Unable to establish TLS handshake with the schema registry. Please provide correct CA-certificate or client certificate using Secrets Manager to access your schema registry.` **Risoluzione: ** verifica che i certificati CA e i certificati client (per MTL) siano corretti e configurati correttamente in Secrets Manager. Throttling **Messaggio di errore**: `Receiving throttling errors when accessing the schema registry. Please increase API TPS limits for your schema registry.` **Risoluzione: ** aumenta i limiti di velocità delle API per il registro degli schemi o riduci il tasso di richieste dall'applicazione. Self-managed errori del registro dello schema **Messaggio di errore**: `Lambda received an internal server an unexpected error from the provided self-managed schema registry.` **Risoluzione: ** controlla lo stato e la configurazione del tuo server di registro degli schemi autogestito.