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à.
# Comprendere MSK Connect
MSK Connect è una funzionalità di Amazon MSK che semplifica lo streaming di dati da e verso i cluster Apache Kafka a vantaggio degli sviluppatori. MSK Connect utilizza le versioni 2.7.1 o 3.7.x di Kafka Connect, che sono framework open source per connettere i cluster Apache Kafka con sistemi esterni come database, indici di ricerca e file system. Con MSK Connect, è possibile implementare connettori completamente gestiti creati per Kafka Connect che trasferiscono o estraggono dati da archivi di dati popolari come Amazon S3 e Amazon Service. OpenSearch È possibile implementare connettori sviluppati da terze parti come Debezium per eseguire lo streaming dei log delle modifiche dai database a un cluster Apache Kafka, oppure implementare un connettore esistente senza modifiche al codice. I connettori si dimensionano automaticamente in base alle variazioni di carico e ti verranno addebitate soltanto le risorse che effettivamente utilizzi.
Utilizza i connettori di origine per importare dati da sistemi esterni nei tuoi argomenti. Con i connettori sink, è possibile esportare i dati dai propri argomenti a sistemi esterni.
MSK Connect supporta connettori per qualsiasi cluster Apache Kafka con connettività a un Amazon VPC, che si tratti di un cluster MSK o di un cluster Apache Kafka ospitato in modo indipendente.
MSK Connect monitora continuamente l'integrità e lo stato di consegna dei connettori, corregge e gestisce l'hardware sottostante e dimensiona automaticamente i connettori in base alle variazioni della velocità di trasmissione effettiva.
Per le nozioni di base su MSK Connect, consulta la pagina [Guida introduttiva a MSK Connect](msk-connect-getting-started.md).
Per ulteriori informazioni sulle AWS risorse che è possibile creare con MSK Connect, vedere [Comprendi i connettori](msk-connect-connectors.md)[Crea plugin personalizzati](msk-connect-plugins.md), e[Scopri chi lavora con MSK Connect](msk-connect-workers.md).
Per informazioni sull'API di MSK Connect, consulta la [documentazione di riferimento sull'API di Amazon MSK Connect](https://docs.aws.amazon.com/MSKC/latest/mskc/Welcome.html).
## Vantaggi dell'utilizzo di Amazon MSK Connect
Apache Kafka è una delle piattaforme di streaming open source più utilizzate per l'acquisizione e l'elaborazione di flussi di dati in tempo reale. Con Apache Kafka, puoi disaccoppiare e scalare in modo indipendente le tue applicazioni che producono e consumano dati.
Kafka Connect è un componente importante per la creazione e l'esecuzione di applicazioni di streaming con Apache Kafka. Kafka Connect offre un modo standardizzato per lo spostamento dei dati tra Kafka e sistemi esterni. Kafka Connect è altamente scalabile e può gestire grandi volumi di dati Kafka Connect fornisce un potente set di operazioni e strumenti API per configurare, implementare e monitorare i connettori che spostano i dati tra argomenti Kafka e sistemi esterni. Puoi utilizzare questi strumenti per personalizzare ed estendere le funzionalità di Kafka Connect per soddisfare le esigenze specifiche della tua applicazione di streaming.
Potresti incontrare delle difficoltà quando gestisci i cluster Apache Kafka Connect da soli o quando cerchi di migrare applicazioni open source Apache Kafka Connect verso. AWS Queste sfide includono il tempo necessario per configurare l'infrastruttura e implementare le applicazioni, gli ostacoli tecnici alla configurazione dei cluster Apache Kafka Connect autogestiti e il sovraccarico operativo amministrativo.
Per affrontare queste sfide, ti consigliamo di utilizzare Amazon Managed Streaming for Apache Kafka Connect (Amazon MSK Connect) per migrare le tue applicazioni open source Apache Kafka Connect verso. AWS Amazon MSK Connect semplifica l'utilizzo di Kafka Connect per lo streaming di dati da e verso cluster Apache Kafka e sistemi esterni, come database, indici di ricerca e file system.
Ecco alcuni dei vantaggi della migrazione ad Amazon MSK Connect:
+ **Eliminazione del sovraccarico operativo**: Amazon MSK Connect elimina il carico operativo associato all'applicazione di patch, al provisioning e al ridimensionamento dei cluster Apache Kafka Connect. Amazon MSK Connect monitora continuamente lo stato dei cluster Connect e automatizza l'applicazione di patch e aggiornamenti di versione senza causare interruzioni ai carichi di lavoro.
+ **Riavvio automatico delle attività di Connect**: Amazon MSK Connect può ripristinare automaticamente le attività non riuscite per ridurre le interruzioni della produzione. Gli errori delle attività possono essere causati da errori temporanei, come il superamento del limite di connessione TCP per Kafka e il ribilanciamento delle attività quando nuovi lavoratori si uniscono al gruppo di consumatori per i connettori sink.
+ **Scalabilità orizzontale e verticale automatica**: Amazon MSK Connect consente all'applicazione del connettore di scalare automaticamente per supportare throughput più elevati. Amazon MSK Connect gestisce la scalabilità per te. È sufficiente specificare il numero di lavoratori nel gruppo di auto scaling e le soglie di utilizzo. Puoi utilizzare l'operazione dell'`UpdateConnector`API Amazon MSK Connect per scalare verticalmente verso l'alto o verso il basso la v CPUs tra 1 e 8 v CPUs per supportare un throughput variabile.
+ **Connettività di rete privata**: Amazon MSK Connect si connette privatamente ai sistemi di origine e sink utilizzando nomi AWS PrivateLink DNS privati.
# Guida introduttiva a MSK Connect
Questo è un step-by-step tutorial che utilizza Console di gestione AWS per creare un cluster MSK e un connettore sink che invia i dati dal cluster a un bucket S3.
**Topics**
+ [Configurazione delle risorse necessarie per MSK Connect](mkc-tutorial-setup.md)
+ [Crea un plugin personalizzato](mkc-create-plugin.md)
+ [Crea la macchina client e l'argomento Apache Kafka](mkc-create-topic.md)
+ [Crea connettore](mkc-create-connector.md)
+ [Invia dati al cluster MSK](mkc-send-data.md)
# Configurazione delle risorse necessarie per MSK Connect
In questo passaggio crei le seguenti risorse necessarie per questo scenario introduttivo:
+ Un bucket Amazon S3 che funge da destinazione per la ricezione dei dati dal connettore.
+ Un cluster MSK a cui inviare i dati. Il connettore leggerà i dati da questo cluster e li invierà al bucket S3 di destinazione.
+ Una policy IAM che contiene le autorizzazioni per scrivere nel bucket S3 di destinazione.
+ Un ruolo IAM che consente al connettore di scrivere nel bucket S3 di destinazione. A questo ruolo aggiungerai la policy IAM che crei.
+ Un endpoint Amazon VPC per consentire l'invio di dati dall'Amazon VPC che include il cluster e il connettore ad Amazon S3.
**Creazione del bucket S3**
1. Accedi a Console di gestione AWS e apri la console Amazon S3 all'indirizzo. [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/)
1. Seleziona **Crea bucket**.
1. Per il nome del bucket, specifica un nome descrittivo, ad esempio `amzn-s3-demo-bucket-mkc-tutorial`.
1. Scorri verso il basso e scegli **Crea bucket**.
1. Nell'elenco dei bucket, scegli il bucket appena creato.
1. Scegliere **Create folder (Crea cartella)**.
1. Inserisci `tutorial` come nome della cartella, quindi scorri verso il basso e scegli **Crea cartella**.
**Creazione del cluster**
1. Aprire la console Amazon MSK a [https://console.aws.amazon.com/msk/casa? region=us-east-1\$1/home/.](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/)
1. Nel riquadro a sinistra, in **Cluster MSK**, scegli **Cluster**.
1. Scegli **Crea cluster**.
1. **In Metodo di creazione, scegli Creazione personalizzata**.****
1. Come nome del cluster, specifica **mkc-tutorial-cluster**.
1. In **Tipo di cluster**, scegli **Provisioned.**
1. Scegli **Next (Successivo)**.
1. In **Rete**, scegli un Amazon VPC. Seleziona quindi le zone di disponibilità e le sottoreti che desideri utilizzare. Ricorda il IDs VPC e le sottoreti Amazon che hai selezionato perché ne avrai bisogno più avanti in questo tutorial.
1. Scegli **Next (Successivo)**.
1. In **Metodi di controllo degli accessi**, assicurati che sia selezionato soltanto **Accesso non autenticato**.
1. In **Crittografia**, assicurati che sia selezionato solo **Non crittografato**.
1. Continua con la procedura guidata, quindi scegli **Crea cluster**. In questo modo si accede alla pagina Dettagli per il cluster. In quella pagina, in **Gruppi di sicurezza applicati**, trova l'ID del gruppo di sicurezza. Ricorda quell'ID perché ne avrai bisogno più avanti in questo tutorial.
**Per creare una policy IAM con autorizzazioni di scrittura nel bucket S3**
1. Aprire la console IAM all'indirizzo [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/).
1. Nel riquadro di navigazione, seleziona **Policy**.
1. Scegli **Crea policy**.
1. Nell'**editor delle politiche**, scegli **JSON**, quindi sostituisci il codice JSON nella finestra dell'editor con il seguente JSON.
Nell'esempio seguente, sostituiscilo ** con il nome del tuo bucket S3.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "AllowListBucket",
"Effect": "Allow",
"Action": [
"s3:ListBucket",
"s3:GetBucketLocation"
],
"Resource": "arn:aws:s3:::"
},
{
"Sid": "AllowObjectActions",
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:GetObject",
"s3:DeleteObject",
"s3:AbortMultipartUpload",
"s3:ListMultipartUploadParts",
"s3:ListBucketMultipartUploads"
],
"Resource": "arn:aws:s3:::/*"
}
]
}
```
------
Per istruzioni su come scrivere policy sicure, consulta. [Controllo degli accessi IAM](iam-access-control.md)
1. Scegli **Next (Successivo)**.
1. Nella pagina **Rivedi e crea**, effettua le operazioni seguenti:
1. Per **Nome della politica**, inserisci un nome descrittivo, ad esempio**mkc-tutorial-policy**.
1. In **Autorizzazioni definite in questa politica**, rivedi e and/or modifica le autorizzazioni definite nella tua politica.
1. (Facoltativo) Per facilitare l'identificazione, l'organizzazione o la ricerca della politica, scegli **Aggiungi nuovo tag per aggiungere tag** come coppie chiave-valore. Ad esempio, aggiungi un tag alla tua politica con la coppia chiave-valore e. **Environment** **Test**
Per ulteriori informazioni sull'utilizzo dei tag, consulta [Tags for AWS Identity and Access Management resources](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_tags.html) nella *IAM User Guide*.
1. Scegli **Crea policy**.
**Creazione del ruolo IAM che può scrivere nel bucket di destinazione**
1. Nel pannello di navigazione della console IAM, scegli **Ruoli**, quindi scegli **Crea ruolo**.
1. Nella pagina **Seleziona un’entità attendibile**, esegui le operazioni seguenti:
1. Per **Tipo di entità attendibile**, seleziona **Servizio AWS**.
1. Per **Service o use case**, scegli **S3**.
1. In **Caso d'uso**, scegli **S3**.
1. Scegli **Next (Successivo)**.
1. Nella pagina **Add permissions** (Aggiungi autorizzazioni), esegui le operazioni seguenti:
1. Nella casella di ricerca sotto **Politiche di autorizzazione**, inserisci il nome della politica che hai creato in precedenza per questo tutorial. Ad esempio, **mkc-tutorial-policy**. Quindi, scegli la casella a sinistra del nome della politica.
1. (Facoltativo) Impostare un [limite delle autorizzazioni](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_boundaries.html). Questa è una funzionalità avanzata disponibile per i ruoli di servizio, ma non per i ruoli collegati ai servizi. Per informazioni sull'impostazione di un limite di autorizzazioni, consulta [Creating roles and attaching policies (console)](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_job-functions_create-policies.html) nella *IAM* User Guide.
1. Scegli **Next (Successivo)**.
1. Nella pagina **Name, review, and create** (Assegna un nome, rivedi e crea), esegui le operazioni seguenti:
1. Per il **nome del ruolo**, inserisci un nome descrittivo, ad esempio. **mkc-tutorial-role**
**Importante**
Quando assegni un nome a un ruolo, tieni presente quanto segue:
I nomi dei ruoli devono essere univoci all'interno del tuo Account AWS account e non possono essere resi unici per caso.
Ad esempio, non creare ruoli denominati **PRODROLE** e **prodrole**. Quando il nome di un ruolo viene utilizzato in una policy o come parte di un ARN, il nome del ruolo fa distinzione tra maiuscole e minuscole, tuttavia quando un nome di ruolo viene visualizzato ai clienti nella console, ad esempio durante il processo di accesso, il nome del ruolo non fa distinzione tra maiuscole e minuscole.
Non è possibile modificare il nome del ruolo dopo averlo creato, in quanto altre entità possono fare riferimento al ruolo.
1. (Facoltativo) In **Descrizione**, inserisci una descrizione per il ruolo.
1. **(Facoltativo) Per modificare i casi d'uso e le autorizzazioni per il ruolo, nel **Passaggio 1: Seleziona le entità attendibili** o nel **Passaggio 2: Aggiungi le sezioni relative alle autorizzazioni**, scegli Modifica.**
1. (Facoltativo) Per facilitare l'identificazione, l'organizzazione o la ricerca del ruolo, scegli **Aggiungi nuovo tag per aggiungere tag** come coppie chiave-valore. Ad esempio, aggiungi un tag al tuo ruolo con la coppia chiave-valore e. **ProductManager** **John**
Per ulteriori informazioni sull'utilizzo dei tag, consulta [Tags for AWS Identity and Access Management resources](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_tags.html) nella *IAM User Guide*.
1. Verificare il ruolo e quindi scegliere **Create role (Crea ruolo)**.
**Autorizzazione di MSK Connect ad assumere il ruolo**
1. Nella console IAM, nel riquadro a sinistra, in **Gestione degli accessi**, scegli **Ruoli**.
1. Trova il ruolo `mkc-tutorial-role` e selezionalo.
1. Nel **Riepilogo** del ruolo, scegli la scheda **Relazioni di attendibilità**.
1. Seleziona **Modifica relazione di attendibilità**.
1. Sostituisci la policy di attendibilità esistente con il JSON seguente.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "kafkaconnect.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
1. Scegli **Update Trust Policy (Aggiorna policy di trust)**.
**Creazione di un endpoint VPC dal VPC del cluster ad Amazon S3**
1. Apri la console Amazon VPC all'indirizzo [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
1. Nel riquadro a sinistra, scegli **Endpoint**.
1. Seleziona **Crea endpoint**.
1. In **Nome del servizio**, scegli il servizio **com.amazonaws.us-east-1.s3** e il tipo di **Gateway**.
1. Scegli il VPC del cluster, quindi seleziona la casella a sinistra della tabella di routing associata alle sottoreti del cluster.
1. Seleziona **Crea endpoint**.
**Fase successiva**
[Crea un plugin personalizzato](mkc-create-plugin.md)
# Crea un plugin personalizzato
Un plug-in contiene il codice che definisce la logica del connettore. In questo passaggio crei un plug-in personalizzato con il codice per il connettore sink Amazon S3 Lenses. In un passaggio successivo, quando creerai il connettore MSK, specificherai che il relativo codice si trova in questo plug-in personalizzato. È possibile utilizzare lo stesso plug-in per creare più connettori MSK con configurazioni diverse.
**Creazione del plug-in personalizzato**
1. Scarica il connettore [S3](https://www.confluent.io/hub/confluentinc/kafka-connect-s3).
1. Carica il file ZIP in un bucket S3 al quale puoi accedere. Per istruzioni su come caricare i file in Amazon S3, consulta la pagina [Uploading objects](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html) nella Guida per l'utente di Amazon S3.
1. Apri la console Amazon MSK all'indirizzo [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Nel riquadro a sinistra, espandi **MSK Connect**, quindi scegli **Plug-in personalizzati**.
1. Scegli **Crea plug-in personalizzato**.
1. Seleziona **Sfoglia S3**.
1. Nell'elenco dei bucket, trova il bucket in cui hai caricato il file ZIP e selezionalo.
1. Nell'elenco degli oggetti nel bucket, seleziona il pulsante di opzione a sinistra del file ZIP, quindi seleziona il pulsante con l'etichetta **Scegli**.
1. Inserisci `mkc-tutorial-plugin` come nome del plug-in personalizzato, quindi scegli **Crea plug-in personalizzato**.
Potrebbero essere necessari AWS alcuni minuti per completare la creazione del plug-in personalizzato. Al termine del processo di creazione, nella parte superiore della finestra del browser viene visualizzato il seguente messaggio in un banner.
```
Custom plugin mkc-tutorial-plugin was successfully created
The custom plugin was created. You can now create a connector using this custom plugin.
```
**Fase successiva**
[Crea la macchina client e l'argomento Apache Kafka](mkc-create-topic.md)
# Crea la macchina client e l'argomento Apache Kafka
In questo passaggio viene creata un'istanza Amazon EC2 da utilizzare come istanza client Apache Kafka. Quindi usi questa istanza per creare un argomento sul cluster.
**Per creare un computer client**
1. Apri la console Amazon EC2 all'indirizzo [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/).
1. Scegliere **Launch Instances (Avvia istanze)**.
1. Inserisci un **Nome** per il computer client, ad esempio **mkc-tutorial-client**.
1. Lascia **Amazon Linux 2 AMI (HVM) - Kernel 5.10, tipo di volume SSD** selezionato per **Tipo di Amazon Machine Image (AMI)**.
1. Scegli il tipo di istanza **t2.xlarge**.
1. In **Coppia di chiavi (accesso)**, scegli **Crea una nuova coppia di chiavi**. Inserisci **mkc-tutorial-key-pair** in **Nome della coppia di chiavi**, quindi scegli **Scarica coppia di chiavi**. In alternativa, è possibile utilizzare una coppia di chiavi esistente.
1. Scegliere **Launch Instance (Avvia istanza)**.
1. Scegliere **View Instances (Vedi istanze)**. Quindi, nella colonna **Gruppi di sicurezza**, scegli il gruppo di sicurezza associato alla nuova istanza. Copia l'ID del gruppo di sicurezza e salvalo per un secondo momento.
**Autorizzazione del client appena creato all'invio di dati al cluster**
1. Apri la console Amazon VPC all'indirizzo [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
1. Nel riquadro a sinistra, in **Sicurezza**, scegli **Gruppi di sicurezza**. Nella colonna **ID del gruppo di sicurezza**, trova il gruppo di sicurezza del cluster. Hai salvato l'ID di questo gruppo di sicurezza quando hai creato il cluster in [Configurazione delle risorse necessarie per MSK Connect](mkc-tutorial-setup.md). Scegli questo gruppo di sicurezza selezionando la casella a sinistra della riga. Assicurati che nessun altro gruppo di sicurezza sia selezionato contemporaneamente.
1. Nella sezione inferiore della pagina, scegli la scheda **Regole in entrata**.
1. Sceglere **Edit inbound rules (Modifica regole in entrata)**.
1. In basso a sinistra dello schermo, scegli **Aggiungi regola**.
1. Nella nuova regola, scegliere **All traffic (Tutto il traffico)** nella colonna **Type (Tipo)** . Nel campo a destra della colonna **Origine**, inserisci l'ID del gruppo di sicurezza del computer client. Questo è l'ID del gruppo di sicurezza che hai salvato dopo aver creato il computer client.
1. Scegliere **Salva regole**. Il cluster MSK ora accetterà tutto il traffico proveniente dal client creato nella procedura precedente.
**Per creare un argomento**
1. Apri la console Amazon EC2 all'indirizzo [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/).
1. Nella tabella delle istanze, scegli `mkc-tutorial-client`.
1. Nella parte superiore dello schermo, scegli **Connetti**, quindi segui le istruzioni per connetterti all'istanza.
1. Installa Java sull'istanza client eseguendo il seguente comando:
```
sudo yum install java-1.8.0
```
1. Eseguire il seguente comando per scaricare Apache Kafka.
```
wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz
```
**Nota**
Se desideri utilizzare un sito mirror diverso da quello utilizzato in questo comando, puoi sceglierne uno diverso sul sito Web di [Apache](https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.1/kafka_2.11-1.1.1.tgz) .
1. Eseguire il comando seguente nella directory in cui è stato scaricato il file TAR nella fase precedente.
```
tar -xzf kafka_2.12-2.2.1.tgz
```
1. Passare alla directory **kafka\$12.12-2.2.1** .
1. Aprire la console Amazon MSK a [https://console.aws.amazon.com/msk/casa? region=us-east-1\$1/home/.](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/)
1. Nel riquadro a sinistra, scegli **Cluster**, quindi scegli il nome `mkc-tutorial-cluster`.
1. Scegli **Visualizza le informazioni sul client**.
1. Copia la stringa di connessione **Non crittografato**.
1. Seleziona **Fatto**.
1. Eseguite il comando seguente sull'istanza del client (`mkc-tutorial-client`), sostituendolo *bootstrapServerString* con il valore salvato quando avete visualizzato le informazioni sul client del cluster.
```
/bin/kafka-topics.sh --create --bootstrap-server bootstrapServerString --replication-factor 2 --partitions 1 --topic mkc-tutorial-topic
```
Se il comando va a buon fine, viene visualizzato il seguente messaggio: `Created topic mkc-tutorial-topic.`
**Fase successiva**
[Crea connettore](mkc-create-connector.md)
# Crea connettore
Questa procedura descrive come creare un connettore utilizzando Console di gestione AWS.
**Creazione del connettore**
1. Accedere a e aprire la console Amazon MSK da [https://console.aws.amazon.com/msk/casa? Console di gestione AWS region=us-east-1\$1/home/.](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/)
1. Nel riquadro a sinistra, espandi **MSK Connect**, quindi scegli **Connettori**.
1. Scegli **Create connector** (Crea connettore).
1. Nell'elenco dei plugin, scegli `mkc-tutorial-plugin`, quindi scegli **Avanti**.
1. Per il nome del connettore, inserisci `mkc-tutorial-connector`.
1. Nell'elenco dei cluster, scegli `mkc-tutorial-cluster`.
1. Nella sezione **Impostazioni di rete del connettore, scegli una delle seguenti opzioni** per il tipo di rete:
+ **IPv4**(impostazione predefinita): IPv4 solo per la connettività verso destinazioni oltre
+ **Dual-stack**: per la connettività alle destinazioni su entrambe IPv4 le piattaforme IPv6 (disponibile solo se alle sottoreti sono associati blocchi IPv4 IPv6 CIDR)
1. Copia la configurazione seguente e incollala nel campo di configurazione del connettore.
Assicurati di sostituire la regione con il codice del luogo in Regione AWS cui stai creando il connettore. Inoltre, sostituisci il nome del bucket Amazon S3 ** con il nome del tuo bucket nell'esempio seguente.
```
connector.class=io.confluent.connect.s3.S3SinkConnector
s3.region=us-east-1
format.class=io.confluent.connect.s3.format.json.JsonFormat
flush.size=1
schema.compatibility=NONE
tasks.max=2
topics=mkc-tutorial-topic
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
storage.class=io.confluent.connect.s3.storage.S3Storage
s3.bucket.name=
topics.dir=tutorial
```
1. In **Autorizzazioni di accesso**, scegli `mkc-tutorial-role`.
1. Scegli **Next (Successivo)**. Nella pagina **Sicurezza**, scegli di nuovo **Avanti**.
1. Nella pagina **Log**, seleziona **Avanti**.
1. **Nella pagina **Rivedi e crea**, rivedi la configurazione del connettore e scegli Crea connettore.**
**Fase successiva**
[Invia dati al cluster MSK](mkc-send-data.md)
# Invia dati al cluster MSK
In questo passaggio si inviano i dati all'argomento Apache Kafka creato in precedenza, quindi si cercano gli stessi dati nel bucket S3 di destinazione.
**Invio dei dati al cluster MSK**
1. Nella cartella `bin` dell'installazione di Apache Kafka sull'istanza client, crea un file di testo denominato `client.properties` con i seguenti contenuti.
```
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
```
1. Esegui il comando seguente per creare un produttore della console. Sostituisci *BootstrapBrokerString* con il valore ottenuto quando hai eseguito il comando precedente.
```
/bin/kafka-console-producer.sh --broker-list BootstrapBrokerString --producer.config client.properties --topic mkc-tutorial-topic
```
1. Immettere qualsiasi messaggio desiderato e premere **Enter (Invio)**. Ripetere questa fase due o tre volte. Ogni volta che si immette una riga e si preme **Enter (Invio)**, tale riga viene inviata al cluster Apache Kafka come un messaggio separato.
1. Cerca nel bucket Amazon S3 di destinazione per trovare i messaggi inviati nel passaggio precedente.
# Comprendi i connettori
Un connettore integra sistemi esterni e servizi Amazon con Apache Kafka copiando continuamente i dati in streaming da un'origine dati nel cluster Apache Kafka o dal cluster in un sink di dati. Un connettore può anche eseguire operazioni logiche leggere come la trasformazione, la conversione del formato o il filtraggio dei dati prima di consegnarli a una destinazione. I connettori di origine estraggono i dati da un'origine dati e li inviano al cluster, mentre i connettori sink estraggono i dati dal cluster e li inviano a un sink di dati.
Nel diagramma seguente viene illustrata l'architettura di un connettore. Un worker è un processo di macchina virtuale Java (JVM) che esegue la logica del connettore. Ogni worker crea una serie di attività che vengono eseguite in thread paralleli e si occupano di copiare i dati. Le attività non memorizzano lo stato e possono quindi essere avviate, interrotte o riavviate in qualsiasi momento per fornire una pipeline di dati resiliente e scalabile.
![\[Diagramma che mostra l'architettura di un cluster di connettori.\]](http://docs.aws.amazon.com/it_it/msk/latest/developerguide/images/mkc-worker-architecture.png)
# Comprendi la capacità dei connettori
La capacità totale di un connettore dipende dal numero di lavoratori del connettore e dal numero di MSK Connect Units (MCUs) per lavoratore. Ogni MCU rappresenta 1 vCPU di elaborazione e 4 GiB di memoria. La memoria MCU riguarda la memoria totale di un'istanza worker e non la memoria heap in uso.
Gli operatori di MSK Connect utilizzano gli indirizzi IP nelle sottoreti fornite dal cliente. Ogni lavoratore utilizza un indirizzo IP da una delle sottoreti fornite dal cliente. È necessario assicurarsi di disporre di un numero sufficiente di indirizzi IP disponibili nelle sottoreti fornite a una CreateConnector richiesta per tenere conto della capacità specificata, specialmente quando si scalano automaticamente i connettori in cui il numero di lavoratori può variare.
Per creare un connettore, è necessario scegliere tra una delle due modalità di capacità seguenti.
+ *Assegnato*: scegli questa modalità se conosci i requisiti di capacità del tuo connettore. Specifica due valori:
+ Il numero di worker.
+ Il numero di dipendenti per lavoratore. MCUs
+ *Dimensionamento automatico*: scegli questa modalità se i requisiti di capacità del connettore sono variabili o se non li conosci in anticipo. Quando utilizzi la modalità di scalabilità automatica, Amazon MSK Connect sostituisce la `tasks.max` proprietà del connettore con un valore proporzionale al numero di lavoratori in esecuzione nel connettore e al numero di lavoratori per lavoratore. MCUs
Devi specificare tre set di valori:
+ Il numero minimo e massimo di worker.
+ Le percentuali di incremento e riduzione per l'utilizzo della CPU, determinate dal parametro `CpuUtilization`. Quando il parametro `CpuUtilization` del connettore supera la percentuale di dimensionamento orizzontale, MSK Connect aumenta il numero di worker che utilizzano il connettore. Quando il parametro `CpuUtilization` scende al di sotto della percentuale di ridimensionamento, MSK Connect riduce il numero di worker. Il numero di worker rimane sempre compreso tra il numero minimo e massimo specificato al momento della creazione del connettore.
+ MCUs Il numero di per lavoratore.
+ (Facoltativo) *Numero massimo di attività di scalabilità automatica*: il numero massimo di attività assegnate al connettore durante le operazioni di scalabilità automatica. Questo parametro consente di impostare un limite massimo per la creazione di attività, garantendo un maggiore controllo sull'utilizzo delle risorse e sul parallelismo in relazione alle partizioni tematiche di Kafka.
Per ulteriori informazioni sui lavoratori[Scopri chi lavora con MSK Connect](msk-connect-workers.md), vedere e per ulteriori informazioni sul numero massimo di attività con scalabilità automatica, vedere. [Comprendi il numero massimo di attività con scalabilità automatica](msk-connect-max-autoscaling-task-count.md) Per ulteriori informazioni sui parametri di MSK Connect, consulta la pagina [Monitoraggio di Amazon MSK Connect](mkc-monitoring-overview.md).
# Comprendi il numero massimo di attività con scalabilità automatica
Il `maxAutoscalingTaskCount` parametro è un campo di capacità opzionale disponibile per la scalabilità automatica dei connettori in Amazon MSK Connect. Questo parametro consente di impostare un limite massimo al numero massimo di attività che possono essere create durante le operazioni di scalabilità automatica dei connettori, garantendo un maggiore controllo sull'utilizzo delle risorse e sulle prestazioni.
Quando utilizzi la modalità di capacità con scalabilità automatica, Amazon MSK Connect sostituisce automaticamente la `tasks.max` proprietà del connettore con un valore proporzionale al numero di lavoratori e per lavoratore. MCUs Il `maxAutoscalingTaskCount` parametro fornisce un'opzione configurabile aggiuntiva per limitare il numero massimo di attività create per il connettore.
Questa funzionalità è particolarmente utile quando si desidera controllare il livello di parallelismo in relazione al numero di partizioni tematiche nel cluster Kafka. Impostando questo limite, è possibile ottimizzare le prestazioni e prevenire una distribuzione inefficiente delle attività che potrebbe verificarsi quando il numero di attività calcolato automaticamente supera i requisiti del carico di lavoro.
## Requisiti di configurazione
Il `maxAutoscalingTaskCount` parametro deve soddisfare il seguente requisito:
```
maxAutoscalingTaskCount ≥ maxWorkerCount
```
Questo requisito garantisce un utilizzo efficiente delle risorse mantenendo almeno un'attività per lavoratore. Il sistema applica questo valore minimo per ottimizzare la funzionalità del connettore.
Quando si specifica`maxAutoscalingTaskCount`, il limite viene applicato immediatamente dopo la creazione del connettore e durante tutti gli eventi di ridimensionamento successivi. Man mano che il numero di lavoratori aumenta o diminuisce durante le operazioni di scalabilità automatica, il sistema continua a rispettare questo limite. Il `tasks.max` valore si adatta proporzionalmente al numero di lavoratori e MCUs per lavoratore, ma non supera mai il valore configurato. `maxAutoscalingTaskCount`
Se non si specifica questo parametro, il connettore utilizza il calcolo standard senza alcun limite: `tasks.max = workerCount × mcuCount × tasksPerMcu` (dove tasksPerMcu è 2).
## Quando usare maxAutoscalingTask Count
Prendi `maxAutoscalingTaskCount` in considerazione l'utilizzo nei seguenti scenari:
+ *Numero limitato di partizioni*: quando gli argomenti di Kafka hanno un numero fisso di partizioni inferiore al numero di operazioni calcolato automaticamente, l'impostazione di un limite impedisce la creazione di attività inattive senza lavoro da eseguire.
+ *Ottimizzazione delle prestazioni*: dopo aver individuato che un numero di attività specifico fornisce un throughput ottimale per il carico di lavoro, è possibile limitare il numero massimo di attività per mantenere prestazioni costanti.
+ *Gestione delle risorse*: quando si desidera controllare il massimo parallelismo e il consumo di risorse del connettore indipendentemente dal numero di lavoratori in esecuzione.
## Esempio
Per un connettore con la seguente configurazione:
```
minWorkerCount: 1
maxWorkerCount: 4
mcuCount: 8
maxAutoscalingTaskCount: 15
```
`maxAutoscalingTaskCount`In caso contrario, se scalato a 4 operatori, il connettore creerebbe 64 attività (4 lavoratori × 8 MCUs × 2 attività per MCU). Se `maxAutoscalingTaskCount` impostato su 15, il connettore crea solo 15 attività, il che potrebbe essere più appropriato se l'argomento di Kafka ha 15 o meno partizioni.
# Configura il tipo di rete dual-stack
Amazon MSK Connect supporta il tipo di rete dual-stack per i nuovi connettori. Con la rete dual-stack, i connettori possono connettersi a destinazioni su entrambi i sistemi. IPv4 IPv6 Tieni presente che la IPv6 connettività è disponibile solo in modalità dual-stack (IPv4 \$1 IPv6): IPv6 la sola rete non è supportata.
Per impostazione predefinita, i nuovi connettori utilizzano il tipo di rete. IPv4 Per creare un connettore con tipo di rete dual-stack, assicurati di aver soddisfatto i prerequisiti descritti nella sezione seguente. Tieni presente che, una volta creato un connettore utilizzando un tipo di rete dual-stack, non puoi modificarne il tipo di rete. Per modificare i tipi di rete, è necessario eliminare e ricreare il connettore.
Amazon MSK Connect supporta anche la connettività degli endpoint delle API di servizio su entrambi IPv6 e. IPv4 Per utilizzare la IPv6 connettività per le chiamate API, devi utilizzare gli endpoint dual-stack. Per ulteriori informazioni sugli endpoint del servizio MSK Connect, consulta Endpoint e [quote Amazon MSK Connect](https://docs.aws.amazon.com/general/latest/gr/msk-connect.html).
## Prerequisiti per l'utilizzo del tipo di rete dual-stack
Prima di configurare il tipo di rete dual-stack per i connettori, assicurati che a tutte le sottoreti fornite durante la creazione del connettore siano assegnati entrambi i blocchi CIDR. IPv6 IPv4
## Considerazioni sull'utilizzo del tipo di rete dual-stack
+ IPv6 il supporto è attualmente disponibile solo in modalità dual-stack (\$1), non come -only IPv4 IPv6 IPv6
+ I connettori con dual-stack abilitato possono connettersi su entrambi IPv4 e IPv6 ai sistemi di dati MSK e Sink o Source
+ Il tipo di rete non può essere modificato dopo la creazione del connettore: è necessario eliminare e ricreare il connettore per modificare i tipi di rete
+ Tutte le sottoreti specificate durante la creazione del connettore devono supportare il dual-stack affinché la creazione del connettore abbia successo con il tipo di rete dual-stack
+ Se si utilizzano sottoreti dual-stack ma non viene specificato alcun tipo di rete, per impostazione predefinita il connettore verrà impostato su -only per la compatibilità con le versioni precedenti IPv4
+ Per i connettori esistenti, non è possibile aggiornare il tipo di rete: è necessario eliminare e ricreare il connettore per modificare i tipi di rete
+ L'utilizzo di reti dual-stack non comporta costi aggiuntivi
# Creazione di un connettore per
Questa procedura descrive come creare un connettore utilizzando. Console di gestione AWS
**Creazione di un connettore utilizzando il Console di gestione AWS**
1. Apri la console Amazon MSK all'indirizzo [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Nel riquadro a sinistra, in **MSK Connect**, scegli **Connettori**.
1. Scegli **Create connector** (Crea connettore).
1. Per creare il connettore, puoi scegliere se utilizzare un plug-in personalizzato esistente o creare innanzitutto un nuovo plug-in personalizzato. Per informazioni sui plug-in personalizzati e su come crearli, consulta la pagina [Crea plugin personalizzati](msk-connect-plugins.md). In questa procedura, supponiamo che tu abbia un plug-in personalizzato che desideri utilizzare. Nell'elenco dei plug-in personalizzati, trova quello che desideri utilizzare, seleziona la casella alla sua sinistra, quindi scegli **Avanti**.
1. Inserisci un nome e facoltativamente una descrizione.
1. Scegli il cluster a cui desideri connetterti.
1. Nella sezione **Impostazioni di rete del connettore**, scegli una delle seguenti opzioni per il tipo di rete:
+ **IPv4**(impostazione predefinita): IPv4 solo per la connettività verso destinazioni oltre
+ **Dual-stack**: per la connettività alle destinazioni su entrambe IPv4 le piattaforme IPv6 (disponibile solo se alle sottoreti sono associati blocchi IPv4 IPv6 CIDR)
1. Specifica la configurazione del connettore. I parametri di configurazione da specificare dipendono dal tipo di connettore che si desidera creare. Tuttavia, alcuni parametri sono comuni a tutti i connettori, ad esempio i parametri `connector.class` e `tasks.max`. Di seguito è riportato un esempio di configurazione per il [connettore sink Amazon S3 Confluent](https://www.confluent.io/hub/confluentinc/kafka-connect-s3).
```
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=2
topics=my-example-topic
s3.region=us-east-1
s3.bucket.name=amzn-s3-demo-bucket
flush.size=1
storage.class=io.confluent.connect.s3.storage.S3Storage
format.class=io.confluent.connect.s3.format.json.JsonFormat
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
schema.compatibility=NONE
```
1. Successivamente, configura la capacità del connettore. È possibile scegliere tra due modalità di capacità: assegnata e con dimensionamento automatico. Per informazioni su queste due opzioni, consulta [Comprendi la capacità dei connettori](msk-connect-capacity.md).
1. (Facoltativo) Nella sezione **Numero massimo di attività di scalabilità automatica, utilizzate il campo Numero massimo di attività** di scalabilità automatica per inserire il numero massimo di attività da allocare al connettore durante le operazioni di scalabilità automatica. Il valore deve essere almeno uguale al numero massimo di lavoratori. Se non si specifica un valore, il connettore utilizza il calcolo standard senza limiti. Per ulteriori informazioni, consulta [Comprendi il numero massimo di attività con scalabilità automatica](msk-connect-max-autoscaling-task-count.md).
1. Scegli la configurazione del worker predefinita o una configurazione del worker personalizzata. Per informazioni sulla creazione di configurazioni del worker personalizzate, consulta la pagina [Scopri chi lavora con MSK Connect](msk-connect-workers.md).
1. Successivamente, specifica il ruolo di esecuzione del servizio. Questo deve essere un ruolo IAM che MSK Connect può assumere e che concede al connettore tutte le autorizzazioni necessarie per accedere alle risorse necessarie. AWS Tali autorizzazioni dipendono dalla logica del connettore. Per informazioni su come creare questo ruolo, consulta [Comprendi il ruolo di esecuzione del servizio](msk-connect-service-execution-role.md).
1. Scegli **Avanti**, esamina le informazioni di sicurezza, quindi scegli nuovamente **Avanti**.
1. Specifica le opzioni di registrazione desiderate, quindi scegli **Avanti**. Per ulteriori informazioni sulla registrazione, consulta [Registrazione per MSK Connect](msk-connect-logging.md).
1. **Nella pagina **Rivedi e crea**, rivedi la configurazione del connettore e scegli Crea connettore.**
Per utilizzare l'API MSK Connect per creare un connettore, vedere [CreateConnector](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateConnector.html).
È possibile utilizzare l'`UpdateConnector`API per modificare la configurazione del connettore. Per ulteriori informazioni, consulta [Aggiorna un connettore](mkc-update-connector.md).
# Aggiorna un connettore
Questa procedura descrive come aggiornare la configurazione di un connettore MSK Connect esistente utilizzando. Console di gestione AWS
**Aggiornamento della configurazione del connettore utilizzando il Console di gestione AWS**
1. Apri la console Amazon MSK all'indirizzo [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Nel riquadro a sinistra, in **MSK Connect**, scegli **Connettori**.
1. Seleziona un connettore esistente.
1. Scegli **Modifica configurazione del connettore**.
1. Aggiorna la configurazione del connettore. Non è possibile sovrascrivere `connector.class` l'utilizzo UpdateConnector di. L'esempio seguente mostra una configurazione di esempio per il connettore Confluent Amazon S3 Sink.
```
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=2
topics=my-example-topic
s3.region=us-east-1
s3.bucket.name=amzn-s3-demo-bucket
flush.size=1
storage.class=io.confluent.connect.s3.storage.S3Storage
format.class=io.confluent.connect.s3.format.json.JsonFormat
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
schema.compatibility=NONE
```
1. Seleziona **Invia**.
1. È quindi possibile monitorare lo stato corrente dell'operazione nella scheda **Operazioni** del connettore.
Per utilizzare l'API MSK Connect per aggiornare la configurazione di un connettore, vedere [UpdateConnector](https://docs.aws.amazon.com/MSKC/latest/mskc/API_UpdateConnector.html).
# Connessione dai connettori
Le seguenti best practice possono migliorare le prestazioni della connettività ad Amazon MSK Connect.
## Non si sovrappongono IPs per Amazon VPC, peering o Transit Gateway
Se utilizzi il peering Amazon VPC o Transit Gateway con Amazon MSK Connect, non configurare il connettore per raggiungere le risorse VPC peerizzate con gli intervalli CIDR: IPs
+ "10.99.0.0/16"
+ "192.168.0.0/16"
+ "172.21.0.0/16"
# Crea plugin personalizzati
Un plugin è una AWS risorsa che contiene il codice che definisce la logica del connettore. Quando si crea il plug-in, si carica un file JAR (o un file ZIP che contiene uno o più file JAR) in un bucket S3 e si specifica la posizione del bucket. Quando il plugin viene creato, MSK Connect copia il contenuto dell'oggetto S3 in quel momento. Non mantiene un collegamento all'oggetto S3, quindi eventuali modifiche successive all'oggetto non influiranno sul plugin o sui suoi connettori. Quando si crea un connettore, si specifica il plug-in che si desidera che MSK Connect utilizzi. La relazione tra i plugin e i connettori è one-to-many: puoi creare uno o più connettori dallo stesso plugin.
**Nota**
I plugin personalizzati non possono essere aggiornati sul posto. Per utilizzare una nuova versione del codice del plug-in, eliminate tutti i connettori che fanno riferimento al plug-in, eliminate il plug-in e quindi ricreatelo.
**Pacchettizzazione delle dipendenze per plugin personalizzati**
Ti consigliamo di includere tutti i file JAR e le dipendenze necessari per il tuo plugin. Package del connettore in uno dei seguenti formati:
Un file ZIP che contiene tutti i file JAR e le dipendenze necessari per il plugin.
Un singolo uber JAR che contiene tutti i file di classe per il plugin e le sue dipendenze.
Il mancato raggruppamento delle dipendenze del plug-in può influire sulla disponibilità o sulla compatibilità nell'ambiente di runtime e causare errori imprevisti.
Per informazioni su come sviluppare il codice per un connettore, consulta la pagina [Connector Development Guide](https://kafka.apache.org/documentation/#connect_development) nella documentazione di Apache Kafka.
**Creazione di un plug-in personalizzato utilizzando il Console di gestione AWS**
1. Apri la console Amazon MSK all'indirizzo [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Nel riquadro a sinistra, in **MSK Connect**, scegli **Plug-in personalizzati**.
1. Scegli **Crea plug-in personalizzato**.
1. Seleziona **Sfoglia S3**.
1. Nell'elenco dei bucket S3, scegli il bucket contenente il file JAR o ZIP per il plug-in.
1. Nell'elenco degli oggetti, seleziona la casella a sinistra del file JAR o ZIP per il plug-in, quindi seleziona **Scegli**.
1. Scegli **Crea plug-in personalizzato**.
Per utilizzare l'API MSK Connect per creare un plug-in personalizzato, vedere [CreateCustomPlugin](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateCustomPlugin.html).
# Scopri chi lavora con MSK Connect
Un worker è un processo di macchina virtuale Java (JVM) che esegue la logica del connettore. Ogni worker crea una serie di attività che vengono eseguite in thread paralleli e si occupano di copiare i dati. Le attività non memorizzano lo stato e possono quindi essere avviate, interrotte o riavviate in qualsiasi momento per fornire una pipeline di dati resiliente e scalabile. Le modifiche al numero di worker, dovute a un evento di dimensionamento o a guasti imprevisti, vengono rilevate automaticamente dai worker rimanenti. Essi si coordinano per riequilibrare le attività tra il gruppo di worker rimanenti. I worker di Connect utilizzano i gruppi di consumatori di Apache Kafka per tali operazioni di coordinamento e riequilibrio.
Se i requisiti di capacità del connettore sono variabili o difficili da stimare, è possibile consentire a MSK Connect di dimensionare il numero di worker in base alle esigenze entro un limite inferiore e un limite superiore specificati. In alternativa, è possibile specificare il numero esatto di worker da utilizzare per l'esecuzione della logica di connessione. Per ulteriori informazioni, consulta [Comprendi la capacità dei connettori](msk-connect-capacity.md).
**Gli operatori di MSK Connect utilizzano gli indirizzi IP**
Gli operatori di MSK Connect utilizzano gli indirizzi IP nelle sottoreti fornite dal cliente. Ogni lavoratore utilizza un indirizzo IP da una delle sottoreti fornite dal cliente. È necessario assicurarsi di disporre di un numero sufficiente di indirizzi IP disponibili nelle sottoreti fornite a una CreateConnector richiesta per tenere conto della capacità specificata, specialmente quando si scalano automaticamente i connettori in cui il numero di lavoratori può variare.
## Configurazione dei worker predefinita
MSK Connect fornisce la seguente configurazione predefinita per i worker:
```
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
```
# Proprietà di configurazione dei worker supportate
MSK Connect fornisce una configurazione predefinita per i worker. Se lo desideri, puoi creare una configurazione dei worker personalizzata da utilizzare con i connettori. L'elenco seguente include informazioni sulle proprietà di configurazione dei worker supportate o meno da Amazon MSK Connect.
+ Sono obbligatorie solo le proprietà `key.converter` e `value.converter`.
+ MSK Connect supporta le seguenti proprietà di configurazione di `producer.`.
```
producer.acks
producer.batch.size
producer.buffer.memory
producer.compression.type
producer.enable.idempotence
producer.key.serializer
producer.linger.ms
producer.max.request.size
producer.metadata.max.age.ms
producer.metadata.max.idle.ms
producer.partitioner.class
producer.reconnect.backoff.max.ms
producer.reconnect.backoff.ms
producer.request.timeout.ms
producer.retry.backoff.ms
producer.value.serializer
```
+ MSK Connect supporta le seguenti proprietà di configurazione di `consumer.`.
```
consumer.allow.auto.create.topics
consumer.auto.offset.reset
consumer.check.crcs
consumer.fetch.max.bytes
consumer.fetch.max.wait.ms
consumer.fetch.min.bytes
consumer.heartbeat.interval.ms
consumer.key.deserializer
consumer.max.partition.fetch.bytes
consumer.max.poll.interval.ms
consumer.max.poll.records
consumer.metadata.max.age.ms
consumer.partition.assignment.strategy
consumer.reconnect.backoff.max.ms
consumer.reconnect.backoff.ms
consumer.request.timeout.ms
consumer.retry.backoff.ms
consumer.session.timeout.ms
consumer.value.deserializer
```
+ Sono supportate tutte le altre proprietà di configurazione che non iniziano con i prefissi `producer.` o `consumer.`, *ad eccezione* delle seguenti proprietà.
```
access.control.
admin.
admin.listeners.https.
client.
connect.
inter.worker.
internal.
listeners.https.
metrics.
metrics.context.
rest.
sasl.
security.
socket.
ssl.
topic.tracking.
worker.
bootstrap.servers
config.storage.topic
connections.max.idle.ms
connector.client.config.override.policy
group.id
listeners
metric.reporters
plugin.path
receive.buffer.bytes
response.http.headers.config
scheduled.rebalance.max.delay.ms
send.buffer.bytes
status.storage.topic
```
Per ulteriori informazioni sulle proprietà di configurazione dei worker e su cosa rappresentano, consulta la pagina [Kafka Connect Configs](https://kafka.apache.org/documentation/#connectconfigs) nella documentazione di Apache Kafka.
# Crea una configurazione di lavoro personalizzata
Questa procedura descrive come creare una configurazione di worker personalizzata utilizzando Console di gestione AWS.
**Creazione di una configurazione di lavoro personalizzata utilizzando il Console di gestione AWS**
1. Apri la console Amazon MSK all'indirizzo [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Nel riquadro a sinistra, in **MSK Connect**, scegli **Configurazioni del worker**.
1. Seleziona **Configurazione del worker**.
1. Inserisci un nome e una descrizione opzionale, quindi aggiungi le proprietà e i valori su cui desideri impostarli.
1. Seleziona **Configurazione del worker**.
Per utilizzare l'API MSK Connect per creare una configurazione del lavoratore, vedere [CreateWorkerConfiguration](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateWorkerConfiguration.html).
# Gestisci gli offset dei connettori di origine utilizzando `offset.storage.topic`
Questa sezione fornisce informazioni per aiutarti a gestire gli offset dei connettori di origine tramite l'*argomento di archiviazione degli offset*. L'argomento di archiviazione degli offset è un argomento interno che Kafka Connect utilizza per archiviare gli offset di configurazione dei connettori e delle attività.
## Considerazioni
Durante la gestione degli offset del connettore di origine, tieni in considerazione i seguenti aspetti.
+ Per specificare un argomento archiviazione degli offset, fornisci il nome dell'argomento Kafka in cui gli offset dei connettori vengono archiviati come valore di `offset.storage.topic` nella configurazione del worker.
+ Presta attenzione quando apporti modifiche alla configurazione di un connettore. Se un connettore di origine utilizza i valori della configurazione per record di offset chiave, la modifica dei valori di configurazione può causare un comportamento indesiderato del connettore. Ti consigliamo di fare riferimento alla documentazione del tuo plug-in per informazioni.
+ **Personalizza il numero predefinito di partizioni**: oltre a personalizzare la configurazione del worker mediante l'aggiunta di `offset.storage.topic`, è possibile personalizzare il numero di partizioni per gli argomenti di archiviazione degli offset e degli stati. Le partizioni predefinite per gli argomenti interni sono le seguenti.
+ `config.storage.topic`: 1, non configurabile, deve essere un argomento a partizione singola
+ `offset.storage.topic`: 25, configurabile fornendo `offset.storage.partitions`
+ `status.storage.topic`: 5, configurabile fornendo `status.storage.partitions`
+ **Eliminazione manuale degli argomenti**: Amazon MSK Connect crea nuovi argomenti interni di Kafka Connect (il nome dell'argomento inizia con `__amazon_msk_connect`) a ogni implementazione di connettori. I vecchi argomenti associati ai connettori eliminati non vengono rimossi automaticamente perché gli argomenti interni, ad esempio `offset.storage.topic`, possono essere riutilizzati tra i connettori. Tuttavia, è possibile eliminare manualmente gli argomenti interni non utilizzati creati da MSK Connect. Gli argomenti interni sono denominati secondo il formato `__amazon_msk_connect__connector_name_connector_id`.
Per eliminare gli argomenti interni, è possibile utilizzare l'espressione regolare `__amazon_msk_connect__connector_name_connector_id`. Evita di eliminare un argomento interno attualmente utilizzato da un connettore in esecuzione.
+ **Utilizzo dello stesso nome per gli argomenti interni creati d MSK Connect**: se desideri riutilizzare l'argomento di archiviazione degli offset per utilizzare gli offset di un connettore creato in precedenza, dovrai assegnare al nuovo connettore lo stesso nome di quello precedente. Nella configurazione del worker, è possibile impostare la proprietà `offset.storage.topic` per assegnare lo stesso nome a `offset.storage.topic` e riutilizzarlo tra connettori diversi. Questa configurazione è descritta nella sezione [Gestione degli offset dei connettori](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-create-custom-worker-config). MSK Connect non consente a connettori diversi di condividere `config.storage.topic` e `status.storage.topic`. Questi argomenti vengono creati ogni volta che si crea un nuovo connettore in MSKC. Vengono denominati automaticamente secondo il formato `__amazon_msk_connect__connector_name_connector_id` e quindi sono diversi per ciascuno dei connettori creati.
# Utilizza l'argomento predefinito relativo alla memorizzazione degli offset
Per impostazione predefinita, Amazon MSK Connect genera un nuovo argomento di archiviazione degli offset sul cluster Kafka per ogni connettore creato. MSK costruisce il nome dell'argomento predefinito utilizzando parti dell'ARN del connettore. Ad esempio, `__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2`.
# Usa l'argomento di archiviazione offset personalizzato
Per garantire la continuità degli offset tra i connettori di origine, puoi utilizzare un argomento di archiviazione degli offset a tua scelta anziché l'argomento predefinito. La definizione di un argomento di archiviazione degli offset consente di eseguire attività come la creazione di un connettore di origine che riprenda la lettura dall'ultimo offset di un connettore precedente.
Per definire un argomento di archiviazione degli offset, è necessario fornire un valore per la proprietà `offset.storage.topic` nella configurazione del worker prima di creare un connettore. Se si desidera riutilizzare l'argomento di archiviazione degli offset per utilizzare gli offset di un connettore creato in precedenza, è necessario assegnare al nuovo connettore lo stesso nome di quello precedente. Se si crea un argomento di archiviazione degli offset personalizzato, è necessario impostare [https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy](https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy) su `compact` nella configurazione dell'argomento.
**Nota**
Se si specifica un argomento di archiviazione degli offset quando si crea un connettore *sink*, MSK Connect crea l'argomento, se non esiste ancora. Tuttavia, l'argomento non verrà utilizzato per archiviare gli offset dei connettori.
Gli offset dei connettori sink vengono invece gestiti utilizzando il protocollo del gruppo di consumatori di Kafka. Ogni connettore sink crea un gruppo denominato `connect-{CONNECTOR_NAME}`. Finché esiste il gruppo di consumatori, tutti i connettori sink creati successivamente con lo stesso valore di `CONNECTOR_NAME` continueranno dall'ultimo offset confermato.
**Importante**
Se desideri aggiornare una configurazione di connettore esistente mantenendo la continuità dell'offset, usa l' UpdateConnector API. Per ulteriori informazioni, consulta [Aggiorna un connettore](mkc-update-connector.md).
**Example : specificazione di un argomento relativo alla memorizzazione dell'offset durante la ricreazione di un connettore di origine**
Se è necessario eliminare e ricreare un connettore mantenendo la continuità dell'offset, è possibile specificare un argomento relativo alla memorizzazione dell'offset nella configurazione del worker. Ad esempio, supponete di avere un connettore CDC (Change Data Capture) e di volerlo ricreare senza perdere la vostra posizione nello stream CDC. Di seguito viene illustrato come realizzare tale operazione.
1. Sul computer client, esegui il comando seguente per trovare il nome dell'argomento archiviazione degli offset del connettore. Sostituisci `` con la stringa del broker di bootstrap del cluster. Per istruzioni su come recuperare la stringa del broker di bootstrap, consulta la pagina [Ottieni i broker bootstrap per un cluster Amazon MSK](msk-get-bootstrap-brokers.md).
```
/bin/kafka-topics.sh --list --bootstrap-server
```
L'output seguente mostra un elenco di tutti gli argomenti del cluster, inclusi gli argomenti predefiniti relativi ai connettori interni. In questo esempio, il connettore CDC esistente utilizza l'[argomento di archiviazione degli offset predefinito](msk-connect-default-offset-storage-topic.md) creato da MSK Connect. Questo è il motivo per cui l'argomento di archiviazione degli offset è chiamato `__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2`.
```
__consumer_offsets
__amazon_msk_canary
__amazon_msk_connect_configs_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
__amazon_msk_connect_status_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
my-msk-topic-1
my-msk-topic-2
```
1. Apri la console Amazon MSK all'indirizzo [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk).
1. Scegli il connettore dall'elenco **Connettori**. Copia e salva il contenuto del campo **Configurazione del connettore** in modo da poterlo modificare e utilizzare per creare il nuovo connettore.
1. Scegli **Elimina** per confermare l'eliminazione. Inserisci il nome del connettore nel campo di immissione del testo per confermare l'eliminazione.
1. Crea una configurazione di worker personalizzata con valori adatti al tuo scenario. Per istruzioni, consulta [Crea una configurazione di lavoro personalizzata](msk-connect-create-custom-worker-config.md).
Nella configurazione del worker, è necessario specificare il nome dell'argomento di archiviazione degli offset recuperato in precedenza come valore di `offset.storage.topic`, come nella configurazione seguente.
```
config.providers.secretManager.param.aws.region=eu-west-3
key.converter=
value.converter=
config.providers.secretManager.class=com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider
config.providers=secretManager
offset.storage.topic=__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
```
1.
**Importante**
Al nuovo connettore deve essere assegnato lo stesso nome del vecchio connettore.
Crea un nuovo connettore utilizzando la configurazione del worker impostata nel passaggio precedente. Per istruzioni, consulta [Creazione di un connettore per](mkc-create-connector-intro.md).
# Tutorial: esternalizzazione di informazioni sensibili utilizzando provider di configurazione
Questo esempio mostra come esternalizzare le informazioni sensibili per Amazon MSK Connect utilizzando un provider di configurazione open source. Un provider di configurazione consente di specificare variabili anziché testo non crittografato in una configurazione di connettore o di worker e i worker in esecuzione nel connettore risolvono queste variabili in fase di runtime. Ciò impedisce che le credenziali e altri segreti vengano archiviati in testo non crittografato. Il provider di configurazione nell'esempio supporta il recupero dei parametri di configurazione da AWS Secrets Manager, Amazon S3 e Systems Manager (SSM). Nel [passaggio 2](#msk-connect-config-providers), viene illustrato come configurare l'archiviazione e il recupero di informazioni sensibili per il servizio che desideri configurare.
## Considerazioni
Considera quanto segue durante l'utilizzo del provider di configurazione MSK con Amazon MSK Connect:
+ Quando utilizzi i provider di configurazione, assegna le autorizzazioni appropriate al ruolo di esecuzione del servizio IAM.
+ Definisci i provider di configurazione nelle configurazioni dei worker e la rispettiva implementazione nella configurazione del connettore.
+ Se un plug-in non definisce i valori di configurazione sensibili come segreti, tali valori possono apparire nei log dei connettori. Kafka Connect tratta i valori di configurazione non definiti allo stesso modo di qualsiasi altro valore non crittografato. Per ulteriori informazioni, consulta [Impedire la visualizzazione di segreti nei log dei connettori](msk-connect-logging.md#msk-connect-logging-secrets).
+ Per impostazione predefinita, spesso MSK Connect riavvia un connettore se questo utilizza un provider di configurazione. Per disattivare questo comportamento di riavvio, è possibile impostare il valore di `config.action.reload` su `none` nella configurazione del connettore.
## Crea un plug-in personalizzato e caricalo su S3
Per creare un plugin personalizzato, crea un file zip che contenga il connettore ed esegui i seguenti comandi sul tuo computer locale. msk-config-provider
**Creazione di un plug-in personalizzato utilizzando una finestra di terminale e Debezium come connettore**
Usa la AWS CLI per eseguire comandi come superutente con credenziali che ti consentono di accedere al tuo bucket S3. AWS *Per informazioni sull'installazione e la configurazione della AWS CLI, consulta [Guida introduttiva alla AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html) nella Guida per l'utente.AWS Command Line Interface * *Per informazioni sull'uso della AWS CLI con Amazon S3, consulta Using [Amazon S3 with the AWS CLI nella Guida per l'](https://docs.aws.amazon.com/cli/latest/userguide/cli-services-s3.html)utente.AWS Command Line Interface *
1. In una finestra del terminale, crea una cartella denominata `custom-plugin` nel tuo spazio di lavoro tramite il seguente comando.
```
mkdir custom-plugin && cd custom-plugin
```
1. Scarica l'ultima versione stabile del **plug-in per il connettore MySQL** dal [sito di Debezium](https://debezium.io/releases/) tramite il seguente comando.
```
wget https://repo1.maven.org/maven2/io/debezium/debezium-connectormysql/
2.2.0.Final/debezium-connector-mysql-2.2.0.Final-plugin.tar.gz
```
Estrai il file gzip scaricato nella cartella `custom-plugin` tramite il seguente comando.
```
tar xzf debezium-connector-mysql-2.2.0.Final-plugin.tar.gz
```
1. Scarica il [file zip del provider di configurazione MSK](https://github.com/aws-samples/msk-config-providers/releases/download/r0.4.0/msk-config-providers-0.4.0-with-dependencies.zip) tramite il seguente comando.
```
wget https://github.com/aws-samples/msk-config-providers/releases/download/r0.4.0/msk-config-providers-0.4.0-with-dependencies.zip
```
Estrai il file zip scaricato nella cartella `custom-plugin` tramite il seguente comando.
```
unzip msk-config-providers-0.4.0-with-dependencies.zip
```
1. Comprimi il contenuto del provider di configurazione MSK del passaggio precedente e del connettore personalizzato in un unico file denominato `custom-plugin.zip`.
```
zip -r ../custom-plugin.zip *
```
1. Carica il file su S3 per utilizzarlo come riferimento in seguito.
```
aws s3 cp ../custom-plugin.zip s3:
```
1. Sulla console Amazon MSK, nella sezione **MSK Connect**, scegli Custom Plugin, quindi scegli **Crea **plug-in personalizzato**** e sfoglia il bucket **s3: < *S3\$1URI\$1BUCKET\$1LOCATION* > S3** per selezionare il file ZIP del plug-in personalizzato che hai appena caricato.
![\[Amazon S3 bucket interface showing a single custom-plugin.zip file in the debezium folder.\]](http://docs.aws.amazon.com/it_it/msk/latest/developerguide/images/s3-object-browser.png)
1. Inserisci **debezium-custom-plugin** come nome del plug-in. Facoltativamente, inserisci una descrizione e scegli **Crea plug-in personalizzato**.
![\[Amazon S3 bucket interface showing a single custom-plugin.zip file in the debezium folder.\]](http://docs.aws.amazon.com/it_it/msk/latest/developerguide/images/create-custom-plugin.png)
## Configura parametri e autorizzazioni per diversi provider
È possibile configurare i valori dei parametri in questi tre servizi:
+ Secrets Manager
+ Systems Manager Parameter Store
+ S3 - Simple Storage Service
Scegli una delle schede seguenti per istruzioni sulla configurazione dei parametri e delle autorizzazioni pertinenti per quel servizio.
------
#### [ Configure in Secrets Manager ]
**Configurazione dei valori dei parametri in Secrets Manager**
1. Apri la [console Secrets Manager](https://console.aws.amazon.com/secretsmanager/).
1. Crea un nuovo segreto per archiviare le credenziali o i segreti. Per istruzioni, consulta [Creare un Gestione dei segreti AWS segreto nella Guida](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html) per l'*Gestione dei segreti AWS utente*.
1. Copia l'ARN del segreto.
1. Aggiungi le autorizzazioni di Secrets Manager dalla seguente policy di esempio al tuo [ruolo di esecuzione del servizio](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). Sostituisci l'esempio ARN,`arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234`, con l'ARN del tuo segreto.
1. Aggiungi le istruzioni per la configurazione del worker e il connettore.
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"secretsmanager:GetResourcePolicy",
"secretsmanager:GetSecretValue",
"secretsmanager:DescribeSecret",
"secretsmanager:ListSecretVersionIds"
],
"Resource": [
"arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234"
]
}
]
}
```
1. Per utilizzare il provider di configurazione Secrets Manager, copia le seguenti righe di codice nella casella di testo della configurazione del worker nel passaggio 3:
```
# define name of config provider:
config.providers = secretsmanager
# provide implementation classes for secrets manager:
config.providers.secretsmanager.class = com.amazonaws.kafka.config.providers.SecretsManagerConfigProvider
# configure a config provider (if it needs additional initialization), for example you can provide a region where the secrets or parameters are located:
config.providers.secretsmanager.param.region = us-east-1
```
1. Per il provider di configurazione Secrets Manager, copia le seguenti righe di codice nella configurazione del connettore nel passaggio 4.
```
#Example implementation for secrets manager variable
database.user=${secretsmanager:MSKAuroraDBCredentials:username}
database.password=${secretsmanager:MSKAuroraDBCredentials:password}
```
È possibile utilizzare il passaggio precedente anche con altri provider di configurazione.
------
#### [ Configure in Systems Manager Parameter Store ]
**Configurazione dei valori dei parametri in Archivio dei parametri Systems Manager**
1. Aprire la [console Systems Manager](https://console.aws.amazon.com/systems-manager/).
1. Nel riquadro di navigazione, selezionare **Parameter Store (Archivio parametri)**.
1. Crea un nuovo parametro da archiviare in Systems Manager. Per istruzioni, vedere [Create a Systems Manager (console)](https://docs.aws.amazon.com/systems-manager/latest/userguide/parameter-create-console.html) nella Guida per l' AWS Systems Manager utente.
1. Copia l'ARN del parametro.
1. Aggiungi le autorizzazioni di Systems Manager dalla seguente policy di esempio al tuo [ruolo di esecuzione del servizio](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). Sostituisci ** con l'ARN del tuo parametro.
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"ssm:GetParameterHistory",
"ssm:GetParametersByPath",
"ssm:GetParameters",
"ssm:GetParameter"
],
"Resource": "arn:aws:ssm:us-east-1:123456789000:parameter/MyParameterName"
}
]
}
```
1. Per utilizzare il provider di configurazione Archivio dei parametri, copia le seguenti righe di codice nella casella di testo della configurazione del worker nel passaggio 3:
```
# define name of config provider:
config.providers = ssm
# provide implementation classes for parameter store:
config.providers.ssm.class = com.amazonaws.kafka.config.providers.SsmParamStoreConfigProvider
# configure a config provider (if it needs additional initialization), for example you can provide a region where the secrets or parameters are located:
config.providers.ssm.param.region = us-east-1
```
1. Per il provider di configurazione Archivio dei parametri, copia le seguenti righe di codice nella configurazione del connettore nel passaggio 5.
```
#Example implementation for parameter store variable
schema.history.internal.kafka.bootstrap.servers=${ssm::MSKBootstrapServerAddress}
```
È possibile utilizzare i due passaggi precedenti anche con altri provider di configurazione.
------
#### [ Configure in Amazon S3 ]
**Per configurare objects/files in Amazon S3**
1. Apri la [console Amazon S3](https://console.aws.amazon.com/s3/).
1. Carica l'oggetto in un bucket S3. Per istruzioni, consulta la pagina [Uploading objects](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html).
1. Copia l'ARN dell'oggetto.
1. Aggiungi le autorizzazioni di Amazon S3 Object Read dalla seguente policy di esempio al tuo [ruolo di esecuzione del servizio](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). Sostituisci l'esempio ARN,`arn:aws:s3:::`, con l'ARN del tuo oggetto.
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": "s3:GetObject",
"Resource": "arn:aws:s3:::"
}
]
}
```
1. Per utilizzare il provider di configurazione Amazon S3, copia le seguenti righe di codice nella casella di testo della configurazione del worker nel passaggio 3:
```
# define name of config provider:
config.providers = s3import
# provide implementation classes for S3:
config.providers.s3import.class = com.amazonaws.kafka.config.providers.S3ImportConfigProvider
```
1. Per il provider di configurazione Amazon S3, copia le seguenti righe di codice nella configurazione del connettore nel passaggio 4.
```
#Example implementation for S3 object
database.ssl.truststore.location = ${s3import:us-west-2:my_cert_bucket/path/to/trustore_unique_filename.jks}
```
È possibile utilizzare i due passaggi precedenti anche con altri provider di configurazione.
------
## Creazione di una configurazione del worker personalizzata con informazioni sul proprio provider di configurazione
1. Seleziona **Configurazioni dei worker** nella sezione **Amazon MSK Connect**.
1. Seleziona **Crea configurazione del worker**.
1. Inserisci `SourceDebeziumCustomConfig` nella casella di testo Nome della configurazione del worker. La descrizione è facoltativa.
1. Copia il codice di configurazione pertinente in base ai provider desiderati e incollalo nella casella di testo **Configurazione del worker**.
1. Questo è un esempio di configurazione del worker per tutti e tre i provider:
```
key.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
offset.storage.topic=offsets_my_debezium_source_connector
# define names of config providers:
config.providers=secretsmanager,ssm,s3import
# provide implementation classes for each provider:
config.providers.secretsmanager.class = com.amazonaws.kafka.config.providers.SecretsManagerConfigProvider
config.providers.ssm.class = com.amazonaws.kafka.config.providers.SsmParamStoreConfigProvider
config.providers.s3import.class = com.amazonaws.kafka.config.providers.S3ImportConfigProvider
# configure a config provider (if it needs additional initialization), for example you can provide a region where the secrets or parameters are located:
config.providers.secretsmanager.param.region = us-east-1
config.providers.ssm.param.region = us-east-1
```
1. Fai clic su Crea configurazione del worker.
## Crea il connettore
1. Crea un nuovo connettore seguendo le istruzioni riportate nella sezione [Creazione di un nuovo connettore](https://docs.aws.amazon.com/msk/latest/developerguide/mkc-create-connector.html).
1. Scegli il file `custom-plugin.zip` che hai caricato nel tuo bucket S3 in [Crea un plug-in personalizzato e caricalo su S3](#msk-connect-config-providers-create-custom-plugin) come origine del plug-in personalizzato.
1. Copia il codice di configurazione pertinente in base ai provider desiderati e incollalo nel campo Configurazione del cluster.
1. Questo è un esempio della configurazione dei connettori per tutti e tre i provider:
```
#Example implementation for parameter store variable
schema.history.internal.kafka.bootstrap.servers=${ssm::MSKBootstrapServerAddress}
#Example implementation for secrets manager variable
database.user=${secretsmanager:MSKAuroraDBCredentials:username}
database.password=${secretsmanager:MSKAuroraDBCredentials:password}
#Example implementation for Amazon S3 file/object
database.ssl.truststore.location = ${s3import:us-west-2:my_cert_bucket/path/to/trustore_unique_filename.jks}
```
1. Seleziona **Usa una configurazione personalizzata** e scegli **SourceDebeziumCustomConfig**dal menu a discesa **Worker Configuration**.
1. Segui i passaggi rimanenti indicati nelle istruzioni nella sezione [Creazione di un connettore](https://docs.aws.amazon.com/msk/latest/developerguide/mkc-create-connector.html).
# Ruoli e policy IAM per MSK Connect
Questa sezione ti aiuta a configurare le politiche e i ruoli IAM appropriati per distribuire e gestire in modo sicuro Amazon MSK Connect all'interno AWS del tuo ambiente. Le sezioni seguenti spiegano il ruolo di esecuzione del servizio che deve essere utilizzato con MSK Connect, inclusa la politica di attendibilità richiesta e le autorizzazioni aggiuntive necessarie per la connessione a un cluster MSK autenticato tramite IAM. La pagina fornisce anche esempi di policy IAM complete per garantire l'accesso completo alle funzionalità di MSK Connect, oltre a dettagli sulle policy AWS gestite disponibili per il servizio.
**Topics**
+ [Comprendi il ruolo di esecuzione del servizio](msk-connect-service-execution-role.md)
+ [Esempio di policy IAM per MSK Connect](mkc-iam-policy-examples.md)
+ [Prevenire il problema della confusione tra i vari servizi](cross-service-confused-deputy-prevention.md)
+ [AWS politiche gestite per MSK Connect](mkc-security-iam-awsmanpol.md)
+ [Utilizzare ruoli collegati ai servizi per MSK Connect](mkc-using-service-linked-roles.md)
# Comprendi il ruolo di esecuzione del servizio
**Nota**
Amazon MSK Connect non supporta l'utilizzo del [ruolo collegato ai servizi](mkc-using-service-linked-roles.md) come ruolo di esecuzione del servizio. È necessario creare un ruolo di esecuzione del servizio separato. Per istruzioni su come creare un ruolo IAM personalizzato, consulta [Creazione di un ruolo per delegare le autorizzazioni a un AWS servizio](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-service.html) nella Guida per l'*utente IAM*.
Quando si crea un connettore con MSK Connect, è necessario specificare un ruolo AWS Identity and Access Management (IAM) da utilizzare con esso. Il ruolo di esecuzione del servizio deve disporre della seguente policy di attendibilità affinché MSK Connect lo possa assumere. Per ulteriori informazioni sulle chiavi di contesto delle condizioni in questa policy, consulta la pagina [Prevenire il problema della confusione tra i vari servizi](cross-service-confused-deputy-prevention.md).
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "kafkaconnect.amazonaws.com"
},
"Action": "sts:AssumeRole",
"Condition": {
"StringEquals": {
"aws:SourceAccount": "123456789012"
},
"ArnLike": {
"aws:SourceArn": "arn:aws:kafkaconnect:us-east-1:123456789012:connector/myConnector/abc12345-abcd-4444-a8b9-123456f513ed-2"
}
}
}
]
}
```
------
Se il cluster Amazon MSK che desideri utilizzare con il connettore è un cluster che utilizza l'autenticazione IAM, devi aggiungere la seguente policy di autorizzazione al ruolo di esecuzione del servizio del connettore. Per informazioni su come trovare l'UUID del cluster e su come costruire l'argomento, consulta. ARNs [Risorse relative alla politica di autorizzazione](kafka-actions.md#msk-iam-resources)
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kafka-cluster:Connect",
"kafka-cluster:DescribeCluster"
],
"Resource": [
"arn:aws:kafka:us-east-1:000000000001:cluster/testClusterName/300d0000-0000-0005-000f-00000000000b-1"
]
},
{
"Effect": "Allow",
"Action": [
"kafka-cluster:ReadData",
"kafka-cluster:DescribeTopic"
],
"Resource": [
"arn:aws:kafka:us-east-1:123456789012:topic/myCluster/300a0000-0000-0003-000a-00000000000b-6/__amazon_msk_connect_read"
]
},
{
"Effect": "Allow",
"Action": [
"kafka-cluster:WriteData",
"kafka-cluster:DescribeTopic"
],
"Resource": [
"arn:aws:kafka:us-east-1:123456789012:topic/testCluster/300f0000-0000-0008-000d-00000000000m-7/__amazon_msk_connect_write"
]
},
{
"Effect": "Allow",
"Action": [
"kafka-cluster:CreateTopic",
"kafka-cluster:WriteData",
"kafka-cluster:ReadData",
"kafka-cluster:DescribeTopic"
],
"Resource": [
"arn:aws:kafka:us-east-1:123456789012:topic/testCluster/300f0000-0000-0008-000d-00000000000m-7/__amazon_msk_connect_*"
]
},
{
"Effect": "Allow",
"Action": [
"kafka-cluster:AlterGroup",
"kafka-cluster:DescribeGroup"
],
"Resource": [
"arn:aws:kafka:us-east-1:123456789012:group/testCluster/300d0000-0000-0005-000f-00000000000b-1/__amazon_msk_connect_*",
"arn:aws:kafka:us-east-1:123456789012:group/testCluster/300d0000-0000-0005-000f-00000000000b-1/connect-*"
]
}
]
}
```
------
A seconda del tipo di connettore, potrebbe anche essere necessario allegare al ruolo di esecuzione del servizio una politica di autorizzazioni che gli consenta di accedere alle risorse. AWS Ad esempio, se il connettore deve inviare dati a un bucket S3, il ruolo di esecuzione del servizio deve disporre di una policy di autorizzazione che conceda l'autorizzazione alla scrittura su quel bucket. A scopo di test, puoi utilizzare una delle policy IAM predefinite che forniscono l'accesso completo, come `arn:aws:iam::aws:policy/AmazonS3FullAccess`. Tuttavia, per motivi di sicurezza, si consiglia di utilizzare la politica più restrittiva che consenta al connettore di leggere dalla AWS fonte o scrivere nel AWS sink.
# Esempio di policy IAM per MSK Connect
Per fornire a un utente non amministratore l'accesso completo a tutte le funzionalità di MSK Connect, collega una policy come la seguente al ruolo IAM dell'utente.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "MSKConnectFullAccess",
"Effect": "Allow",
"Action": [
"kafkaconnect:CreateConnector",
"kafkaconnect:DeleteConnector",
"kafkaconnect:DescribeConnector",
"kafkaconnect:ListConnectors",
"kafkaconnect:UpdateConnector",
"kafkaconnect:CreateCustomPlugin",
"kafkaconnect:DeleteCustomPlugin",
"kafkaconnect:DescribeCustomPlugin",
"kafkaconnect:ListCustomPlugins",
"kafkaconnect:CreateWorkerConfiguration",
"kafkaconnect:DeleteWorkerConfiguration",
"kafkaconnect:DescribeWorkerConfiguration",
"kafkaconnect:ListWorkerConfigurations"
],
"Resource": "*"
},
{
"Sid": "IAMPassRole",
"Effect": "Allow",
"Action": "iam:PassRole",
"Resource": "arn:aws:iam::123456789012:role/MSKConnectServiceRole",
"Condition": {
"StringEquals": {
"iam:PassedToService": "kafkaconnect.amazonaws.com"
}
}
},
{
"Sid": "EC2NetworkAccess",
"Effect": "Allow",
"Action": [
"ec2:CreateNetworkInterface",
"ec2:DescribeNetworkInterfaces",
"ec2:DeleteNetworkInterface",
"ec2:DescribeVpcs",
"ec2:DescribeSubnets",
"ec2:DescribeSecurityGroups"
],
"Resource": "*"
},
{
"Sid": "MSKClusterAccess",
"Effect": "Allow",
"Action": [
"kafka:DescribeCluster",
"kafka:DescribeClusterV2",
"kafka:GetBootstrapBrokers"
],
"Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/myCluster/"
},
{
"Sid": "MSKLogGroupAccess",
"Effect": "Allow",
"Action": [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents",
"logs:DescribeLogStreams",
"logs:DescribeLogGroups"
],
"Resource": [
"arn:aws:logs:us-east-1:123456789012:log-group:/aws/msk-connect/*"
]
},
{
"Sid": "S3PluginAccess",
"Effect": "Allow",
"Action": [
"s3:GetObject",
"s3:ListBucket",
"s3:PutObject"
],
"Resource": [
"arn:aws:s3:::amzn-s3-demo-bucket1-custom-plugins",
"arn:aws:s3:::amzn-s3-demo-bucket1-custom-plugins/*"
]
}
]
}
```
------
# Prevenire il problema della confusione tra i vari servizi
Il problema confused deputy è un problema di sicurezza in cui un’entità che non dispone dell’autorizzazione per eseguire un’azione può costringere un’entità maggiormente privilegiata a eseguire l’azione. Nel AWS, l'impersonificazione tra servizi può causare il problema del sostituto confuso. La rappresentazione tra servizi può verificarsi quando un servizio (il *servizio chiamante*) effettua una chiamata a un altro servizio (il *servizio chiamato*). Il servizio chiamante può essere manipolato per utilizzare le proprie autorizzazioni e agire sulle risorse di un altro cliente, a cui normalmente non avrebbe accesso. Per evitare ciò, AWS fornisce strumenti per poterti a proteggere i tuoi dati per tutti i servizi con entità di servizio a cui è stato concesso l’accesso alle risorse dell’account.
Ti consigliamo di utilizzare le chiavi di contesto delle condizioni globali [https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourcearn](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourcearn) e [https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourceaccount](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourceaccount) nelle policy delle risorse per limitare le autorizzazioni con cui MSK Connect fornisce un altro servizio alla risorsa. Se il valore `aws:SourceArn` non contiene l'ID account (ad esempio, l'AR di un bucket Amazon S3 non contiene l'ID account), è necessario utilizzare entrambe le chiavi di contesto delle condizioni globali per limitare le autorizzazioni. Se si utilizzano entrambe le chiavi di contesto delle condizioni globali e il valore `aws:SourceArn` contiene l'ID account, il valore `aws:SourceAccount` e l'account nel valore `aws:SourceArn` deve utilizzare lo stesso ID account nella stessa dichiarazione di policy. Utilizzare `aws:SourceArn` se si desidera consentire l'associazione di una sola risorsa all'accesso tra servizi. Utilizzare `aws:SourceAccount` se si desidera consentire l’associazione di qualsiasi risorsa in tale account all’uso tra servizi.
Nel caso di MSK Connect, il valore di `aws:SourceArn` deve essere un connettore MSK.
Il modo più efficace per proteggersi dal problema "confused deputy" è quello di usare la chiave di contesto della condizione globale `aws:SourceArn` con l’ARN completo della risorsa. Se non si conosce l’ARN completo della risorsa o si scelgono più risorse, è necessario utilizzare la chiave di contesto della condizione globale `aws:SourceArn` con caratteri jolly (`*`) per le parti sconosciute dell’ARN. Ad esempio, *arn:aws:kafkaconnect:us-east-1:123456789012:connector/\$1* rappresenta tutti i connettori che appartengono all'account con ID 123456789012 nella regione Stati Uniti orientali (Virginia settentrionale).
L'esempio seguente mostra il modo in cui puoi utilizzare le chiavi di contesto delle condizioni globali `aws:SourceArn` e `aws:SourceAccount` in MSK Connect per prevenire il problema confused deputy. Sostituisci *123456789012* e arn:aws:kafkaconnect: ::connector//con le tue informazioni sul connettore. *us-east-1* *123456789012* *my-S3-Sink-Connector* *abcd1234-5678-90ab-cdef-1234567890ab* Account AWS
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": " kafkaconnect.amazonaws.com"
},
"Action": "sts:AssumeRole",
"Condition": {
"StringEquals": {
"aws:SourceAccount": "123456789012"
},
"ArnLike": {
"aws:SourceArn": "arn:aws:kafkaconnect:us-east-1:123456789012:connector/my-S3-Sink-Connector/abcd1234-5678-90ab-cdef-1234567890ab"
}
}
}
]
}
```
------
# AWS politiche gestite per MSK Connect
Una politica AWS gestita è una politica autonoma creata e amministrata da. AWS AWS le politiche gestite sono progettate per fornire autorizzazioni per molti casi d'uso comuni, in modo da poter iniziare ad assegnare autorizzazioni a utenti, gruppi e ruoli.
Tieni presente che le policy AWS gestite potrebbero non concedere le autorizzazioni con il privilegio minimo per i tuoi casi d'uso specifici, poiché sono disponibili per tutti i clienti. AWS Si consiglia pertanto di ridurre ulteriormente le autorizzazioni definendo [policy gestite dal cliente](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#customer-managed-policies) specifiche per i propri casi d'uso.
Non è possibile modificare le autorizzazioni definite nelle politiche gestite. AWS Se AWS aggiorna le autorizzazioni definite in una politica AWS gestita, l'aggiornamento ha effetto su tutte le identità principali (utenti, gruppi e ruoli) a cui è associata la politica. AWS è più probabile che aggiorni una policy AWS gestita quando ne Servizio AWS viene lanciata una nuova o quando diventano disponibili nuove operazioni API per i servizi esistenti.
Per ulteriori informazioni, consultare [Policy gestite da AWS](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#aws-managed-policies) nella *Guida per l'utente di IAM*.
## AWS politica gestita: Amazon MSKConnect ReadOnlyAccess
Questa policy concede all'utente le autorizzazioni necessarie per elencare e descrivere le risorse MSK Connect.
È possibile allegare la policy `AmazonMSKConnectReadOnlyAccess` alle identità IAM.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kafkaconnect:ListConnectors",
"kafkaconnect:ListCustomPlugins",
"kafkaconnect:ListWorkerConfigurations"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"kafkaconnect:DescribeConnector"
],
"Resource": [
"arn:aws:kafkaconnect:*:*:connector/*"
]
},
{
"Effect": "Allow",
"Action": [
"kafkaconnect:DescribeCustomPlugin"
],
"Resource": [
"arn:aws:kafkaconnect:*:*:custom-plugin/*"
]
},
{
"Effect": "Allow",
"Action": [
"kafkaconnect:DescribeWorkerConfiguration"
],
"Resource": [
"arn:aws:kafkaconnect:*:*:worker-configuration/*"
]
}
]
}
```
------
## AWS politica gestita: KafkaConnectServiceRolePolicy
Questa policy concede al servizio MSK Connect le autorizzazioni necessarie per creare e gestire le interfacce di rete alle quali è assegnato il tag `AmazonMSKConnectManaged:true`. Queste interfacce di rete forniscono a MSK Connect l'accesso di rete alle risorse del tuo Amazon VPC, come un cluster Apache Kafka, un'origine o un sink.
Non puoi collegarti KafkaConnectServiceRolePolicy alle tue entità IAM. Questa policy è collegata a un ruolo collegato ai servizi che consente a MSK Connect di eseguire operazioni per tuo conto.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"ec2:CreateNetworkInterface"
],
"Resource": "arn:aws:ec2:*:*:network-interface/*",
"Condition": {
"StringEquals": {
"aws:RequestTag/AmazonMSKConnectManaged": "true"
},
"ForAllValues:StringEquals": {
"aws:TagKeys": "AmazonMSKConnectManaged"
}
}
},
{
"Effect": "Allow",
"Action": [
"ec2:CreateNetworkInterface"
],
"Resource": [
"arn:aws:ec2:*:*:subnet/*",
"arn:aws:ec2:*:*:security-group/*"
]
},
{
"Effect": "Allow",
"Action": [
"ec2:CreateTags"
],
"Resource": "arn:aws:ec2:*:*:network-interface/*",
"Condition": {
"StringEquals": {
"ec2:CreateAction": "CreateNetworkInterface"
}
}
},
{
"Effect": "Allow",
"Action": [
"ec2:DescribeNetworkInterfaces",
"ec2:CreateNetworkInterfacePermission",
"ec2:AttachNetworkInterface",
"ec2:DetachNetworkInterface",
"ec2:DeleteNetworkInterface"
],
"Resource": "arn:aws:ec2:*:*:network-interface/*",
"Condition": {
"StringEquals": {
"ec2:ResourceTag/AmazonMSKConnectManaged": "true"
}
}
}
]
}
```
------
## MSK Connect aggiorna le policy AWS gestite
Visualizza i dettagli sugli aggiornamenti delle politiche AWS gestite per MSK Connect da quando questo servizio ha iniziato a tenere traccia di queste modifiche.
| Modifica | Descrizione | Data |
| --- | --- | --- |
| Policy di sola lettura di MSK Connect aggiornata | MSK Connect ha aggiornato la MSKConnect ReadOnlyAccess politica di Amazon per rimuovere le restrizioni sulle operazioni di pubblicazione delle offerte. | 13 ottobre 2021 |
| Inizio del tracciamento delle modifiche da parte di MSK Connect | MSK Connect ha iniziato a tenere traccia delle modifiche per le sue politiche AWS gestite. | 14 settembre 2021 |
# Utilizzare ruoli collegati ai servizi per MSK Connect
Amazon MSK Connect utilizza ruoli collegati ai [servizi AWS Identity and Access Management](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_terms-and-concepts.html#iam-term-service-linked-role) (IAM). Un ruolo collegato ai servizi è un tipo di ruolo IAM univoco collegato direttamente a MSK Connect. I ruoli collegati ai servizi sono predefiniti da MSK Connect e includono tutte le autorizzazioni richieste dal servizio per chiamare altri AWS servizi per conto dell'utente.
Un ruolo collegato ai servizi semplifica la configurazione di MSK Connect perché permette di evitare l'aggiunta manuale delle autorizzazioni necessarie. MSK Connect definisce le autorizzazioni dei relativi ruoli associati ai servizi e, salvo diversamente definito, solo MSK Connect potrà assumere i propri ruoli. Le autorizzazioni definite includono la policy di attendibilità e la policy delle autorizzazioni che non può essere collegata a nessun'altra entità IAM.
Per informazioni sugli altri servizi che supportano i ruoli collegati ai servizi, consulta la sezione [Servizi AWS che funzionano con IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_aws-services-that-work-with-iam.html) e cerca i servizi che riportano **Sì** nella colonna **Ruolo associato ai servizi**. Scegli **Sì** in corrispondenza di un link per visualizzare la documentazione relativa al ruolo collegato ai servizi per tale servizio.
## Autorizzazioni del ruolo collegato ai servizi per MSK Connect
MSK Connect utilizza il ruolo collegato al servizio denominato: **AWSServiceRoleForKafkaConnect**consente ad Amazon MSK Connect di accedere alle risorse Amazon per tuo conto.
Il ruolo AWSService RoleForKafkaConnect collegato al servizio si fida che il servizio assuma il ruolo. `kafkaconnect.amazonaws.com`
Per informazioni sulla policy di autorizzazione utilizzata dal ruolo, consulta la pagina [AWS politica gestita: KafkaConnectServiceRolePolicy](mkc-security-iam-awsmanpol.md#security-iam-awsmanpol-KafkaConnectServiceRolePolicy).
Per consentire a un'entità IAM (come un utente, un gruppo o un ruolo) di creare, modificare o eliminare un ruolo collegato al servizio è necessario configurare le relative autorizzazioni. Per ulteriori informazioni, consulta [Autorizzazioni del ruolo collegato ai servizi](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#service-linked-role-permissions) nella *Guida per l'utente di IAM*.
## Creazione di un ruolo collegato ai servizi per MSK Connect
Non hai bisogno di creare manualmente un ruolo collegato ai servizi. Quando si crea un connettore nella Console di gestione AWS, la o l' AWS API AWS CLI, MSK Connect crea automaticamente il ruolo collegato al servizio.
Se elimini questo ruolo collegato al servizio, è possibile ricrearlo seguendo lo stesso processo utilizzato per ricreare il ruolo nell’account. Quando crei un connettore, MSK Connect crea di nuovo automaticamente il ruolo collegato ai servizi per conto dell'utente.
## Modifica di un ruolo collegato ai servizi per MSK Connect
MSK Connect non consente di modificare il ruolo collegato al AWSService RoleForKafkaConnect servizio. Dopo aver creato un ruolo collegato al servizio, non è possibile modificarne il nome, perché potrebbero farvi riferimento diverse entità. È possibile tuttavia modificarne la descrizione utilizzando IAM. Per ulteriori informazioni, consulta la sezione [Modifica di un ruolo collegato ai servizi](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#edit-service-linked-role) nella *Guida per l'utente di IAM*.
## Eliminazione di un ruolo collegato ai servizi per MSK Connect
È possibile utilizzare la console IAM, AWS CLI o l' AWS API per eliminare manualmente il ruolo collegato al servizio. Per farlo, sarà necessario eliminare innanzitutto manualmente i connettori MSK Connect e quindi eliminare il ruolo manualmente. Per ulteriori informazioni, consulta [Eliminazione del ruolo collegato al servizio](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#delete-service-linked-role) nella *Guida per l’utente di IAM*.
## Regioni supportate per i ruoli collegati ai servizi di MSK Connect
MSK Connect supporta l'utilizzo di ruoli collegati ai servizi in tutte le regioni in cui il servizio è disponibile. Per ulteriori informazioni, consulta [Regioni ed endpoint di AWS](https://docs.aws.amazon.com/general/latest/gr/rande.html).
# Abilita l'accesso a Internet per Amazon MSK Connect
Se il tuo connettore per Amazon MSK Connect richiede l'accesso a Internet, ti consigliamo di utilizzare le seguenti impostazioni Amazon Virtual Private Cloud (VPC) per abilitare tale accesso.
+ Configura il connettore con sottoreti private.
+ Crea un [gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) pubblico o un'[istanza NAT](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_NAT_Instance.html) per il tuo VPC in una sottorete pubblica. *Per ulteriori informazioni, consulta la pagina [Connect subnet a Internet o altro VPCs utilizzando dispositivi NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) nella Guida per l'*Amazon Virtual Private Cloud*utente.*
+ Consenti il traffico in uscita dalle sottoreti private verso il gateway o l'istanza NAT.
# Configurazione di un gateway NAT per Amazon MSK Connect
Nei passaggi seguenti viene illustrato come configurare un gateway NAT per abilitare l'accesso a Internet per un connettore. È necessario completare questi passaggi prima di creare un connettore in una sottorete privata.
## Prerequisiti completi per la configurazione di un gateway NAT
Verifica di disporre dei seguenti elementi.
+ L'ID del Amazon Virtual Private Cloud (VPC) associato al cluster. Ad esempio, *vpc-123456ab*.
+ Le sottoreti private IDs del tuo VPC. Ad esempio, *subnet-a1b2c3de*, *subnet-f4g5h6ij* e così via. Il connettore deve essere configurato con sottoreti private.
## Passaggi per abilitare l'accesso a Internet per il connettore
**Abilitazione dell'accesso a Internet per il connettore**
1. Apri la Amazon Virtual Private Cloud console all'indirizzo [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
1. Crea una sottorete pubblica con un nome descrittivo per il gateway NAT e prendi nota dell'ID della sottorete. Per istruzioni dettagliate, consulta la pagina [Create a subnet in your VPC](https://docs.aws.amazon.com/vpc/latest/userguide/working-with-vpcs.html#AddaSubnet).
1. Crea un gateway Internet in modo che il VPC possa comunicare con Internet e prendi nota dell'ID del gateway. Collega il gateway Internet al VPC. Per istruzioni, consulta la pagina [Create and attach an internet gateway](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Attach_Gateway).
1. Fornisci un gateway NAT pubblico in modo che gli host delle tue sottoreti private possano raggiungere la tua sottorete pubblica. Quando crei il gateway NAT, seleziona la sottorete pubblica creata in precedenza. Per istruzioni, consulta [Creazione di un gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html#nat-gateway-creating).
1. Configura le tabelle di routing. Per completare questa configurazione, occorrono in totale due tabelle di routing. Dovresti già disporre di una tabella di routing principale creata in automatico al momento della creazione del VPC. In questo passaggio creerai una tabella di routing aggiuntiva per la sottorete pubblica.
1. Utilizza le seguenti impostazioni per modificare la tabella di routing principale del tuo VPC in modo che le sottoreti private instradino il traffico verso il tuo gateway NAT. Per le istruzioni, consulta la pagina [Utilizzo delle tabelle di routing](https://docs.aws.amazon.com/vpc/latest/userguide/WorkWithRouteTables.html) nella *Guida per l'utente* di *Amazon Virtual Private Cloud*.
**Tabella di routing MSKC privata**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/it_it/msk/latest/developerguide/msk-connect-internet-access-private-subnets-example.html)
1. Segui le istruzioni riportate nella pagina [Creazione di una tabella di routing personalizzata](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Routing) per creare una tabella di routing per la sottorete pubblica. Quando crei la tabella, inserisci un nome descrittivo nel campo **Tag nome** per identificare a quale sottorete è associata la tabella. Ad esempio, **MSKC pubblica**.
1. Configura la tua tabella di routing **MSKC pubblica** utilizzando le seguenti impostazioni.
****
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/it_it/msk/latest/developerguide/msk-connect-internet-access-private-subnets-example.html)
# Comprendi i nomi di host DNS privati
Con il supporto dei nomi host Private DNS in MSK Connect, è possibile configurare i connettori per fare riferimento a nomi di dominio pubblici o privati. Il supporto dipende dai server DNS specificati nel *set di opzioni DHCP* del VPC.
Un set di opzioni DHCP è un gruppo di configurazioni di rete utilizzate dalle istanze EC2 nel VPC per comunicare tramite la rete VPC. Ogni VPC ha un set di opzioni DHCP predefinito ma è possibile creare un set di opzioni DHCP personalizzato se, ad esempio, si desidera che le istanze nel VPC utilizzino un server DNS diverso per la risoluzione dei nomi di dominio anziché il server DNS fornito da Amazon. Consulta la pagina [DHCP option sets in Amazon VPC](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_DHCP_Options.html).
Prima che la risoluzione DNS privata capability/feature fosse inclusa in MSK Connect, i connettori utilizzavano i resolver DNS VPC del servizio per le query DNS provenienti da un connettore cliente. I connettori non utilizzavano i server DNS definiti nei set di opzioni DHCP del VPC del cliente per la risoluzione DNS.
I connettori potevano fare riferimento solo ai nomi host nelle configurazioni dei connettori dei clienti o nei plug-in risolvibili pubblicamente. Non potevano risolvere nomi host privati definiti in una zona ospitata privatamente o utilizzare server DNS in una rete di altri clienti.
Senza Private DNS, i clienti che hanno scelto di rendere inaccessibili a Internet i propri database, data warehouse e sistemi come Secrets Manager nel proprio VPC, non potrebbero lavorare con i connettori MSK. I clienti utilizzano spesso nomi host DNS privati per conformarsi alle norme di sicurezza aziendali.
# Configura un set di opzioni DHCP VPC per il tuo connettore
I connettori utilizzano automaticamente i server DNS definiti nel set di opzioni DHCP del VPC al momento della creazione del connettore. Prima di creare un connettore, assicurati di configurare il set di opzioni DHCP del VPC per i requisiti di risoluzione del nome host DNS del connettore.
I connettori creati prima che la funzionalità del nome host Private DNS fosse disponibile in MSK Connect continuano a utilizzare la precedente configurazione di risoluzione DNS senza che sia necessaria alcuna modifica.
Se nel connettore hai bisogno soltanto di una risoluzione dei nomi host DNS risolvibile pubblicamente, per semplificare la configurazione ti consigliamo di utilizzare il VPC predefinito del tuo account quando crei il connettore. Per ulteriori informazioni sul server DNS fornito da Amazon o Risolutore Amazon Route 53, consulta la pagina [Amazon DNS Server](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#AmazonDNS) nella *Guida per l'utente di Amazon VPC*.
Se devi risolvere nomi host DNS privati, assicurati che le opzioni DHCP del VPC passato durante la creazione del connettore siano impostate correttamente. Per ulteriori informazioni, consulta la pagina [Work with DHCP option sets](https://docs.aws.amazon.com/vpc/latest/userguide/DHCPOptionSet.html) nella *Guida per l'utente di Amazon VPC*.
Quando configuri un set di opzioni DHCP per la risoluzione dei nomi host DNS privati, assicurati che il connettore possa raggiungere i server DNS personalizzati configurati nel set di opzioni DHCP. In caso contrario, la creazione del connettore avrà esito negativo.
Dopo aver personalizzato il set di opzioni DHCP del VPC, i connettori successivamente creati in tale VPC utilizzano i server DNS specificati nell'insieme di opzioni. Se modifichi il set di opzioni dopo aver creato un connettore, il connettore adotta le impostazioni del nuovo set di opzioni entro un paio di minuti.
# Configura gli attributi DNS per il tuo VPC
Assicurati di avere configurato correttamente gli attributi DNS del VPC come descritto nelle sezioni [DNS attributes in your VPC](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#vpc-dns-support) e [DNS hostnames](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#vpc-dns-hostnames) nella *Guida per l'utente di Amazon VPC*.
Consulta [Risolvere le query DNS tra VPCs e la tua rete](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/resolver.html) nella *Amazon Route 53 Developer Guide* per informazioni sull'uso degli endpoint resolver in entrata e in uscita per connettere altre reti al tuo VPC e lavorare con il tuo connettore.
# Gestisci gli errori di creazione dei connettori
Questa sezione descrive i possibili errori di creazione dei connettori associati alla risoluzione DNS e le operazioni suggerite per risolvere i problemi.
| Errore | Azione suggerita |
| --- | --- |
| La creazione del connettore ha esito negativo se una query di risoluzione DNS non riesce o se i server DNS non sono raggiungibili dal connettore. | Se hai configurato questi registri per il connettore, puoi visualizzare gli errori di creazione dei connettori dovuti a query di risoluzione CloudWatch DNS non riuscite nei tuoi registri. Controlla le configurazioni del server DNS e verifica la connettività di rete ai server DNS dal connettore. |
| Se si modifica la configurazione dei server DNS nel set di opzioni DHCP del VPC mentre un connettore è in esecuzione, le query di risoluzione DNS dal connettore possono avere esito negativo. Se la risoluzione DNS non riesce, alcune attività del connettore possono entrare in uno stato di errore. | Se hai configurato questi registri per il connettore, puoi visualizzare gli errori di creazione dei connettori dovuti a query di risoluzione DNS non riuscite nei tuoi CloudWatch registri. Le attività non riuscite dovrebbero riavviarsi automaticamente per riattivare il connettore. Se ciò non accade, puoi contattare il Supporto per riavviare le attività non riuscite relative al connettore oppure puoi ricreare il connettore. |
# Sicurezza per MSK Connect
Puoi utilizzare un endpoint VPC di interfaccia, alimentato da AWS PrivateLink, per impedire che il traffico tra Amazon VPC e Amazon MSK-Connect compatibile esca dalla rete Amazon. APIs Gli endpoint VPC di interfaccia non richiedono un gateway Internet, un dispositivo NAT, una connessione VPN o una connessione. AWS Direct Connect Per ulteriori informazioni, consulta [Usa Amazon MSK APIs con endpoint VPC di interfaccia](privatelink-vpc-endpoints.md).
# Registrazione per MSK Connect
MSK Connect è in grado di scrivere log eventi che è possibile utilizzare per eseguire il debug del connettore. Quando si crea un connettore, è possibile specificare nessuna, una o più delle seguenti destinazioni di log:
+ Amazon CloudWatch Logs: specificate il gruppo di log a cui desiderate che MSK Connect invii gli eventi di registro del connettore. Per informazioni su come creare un gruppo di log, consulta [Create a log group](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html#Create-Log-Group) nella *CloudWatch Logs* User Guide.
+ Amazon S3: specifica il bucket S3 a cui desideri che MSK Connect invii i log eventi del connettore. Per informazioni su come creare un bucket S3, consulta la pagina [Creating a bucket](https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket-overview.html) nella *Guida per l'utente di Amazon S3*.
+ Amazon Data Firehose: specifichi il flusso di distribuzione a cui desideri che MSK Connect invii gli eventi di registro del connettore. *Per informazioni su come creare un flusso di distribuzione, consulta [Creating an Amazon Data Firehose Delivery stream nella Firehose](https://docs.aws.amazon.com/firehose/latest/dev/basic-create.html) User Guide.*
Per ulteriori informazioni sulla configurazione della registrazione, consulta la pagina [Abilitazione della registrazione dai servizi AWS](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/AWS-logs-and-resource-policy.html) nella *Guida per l'utente di Amazon CloudWatch Logs *.
MSK Connect emette i seguenti tipi di log eventi:
****
| Livello | Description |
| --- | --- |
| INFO | Eventi di runtime di interesse all'avvio e all'arresto. |
| WARN | Situazioni di runtime che non sono errori ma sono indesiderate o impreviste. |
| FATAL | Errori gravi che causano una terminazione anticipata. |
| ERROR | Condizioni impreviste ed errori di runtime non fatali. |
Di seguito è riportato un esempio di evento di registro inviato a CloudWatch Logs:
```
[Worker-0bb8afa0b01391c41] [2021-09-06 16:02:54,151] WARN [Producer clientId=producer-1] Connection to node 1 (b-1.my-test-cluster.twwhtj.c2.kafka.us-east-1.amazonaws.com/INTERNAL_IP) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:782)
```
## Impedire la visualizzazione di segreti nei log dei connettori
**Nota**
Se un plug-in non definisce i valori di configurazione sensibili come segreti, tali valori possono apparire nei log dei connettori. Kafka Connect tratta i valori di configurazione non definiti allo stesso modo di qualsiasi altro valore non crittografato.
Se il plug-in definisce una proprietà come segreta, Kafka Connect oscura il valore della proprietà nei log dei connettori. Ad esempio, i seguenti log dei connettori mostrano che se un plug-in definisce `aws.secret.key` come un tipo `PASSWORD`, il suo valore viene sostituito con `[hidden]`.
```
2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] [2022-01-11 15:18:55,150] INFO SecretsManagerConfigProviderConfig values:
2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] aws.access.key = my_access_key
2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] aws.region = us-east-1
2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] aws.secret.key = [hidden]
2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] secret.prefix =
2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] secret.ttl.ms = 300000
2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] (com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProviderConfig:361)
```
Per evitare che nei file di log dei connettori appaiano dei segreti, gli sviluppatori di plug-in devono utilizzare la costante di enumerazione [https://kafka.apache.org/27/javadoc/org/apache/kafka/common/config/ConfigDef.Type.html#PASSWORD](https://kafka.apache.org/27/javadoc/org/apache/kafka/common/config/ConfigDef.Type.html#PASSWORD) di Kafka Connect per definire le proprietà sensibili. Quando una proprietà è di tipo `ConfigDef.Type.PASSWORD`, Kafka Connect esclude il relativo valore dai log dei connettori anche se il valore viene inviato come testo non crittografato.
# Monitoraggio di Amazon MSK Connect
Il monitoraggio è una parte importante per mantenere l'affidabilità, la disponibilità e le prestazioni di MSK Connect e delle altre AWS soluzioni. Amazon CloudWatch monitora AWS le tue risorse e le applicazioni su cui esegui AWS in tempo reale. È possibile raccogliere e tenere traccia dei parametri, creare pannelli di controllo personalizzati e impostare allarmi per inviare una notifica o intraprendere azioni quando un parametro specificato raggiunge una determinata soglia. Ad esempio, puoi tenere CloudWatch traccia dell'utilizzo della CPU o di altri parametri del connettore, in modo da aumentarne la capacità se necessario. Per ulteriori informazioni, consulta la [Amazon CloudWatch User Guide](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/).
Puoi utilizzare le seguenti operazioni API:
+ `DescribeConnectorOperation`: monitora lo stato delle operazioni di aggiornamento dei connettori.
+ `ListConnectorOperations`: Tieni traccia degli aggiornamenti precedenti eseguiti sul connettore.
La tabella seguente mostra le metriche inviate da MSK CloudWatch Connect all'interno della `ConnectorName` dimensione. MSK Connect fornisce queste metriche per impostazione predefinita e senza costi aggiuntivi. CloudWatch conserva queste metriche per 15 mesi, in modo da poter accedere alle informazioni storiche e avere una prospettiva migliore sulle prestazioni dei connettori. È anche possibile impostare allarmi che controllano determinate soglie e inviare notifiche o intraprendere azioni quando queste soglie vengono raggiunte. Per ulteriori informazioni, consulta la [Amazon CloudWatch User Guide](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/).
| Nome parametro | Description |
| --- | --- |
| CpuUtilization | La percentuale di utilizzo della CPU per sistema e utente. |
| ErroredTaskCount | Il numero di attività che sono state eseguite con errori. |
| MemoryUtilization | La percentuale della memoria totale su un'istanza worker, non solo la memoria heap della macchina virtuale Java (JVM) attualmente in uso. JVM in genere non restituisce la memoria al sistema operativo. Quindi, JVM heap size (MemoryUtilization) di solito inizia con una dimensione minima dell'heap che aumenta in modo incrementale fino a un massimo stabile di circa l'80-90%. L'utilizzo dell'heap JVM potrebbe aumentare o diminuire al variare dell'utilizzo effettivo della memoria da parte del connettore. |
| RebalanceCompletedTotal | Il numero totale di ribilanciamenti completati da questo connettore. |
| RebalanceTimeAvg | Il tempo medio in millisecondi impiegato dal connettore per il ribilanciamento. |
| RebalanceTimeMax | Il tempo massimo in millisecondi impiegato dal connettore per il ribilanciamento. |
| RebalanceTimeSinceLast | Il tempo in millisecondi trascorso dal momento in cui questo connettore ha completato il ribilanciamento più recente. |
| RunningTaskCount | Il numero di attività in esecuzione nel connettore. |
| SinkConsumerByteRate | Il numero medio di byte consumati al secondo dal consumatore Sink del framework Kafka Connect prima dell'applicazione di qualsiasi trasformazione ai dati. |
| SinkRecordReadRate | Il numero medio al secondo di record letti dal cluster Apache Kafka o Amazon MSK. |
| SinkRecordSendRate | Il numero medio al secondo di record emessi dalle trasformazioni e inviati alla destinazione. Questo numero non include i record filtrati. |
| SourceRecordPollRate | Il numero medio al secondo di record prodotti o sottoposti a polling. |
| SourceProducerByteRate | Il numero medio di byte prodotti al secondo dal produttore del codice sorgente del framework Kafka Connect dopo l'applicazione di eventuali trasformazioni ai dati. |
| SourceRecordWriteRate | Il numero medio al secondo di record derivati dalle trasformazioni e scritti sul cluster Apache Kafka o Amazon MSK. |
| TaskStartupAttemptsTotal | Il numero totale di tentativi di avvio di attività eseguiti dal connettore. È possibile utilizzare questo parametro per identificare le anomalie nei tentativi di avvio delle attività. |
| TaskStartupSuccessPercentage | La percentuale media di attività avviate correttamente dal connettore. È possibile utilizzare questo parametro per identificare le anomalie nei tentativi di avvio delle attività. |
| WorkerCount | Il numero minimo di worker in esecuzione nel connettore. |
| BytesInPerSec | Byte di metadati trasferiti al framework Kafka Connect per la comunicazione tra i lavoratori. |
| BytesOutPerSec | Byte di metadati trasferiti dal framework Kafka Connect per la comunicazione tra i lavoratori. |
# Esempi di configurazione delle risorse Amazon MSK Connect
Questa sezione include esempi per aiutarti a configurare risorse Amazon MSK Connect come connettori e provider di configurazione di terze parti comuni.
**Topics**
+ [Configura il connettore sink Amazon S3](mkc-S3sink-connector-example.md)
+ [Configurare il connettore del lavabo EventBridge Kafka per MSK Connect](mkc-eventbridge-kafka-connector.md)
+ [Usa il connettore sorgente Debezium con il provider di configurazione](mkc-debeziumsource-connector-example.md)
# Configura il connettore sink Amazon S3
Questo esempio mostra come utilizzare il connettore sink [Amazon S3 Confluent e AWS CLI come creare un connettore sink](https://www.confluent.io/hub/confluentinc/kafka-connect-s3) Amazon S3 in MSK Connect.
1. Copia il codice JSON seguente e incollalo in un nuovo file. Sostituisci le stringhe segnaposto con valori che corrispondono alla stringa di connessione dei server di bootstrap del cluster Amazon MSK e alla sottorete e al gruppo di sicurezza del cluster. IDs Per informazioni su come configurare un ruolo di esecuzione del servizio, consulta la pagina [Ruoli e policy IAM per MSK Connect](msk-connect-iam.md).
```
{
"connectorConfiguration": {
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"s3.region": "us-east-1",
"format.class": "io.confluent.connect.s3.format.json.JsonFormat",
"flush.size": "1",
"schema.compatibility": "NONE",
"topics": "my-test-topic",
"tasks.max": "2",
"partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"s3.bucket.name": "amzn-s3-demo-bucket"
},
"connectorName": "example-S3-sink-connector",
"kafkaCluster": {
"apacheKafkaCluster": {
"bootstrapServers": "",
"vpc": {
"subnets": [
"",
"",
""
],
"securityGroups": [""]
}
}
},
"capacity": {
"provisionedCapacity": {
"mcuCount": 2,
"workerCount": 4
}
},
"kafkaConnectVersion": "2.7.1",
"serviceExecutionRoleArn": "",
"plugins": [
{
"customPlugin": {
"customPluginArn": "",
"revision": 1
}
}
],
"kafkaClusterEncryptionInTransit": {"encryptionType": "PLAINTEXT"},
"kafkaClusterClientAuthentication": {"authenticationType": "NONE"}
}
```
1. Esegui il AWS CLI comando seguente nella cartella in cui hai salvato il file JSON nel passaggio precedente.
```
aws kafkaconnect create-connector --cli-input-json file://connector-info.json
```
Di seguito è riportato un esempio dell'output che si ottiene eseguendo correttamente il comando.
```
{
"ConnectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-S3-sink-connector/abc12345-abcd-4444-a8b9-123456f513ed-2",
"ConnectorState": "CREATING",
"ConnectorName": "example-S3-sink-connector"
}
```
# Configurare il connettore del lavabo EventBridge Kafka per MSK Connect
Questo argomento mostra come configurare il connettore [EventBridge Kafka sink per MSK Connect](https://github.com/awslabs/eventbridge-kafka-connector). [Questo connettore consente di inviare eventi dal cluster MSK ai bus degli eventi. EventBridge ](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-bus.html) Questo argomento descrive il processo di creazione delle risorse richieste e di configurazione del connettore per consentire un flusso di dati senza interruzioni tra Kafka e. EventBridge
**Topics**
+ [Prerequisiti](#mkc-eb-kafka-prerequisites)
+ [Configurare le risorse necessarie per MSK Connect](#mkc-eb-kafka-set-up-resources)
+ [Crea il connettore](#mkc-eb-kafka-create-connector)
+ [Invia messaggi a Kafka](#mkc-eb-kafka-send-json-encoded-messages)
## Prerequisiti
Prima di distribuire il connettore, assicurati di disporre delle seguenti risorse:
+ Cluster **Amazon MSK: un cluster MSK** attivo per la produzione e l'utilizzo di messaggi Kafka.
+ **Amazon EventBridge Event Bus**: un bus di EventBridge eventi per ricevere eventi sugli argomenti di Kafka.
+ **Ruoli IAM**: crea ruoli IAM con le autorizzazioni necessarie per MSK Connect e il EventBridge connettore.
+ [Accesso a Internet pubblico](msk-connect-internet-access.md) da MSK Connect o da un endpoint EventBridge di [interfaccia VPC creato](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-related-service-vpc.html) nel VPC e nella sottorete del cluster MSK. Ciò consente di evitare di attraversare la rete Internet pubblica e di non richiedere gateway NAT.
+ Una [macchina client](create-serverless-cluster-client.md), come un'istanza Amazon EC2 o [AWS CloudShell](https://aws.amazon.com/cloudshell/), per creare argomenti e inviare record a Kafka.
## Configurare le risorse necessarie per MSK Connect
Si crea un ruolo IAM per il connettore, quindi si crea il connettore. Crei anche una EventBridge regola per filtrare gli eventi Kafka inviati al bus degli EventBridge eventi.
**Topics**
+ [Ruolo IAM per il connettore](#mkc-eb-kafka-iam-role-connector)
+ [Una EventBridge regola per gli eventi in arrivo](#mkc-eb-kafka-create-rule)
### Ruolo IAM per il connettore
Il ruolo IAM associato al connettore deve disporre dell'[PutEvents](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-permissions-reference.html)autorizzazione per consentire l'invio di eventi a EventBridge. Il seguente esempio di policy IAM ti concede l'autorizzazione a inviare eventi a un bus di eventi denominato. `example-event-bus` Assicurati di sostituire la risorsa ARN nell'esempio seguente con l'ARN del tuo event bus.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"events:PutEvents"
],
"Resource": "arn:aws:events:us-east-1:123456789012:event-bus/example-event-bus"
}
]
}
```
------
Inoltre, devi assicurarti che il tuo ruolo IAM per il connettore contenga la seguente politica di attendibilità.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "kafkaconnect.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
### Una EventBridge regola per gli eventi in arrivo
Crei [regole](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-rules.html) che abbinano gli eventi in entrata ai criteri relativi ai dati degli eventi, noti come pattern di [https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html). [Con un pattern di eventi, è possibile definire i criteri per filtrare gli eventi in arrivo e determinare quali eventi devono attivare una particolare regola e successivamente essere indirizzati a un obiettivo designato.](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-targets.html) Il seguente esempio di schema di eventi corrisponde agli eventi di Kafka inviati al bus degli eventi. EventBridge
```
{
"detail": {
"topic": ["msk-eventbridge-tutorial"]
}
}
```
Di seguito è riportato un esempio di evento inviato da Kafka all' EventBridge utilizzo del connettore Kafka sink.
```
{
"version": "0",
"id": "dbc1c73a-c51d-0c0e-ca61-ab9278974c57",
"account": "123456789012",
"time": "2025-03-26T10:15:00Z",
"region": "us-east-1",
"detail-type": "msk-eventbridge-tutorial",
"source": "kafka-connect.msk-eventbridge-tutorial",
"resources": [],
"detail": {
"topic": "msk-eventbridge-tutorial",
"partition": 0,
"offset": 0,
"timestamp": 1742984100000,
"timestampType": "CreateTime",
"headers": [],
"key": "order-1",
"value": {
"orderItems": [
"item-1",
"item-2"
],
"orderCreatedTime": "Wed Mar 26 10:15:00 UTC 2025"
}
}
}
```
Nella EventBridge console, [create una regola](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-create-rule.html) sul bus degli eventi utilizzando questo modello di esempio e specificate un obiettivo, ad esempio un gruppo Logs. CloudWatch La EventBridge console configurerà automaticamente la politica di accesso necessaria per il gruppo CloudWatch Logs.
## Crea il connettore
Nella sezione seguente, si crea e si distribuisce il [connettore EventBridge Kafka sink](https://github.com/awslabs/eventbridge-kafka-connector) utilizzando il. Console di gestione AWS
**Topics**
+ [Passaggio 1: scarica il connettore](#mkc-eb-kafka-download-connector)
+ [Fase 2: creare un bucket Amazon S3](#mkc-eb-kafka-s3-bucket-create)
+ [Fase 3: Creare un plugin in MSK Connect](#mkc-eb-kafka-create-plugin)
+ [Passaggio 4: creazione del connettore](#mkc-eb-kafka-create-connector)
### Passaggio 1: scarica il connettore
Scarica l'ultimo EventBridge Connector Sink JAR dalla [pagina delle GitHub versioni](https://github.com/awslabs/eventbridge-kafka-connector/releases) del connettore EventBridge Kafka. Ad esempio, per scaricare la versione v1.4.1, scegli il link al file JAR per scaricare il `kafka-eventbridge-sink-with-dependencies.jar` connettore. Quindi, salva il file in una posizione preferita sul tuo computer.
### Fase 2: creare un bucket Amazon S3
1. Per archiviare il file JAR in Amazon S3 per utilizzarlo con MSK Connect, apri Console di gestione AWS e scegli Amazon S3.
1. Nella console Amazon S3, scegli **Crea bucket e inserisci un nome di bucket** univoco. Ad esempio, **amzn-s3-demo-bucket1-eb-connector**.
1. Scegli una regione appropriata per il tuo bucket Amazon S3. Assicurati che corrisponda alla regione in cui è distribuito il cluster MSK.
1. Per **le impostazioni di Bucket**, mantieni le selezioni predefinite o modificale secondo necessità.
1. Seleziona **Crea bucket**.
1. Carica il file JAR nel bucket Amazon S3.
### Fase 3: Creare un plugin in MSK Connect
1. Aprire Console di gestione AWS, quindi passare a **MSK Connect**.
1. Nel riquadro di navigazione a sinistra, scegliete **Plugin personalizzati**.
1. Scegli **Crea plug-in**, quindi inserisci il **nome del plug-in**. Ad esempio, **eventbridge-sink-plugin**.
1. Per la **posizione personalizzata del plug-in**, incolla l'**URL dell'oggetto S3**.
1. Aggiungi una descrizione opzionale per il plugin.
1. Scegli **Crea plugin**.
Dopo aver creato il plug-in, è possibile utilizzarlo per configurare e distribuire il connettore EventBridge Kafka in MSK Connect.
### Passaggio 4: creazione del connettore
Prima di creare il connettore, consigliamo di creare l'argomento Kafka richiesto per evitare errori nel connettore. Per creare l'argomento, utilizzate il computer client.
1. Nel riquadro sinistro della console MSK, scegli **Connettori**, quindi scegli **Crea connettore**.
1. Nell'elenco dei plugin, scegli **eventbridge-sink-plugin**, quindi scegli **Avanti**.
1. Per il nome del connettore, immettete**EventBridgeSink**.
1. Nell'elenco dei cluster, scegliete il vostro cluster MSK.
1. Copia la seguente configurazione per il connettore e incollala nel campo Configurazione del **connettore**
Se necessario, sostituite i segnaposto nella seguente configurazione.
+ Rimuovi `aws.eventbridge.endpoint.uri` se il cluster MSK dispone di un accesso pubblico a Internet.
+ Se si utilizza PrivateLink per connettersi in modo sicuro da MSK a EventBridge, sostituire la parte DNS successiva `https://` con il nome DNS privato corretto dell'endpoint di interfaccia VPC (opzionale) per quello creato in precedenza. EventBridge
+ Sostituisci l'ARN del bus di EventBridge eventi nella seguente configurazione con l'ARN del tuo bus di eventi.
+ Aggiorna tutti i valori specifici della regione.
```
{
"connector.class": "software.amazon.event.kafkaconnector.EventBridgeSinkConnector",
"aws.eventbridge.connector.id": "msk-eventbridge-tutorial",
"topics": "msk-eventbridge-tutorial",
"tasks.max": "1",
"aws.eventbridge.endpoint.uri": "https://events.us-east-1.amazonaws.com",
"aws.eventbridge.eventbus.arn": "arn:aws:events:us-east-1:123456789012:event-bus/example-event-bus",
"value.converter.schemas.enable": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"aws.eventbridge.region": "us-east-1",
"auto.offset.reset": "earliest",
"key.converter": "org.apache.kafka.connect.storage.StringConverter"
}
```
Per ulteriori informazioni sulla configurazione dei connettori, vedere. [eventbridge-kafka-connector](https://github.com/awslabs/eventbridge-kafka-connector)
Se necessario, modificate le impostazioni per i lavoratori e la scalabilità automatica. Consigliamo inoltre di utilizzare l'ultima versione disponibile (consigliata) di Apache Kafka Connect dal menu a discesa. In **Autorizzazioni di accesso**, usa il ruolo creato in precedenza. Si consiglia inoltre di abilitare la registrazione a CloudWatch per l'osservabilità e la risoluzione dei problemi. Modifica le altre impostazioni opzionali, come i tag, in base alle tue esigenze. Quindi, distribuisci il connettore e attendi che lo stato passi allo stato In esecuzione.
## Invia messaggi a Kafka
È possibile configurare le codifiche dei messaggi, come Apache Avro e JSON, specificando diversi convertitori utilizzando le impostazioni `value.converter` e, facoltativamente, disponibili in Kafka Connect. `key.converter`
[connector example](#connector-ex)In questo argomento è configurato per funzionare con messaggi con codifica JSON, come indicato dall'uso di for. `org.apache.kafka.connect.json.JsonConverter` `value converter` Quando il connettore è in stato di esecuzione, invia i record all'argomento `msk-eventbridge-tutorial` Kafka dal tuo computer client.
# Usa il connettore sorgente Debezium con il provider di configurazione
Questo esempio mostra come utilizzare il plug-in del connettore Debezium MySQL con un database [Amazon Aurora](https://aws.amazon.com/rds/aurora/) compatibile con MySQL come origine. In questo esempio, abbiamo anche configurato il provider open source [AWS Secrets Manager Config Provider](https://github.com/jcustenborder/kafka-config-provider-aws) per esternalizzare le credenziali del database in Gestione dei segreti AWS. Per ulteriori informazioni sui provider di configurazione, consulta la pagina [Tutorial: esternalizzazione di informazioni sensibili utilizzando provider di configurazione](msk-connect-config-provider.md).
**Importante**
Il plug-in del connettore Debezium MySQL [supporta solo un'attività](https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-property-tasks-max) e non funziona con la modalità di capacità con dimensionamento automatico per Amazon MSK Connect. Dovresti invece utilizzare la modalità di capacità assegnata e impostare il valore `workerCount` su uno una nella configurazione del connettore. Per ulteriori informazioni sulle modalità di capacità di MSK Connect, consulta la pagina [Comprendi la capacità dei connettori](msk-connect-capacity.md).
# Prerequisiti completi per utilizzare il connettore sorgente Debezium
Il tuo connettore deve essere in grado di accedere a Internet in modo da poter interagire con servizi come Gestione dei segreti AWS quelli esterni al tuo. Amazon Virtual Private Cloud I passaggi descritti in questa sezione consentono di completare le seguenti attività per abilitare l'accesso a Internet.
+ Configura una sottorete pubblica che ospita un gateway NAT e indirizza il traffico verso un gateway Internet nel tuo VPC.
+ Crea una route predefinita che indirizza il traffico della sottorete privata verso il gateway NAT.
Per ulteriori informazioni, consulta [Abilita l'accesso a Internet per Amazon MSK Connect](msk-connect-internet-access.md).
**Prerequisiti**
Prima di abilitare l'accesso a Internet, devi disporre dei seguenti elementi:
+ L'ID del Amazon Virtual Private Cloud (VPC) associato al cluster. Ad esempio, *vpc-123456ab*.
+ Le sottoreti private IDs del tuo VPC. Ad esempio, *subnet-a1b2c3de*, *subnet-f4g5h6ij* e così via. Il connettore deve essere configurato con sottoreti private.
**Abilitazione dell'accesso a Internet per il connettore**
1. Apri la console all' Amazon Virtual Private Cloud indirizzo. [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/)
1. Crea una sottorete pubblica con un nome descrittivo per il gateway NAT e prendi nota dell'ID della sottorete. Per istruzioni dettagliate, consulta la pagina [Create a subnet in your VPC](https://docs.aws.amazon.com/vpc/latest/userguide/working-with-vpcs.html#AddaSubnet).
1. Crea un gateway Internet in modo che il VPC possa comunicare con Internet e prendi nota dell'ID del gateway. Collega il gateway Internet al VPC. Per istruzioni, consulta la pagina [Create and attach an internet gateway](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Attach_Gateway).
1. Fornisci un gateway NAT pubblico in modo che gli host delle tue sottoreti private possano raggiungere la tua sottorete pubblica. Quando crei il gateway NAT, seleziona la sottorete pubblica creata in precedenza. Per istruzioni, consulta [Creazione di un gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html#nat-gateway-creating).
1. Configura le tabelle di routing. Per completare questa configurazione, occorrono in totale due tabelle di routing. Dovresti già disporre di una tabella di routing principale creata in automatico al momento della creazione del VPC. In questo passaggio creerai una tabella di routing aggiuntiva per la sottorete pubblica.
1. Utilizza le seguenti impostazioni per modificare la tabella di routing principale del tuo VPC in modo che le sottoreti private instradino il traffico verso il tuo gateway NAT. Per le istruzioni, consulta la pagina [Utilizzo delle tabelle di routing](https://docs.aws.amazon.com/vpc/latest/userguide/WorkWithRouteTables.html) nella *Guida per l'utente* di *Amazon Virtual Private Cloud*.
**Tabella di routing MSKC privata**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/it_it/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html)
1. Segui le istruzioni riportate nella pagina [Creazione di una tabella di routing personalizzata](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Routing) per creare una tabella di routing per la sottorete pubblica. Quando crei la tabella, inserisci un nome descrittivo nel campo **Tag nome** per identificare a quale sottorete è associata la tabella. Ad esempio, **MSKC pubblica**.
1. Configura la tua tabella di routing **MSKC pubblica** utilizzando le seguenti impostazioni.
****
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/it_it/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html)
Ora che hai abilitato l'accesso a Internet per Amazon MSK Connect, puoi creare un connettore.
# Crea un connettore sorgente Debezium
Questa procedura descrive come creare un connettore sorgente Debezium.
1.
**Creazione di un plug-in personalizzato**
1. Scarica il plug-in del connettore MySQL per l'ultima versione stabile dal sito [Debezium](https://debezium.io/releases/). Prendi nota della versione di rilascio di Debezium che scarichi (versione 2.x o la vecchia serie 1.x). Più avanti in questa procedura, creerai un connettore basato sulla tua versione di Debezium.
1. Scarica ed estrai [AWS Secrets Manager Config Provider](https://www.confluent.io/hub/jcustenborder/kafka-config-provider-aws).
1. Colloca i seguenti archivi nella stessa directory:
+ La cartella `debezium-connector-mysql`
+ La cartella `jcusten-border-kafka-config-provider-aws-0.1.1`
1. Comprimi la directory che hai creato nel passaggio precedente in un file ZIP, quindi carica il file ZIP in un bucket S3. Per istruzioni, consulta la pagina [Uploading objects in Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html) nella *Guida per l'utente di Amazon S3*.
1. Copia il codice JSON seguente e incollalo in un file. Ad esempio, `debezium-source-custom-plugin.json`. Sostituisci ** con il nome che desideri assegnare al plug-in, ** con l'ARN del bucket Amazon S3 in cui hai caricato il file ZIP `` e con la chiave del file dell'oggetto ZIP che hai caricato su S3.
```
{
"name": "",
"contentType": "ZIP",
"location": {
"s3Location": {
"bucketArn": "",
"fileKey": ""
}
}
}
```
1. Esegui il seguente AWS CLI comando dalla cartella in cui hai salvato il file JSON per creare un plugin.
```
aws kafkaconnect create-custom-plugin --cli-input-json file://
```
Verrà visualizzato un output simile al seguente.
```
{
"CustomPluginArn": "arn:aws:kafkaconnect:us-east-1:012345678901:custom-plugin/example-custom-plugin-name/abcd1234-a0b0-1234-c1-12345678abcd-1",
"CustomPluginState": "CREATING",
"Name": "example-custom-plugin-name",
"Revision": 1
}
```
1. Esegui il comando seguente per verificare lo stato del plug-in. Lo stato del cluster dovrebbe passare da `CREATING` a `ACTIVE`. Sostituisci il segnaposto ARN con l'ARN ottenuto nell'output del comando precedente.
```
aws kafkaconnect describe-custom-plugin --custom-plugin-arn ""
```
1.
**Configura Gestione dei segreti AWS e crea un segreto per le credenziali del tuo database**
1. Apri la console Secrets Manager all'indirizzo [https://console.aws.amazon.com/secretsmanager/](https://console.aws.amazon.com/secretsmanager/).
1. Crea un nuovo segreto per archiviare le credenziali di accesso al database. Per le istruzioni, consulta la pagina [Create a secret](https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_create-basic-secret.html) nella Guida per l'utente di *Gestione dei segreti AWS*.
1. Copia l'ARN del segreto.
1. Aggiungi le autorizzazioni di Secrets Manager dalla seguente policy di esempio al tuo [Comprendi il ruolo di esecuzione del servizio](msk-connect-service-execution-role.md). Sostituisci ** con l'ARN del tuo segreto.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"secretsmanager:GetResourcePolicy",
"secretsmanager:GetSecretValue",
"secretsmanager:DescribeSecret",
"secretsmanager:ListSecretVersionIds"
],
"Resource": [
"arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234"
]
}
]
}
```
------
Per istruzioni sull'aggiunta di autorizzazioni IAM, consulta la pagina [Adding and removing IAM identity permissions](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) nella *Guida per l'utente di IAM*.
1.
**Creazione di una configurazione del worker personalizzata con informazioni sul proprio provider di configurazione**
1. Copia le seguenti proprietà di configurazione del worker in un file, sostituendo le stringhe segnaposto con valori che corrispondono al tuo scenario. Per ulteriori informazioni sulle proprietà di configurazione per il provider di configurazione di AWS Secrets Manager Config, [SecretsManagerConfigProvider](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-config-provider-aws/configProviders/SecretsManagerConfigProvider.html)consultate la documentazione del plugin.
```
key.converter=
value.converter=
config.providers.secretManager.class=com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider
config.providers=secretManager
config.providers.secretManager.param.aws.region=
```
1. Esegui il AWS CLI comando seguente per creare la tua configurazione di lavoro personalizzata.
Sostituisci i valori seguenti:
+ **- un nome descrittivo per la configurazione personalizzata del lavoratore
+ **- una versione con codifica base64 delle proprietà di testo in chiaro copiate nel passaggio precedente
```
aws kafkaconnect create-worker-configuration --name --properties-file-content
```
1.
**Creazione di un connettore**
1. Copia il codice JSON seguente, che corrisponde alla tua versione di Debezium (2.x o 1.x), e incollalo in un nuovo file. Sostituisci le stringhe `` con valori che corrispondono al tuo scenario. Per informazioni su come configurare un ruolo di esecuzione del servizio, consulta la pagina [Ruoli e policy IAM per MSK Connect](msk-connect-iam.md).
Nota che la configurazione utilizza variabili come `${secretManager:MySecret-1234:dbusername}` anziché testo non crittografato per specificare le credenziali del database. Sostituisci `MySecret-1234` con il nome del tuo segreto, poi includi il nome della chiave che desideri recuperare. È inoltre necessario sostituire `` con l'ARN della configurazione del worker personalizzata.
------
#### [ Debezium 2.x ]
Per le versioni di Debezium 2.x, copia il codice JSON seguente e incollalo in un nuovo file. Sostituisci le stringhe ** con valori che corrispondono al tuo scenario.
```
{
"connectorConfiguration": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "",
"database.port": "3306",
"database.user": "<${secretManager:MySecret-1234:dbusername}>",
"database.password": "<${secretManager:MySecret-1234:dbpassword}>",
"database.server.id": "123456",
"database.include.list": "",
"topic.prefix": "",
"schema.history.internal.kafka.topic": "",
"schema.history.internal.kafka.bootstrap.servers": "",
"schema.history.internal.consumer.security.protocol": "SASL_SSL",
"schema.history.internal.consumer.sasl.mechanism": "AWS_MSK_IAM",
"schema.history.internal.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
"schema.history.internal.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
"schema.history.internal.producer.security.protocol": "SASL_SSL",
"schema.history.internal.producer.sasl.mechanism": "AWS_MSK_IAM",
"schema.history.internal.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
"schema.history.internal.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
"include.schema.changes": "true"
},
"connectorName": "example-Debezium-source-connector",
"kafkaCluster": {
"apacheKafkaCluster": {
"bootstrapServers": "",
"vpc": {
"subnets": [
"",
"",
""
],
"securityGroups": [""]
}
}
},
"capacity": {
"provisionedCapacity": {
"mcuCount": 2,
"workerCount": 1
}
},
"kafkaConnectVersion": "2.7.1",
"serviceExecutionRoleArn": "",
"plugins": [{
"customPlugin": {
"customPluginArn": "",
"revision": 1
}
}],
"kafkaClusterEncryptionInTransit": {
"encryptionType": "TLS"
},
"kafkaClusterClientAuthentication": {
"authenticationType": "IAM"
},
"workerConfiguration": {
"workerConfigurationArn": "",
"revision": 1
}
}
```
------
#### [ Debezium 1.x ]
Per le versioni di Debezium 1.x, copia il codice JSON seguente e incollalo in un nuovo file. Sostituisci le stringhe ** con valori che corrispondono al tuo scenario.
```
{
"connectorConfiguration": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "",
"database.port": "3306",
"database.user": "<${secretManager:MySecret-1234:dbusername}>",
"database.password": "<${secretManager:MySecret-1234:dbpassword}>",
"database.server.id": "123456",
"database.server.name": "",
"database.include.list": "",
"database.history.kafka.topic": "",
"database.history.kafka.bootstrap.servers": "",
"database.history.consumer.security.protocol": "SASL_SSL",
"database.history.consumer.sasl.mechanism": "AWS_MSK_IAM",
"database.history.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
"database.history.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
"database.history.producer.security.protocol": "SASL_SSL",
"database.history.producer.sasl.mechanism": "AWS_MSK_IAM",
"database.history.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
"database.history.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
"include.schema.changes": "true"
},
"connectorName": "example-Debezium-source-connector",
"kafkaCluster": {
"apacheKafkaCluster": {
"bootstrapServers": "",
"vpc": {
"subnets": [
"",
"",
""
],
"securityGroups": [""]
}
}
},
"capacity": {
"provisionedCapacity": {
"mcuCount": 2,
"workerCount": 1
}
},
"kafkaConnectVersion": "2.7.1",
"serviceExecutionRoleArn": "",
"plugins": [{
"customPlugin": {
"customPluginArn": "",
"revision": 1
}
}],
"kafkaClusterEncryptionInTransit": {
"encryptionType": "TLS"
},
"kafkaClusterClientAuthentication": {
"authenticationType": "IAM"
},
"workerConfiguration": {
"workerConfigurationArn": "",
"revision": 1
}
}
```
------
1. Esegui il AWS CLI comando seguente nella cartella in cui hai salvato il file JSON nel passaggio precedente.
```
aws kafkaconnect create-connector --cli-input-json file://connector-info.json
```
Di seguito è riportato un esempio dell'output che si ottiene eseguendo correttamente il comando.
```
{
"ConnectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2",
"ConnectorState": "CREATING",
"ConnectorName": "example-Debezium-source-connector"
}
```
# Aggiorna la configurazione di un connettore Debezium
Per aggiornare la configurazione del connettore Debezium, segui questi passaggi:
1. Copia il seguente codice JSON e incollalo in un nuovo file. Sostituisci le stringhe `` con valori che corrispondono al tuo scenario.
```
{
"connectorArn": ,
"connectorConfiguration": ,
"currentVersion":
}
```
1. Esegui il AWS CLI comando seguente nella cartella in cui hai salvato il file JSON nel passaggio precedente.
```
aws kafkaconnect update-connector --cli-input-json file://connector-info.json
```
Di seguito è riportato un esempio dell'output ottenuto quando si esegue correttamente il comando.
```
{
"connectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2",
"connectorOperationArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector-operation/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2/41b6ad56-3184-479b-850a-a8bedd5a02f3",
"connectorState": "UPDATING"
}
```
1. È ora possibile eseguire il comando seguente per monitorare lo stato corrente dell'operazione:
```
aws kafkaconnect describe-connector-operation --connector-operation-arn
```
Per un esempio di connettore Debezium con i passaggi dettagliati, consulta la pagina [Introducing Amazon MSK Connect - Stream Data to and from Your Apache Kafka Clusters Using Managed Connectors](https://aws.amazon.com/blogs/aws/introducing-amazon-msk-connect-stream-data-to-and-from-your-apache-kafka-clusters-using-managed-connectors/).
# Migrazione ad Amazon MSK Connect
Questa sezione descrive come migrare l'applicazione del connettore Apache Kafka su Amazon Managed Streaming for Apache Kafka Connect (Amazon MSK Connect). Per ulteriori informazioni sui vantaggi della migrazione ad Amazon MSK Connect, consulta. [Vantaggi dell'utilizzo di Amazon MSK Connect](msk-connect.md#msk-connect-benefits)
Questa sezione descrive anche gli argomenti di gestione dello stato utilizzati da Kafka Connect e Amazon MSK Connect e descrive le procedure per la migrazione dei connettori source e sink.
# Comprendi gli argomenti interni utilizzati da Kafka Connect
Un'applicazione Apache Kafka Connect in esecuzione in modalità distribuita memorizza il proprio stato utilizzando argomenti interni nel cluster Kafka e l'appartenenza ai gruppi. I seguenti sono i valori di configurazione che corrispondono agli argomenti interni utilizzati per le applicazioni Kafka Connect:
+ Argomento di configurazione, specificato tramite `config.storage.topic`
Nell'argomento di configurazione, Kafka Connect memorizza la configurazione di tutti i connettori e le attività avviate dagli utenti. Ogni volta che gli utenti aggiornano la configurazione di un connettore o quando un connettore richiede una riconfigurazione (ad esempio, il connettore rileva che può avviare più attività), viene emesso un record su questo argomento. Questo argomento è abilitato alla compattazione, quindi mantiene sempre l'ultimo stato per ogni entità.
+ Argomento Offsets, specificato tramite `offset.storage.topic`
Nell'argomento offset, Kafka Connect memorizza gli offset dei connettori sorgente. Come l'argomento sulla configurazione, l'argomento offset è abilitato alla compattazione. Questo argomento viene utilizzato per scrivere le posizioni di origine solo per i connettori di origine che producono dati a Kafka da sistemi esterni. I connettori Sink, che leggono i dati da Kafka e li inviano a sistemi esterni, memorizzano i dati di consumo utilizzando i normali gruppi di consumatori Kafka.
+ Argomento di stato, specificato tramite `status.storage.topic`
Nell'argomento relativo allo stato, Kafka Connect memorizza lo stato corrente dei connettori e delle attività. Questo argomento viene utilizzato come punto centrale per i dati richiesti dagli utenti dell'API REST. Questo argomento consente agli utenti di interrogare qualsiasi worker e ottenere comunque lo stato di tutti i plugin in esecuzione. Come gli argomenti di configurazione e offset, anche l'argomento status è abilitato alla compattazione.
Oltre a questi argomenti, Kafka Connect fa ampio uso dell'API di appartenenza ai gruppi di Kafka. I gruppi prendono il nome dal nome del connettore. Ad esempio, per un connettore denominato file-sink, il gruppo viene denominato. connect-file-sink Ogni consumatore del gruppo fornisce i record relativi a una singola attività. Questi gruppi e i relativi offset possono essere recuperati utilizzando i normali strumenti dei gruppi di consumatori, come. `Kafka-consumer-group.sh` Per ogni connettore sink, il runtime Connect esegue un normale gruppo di consumatori che estrae i record da Kafka.
# Gestione dello stato delle applicazioni Amazon MSK Connect
Per impostazione predefinita, Amazon MSK Connect crea tre argomenti separati nel cluster Kafka per ogni connettore Amazon MSK per memorizzare la configurazione, l'offset e lo stato del connettore. I nomi degli argomenti predefiniti sono strutturati come segue:
+ *connector-name*\$1\$1msk\$1connect\$1configs\$1 \$1 *connector-id*
+ \$1\$1msk\$1connect\$1status\$1 \$1 *connector-name* *connector-id*
+ \$1\$1msk\$1connect\$1offsets\$1 \$1 *connector-name* *connector-id*
**Nota**
Per fornire la continuità dell'offset tra i connettori di origine, puoi utilizzare un argomento di archiviazione dell'offset a tua scelta, anziché l'argomento predefinito. La definizione di un argomento di archiviazione degli offset consente di eseguire attività come la creazione di un connettore di origine che riprenda la lettura dall'ultimo offset di un connettore precedente. Per specificare un argomento di storage offset, fornisci un valore per la [https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-manage-connector-offsets](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-manage-connector-offsets)proprietà nella configurazione del worker Amazon MSK Connect prima di creare il connettore.
# Migrazione dei connettori di origine ad Amazon MSK Connect
I connettori di origine sono applicazioni Apache Kafka Connect che importano record da sistemi esterni in Kafka. Questa sezione descrive il processo di migrazione delle applicazioni del connettore di origine Apache Kafka Connect che eseguono cluster Kafka Connect locali o autogestiti in esecuzione su Amazon MSK Connect. AWS
L'applicazione Kafka Connect source connector memorizza gli offset in un argomento denominato con il valore impostato per la proprietà config. `offset.storage.topic` Di seguito sono riportati alcuni esempi di messaggi di offset per un connettore JDBC che esegue due attività che importano dati da due tabelle diverse denominate e. `movies` `shows` La riga più recente importata dai film da tavolo ha un ID primario di. `18343` La riga più recente importata dalla tabella shows ha un ID primario di`732`.
```
["jdbcsource",{"protocol":"1","table":"sample.movies"}] {"incrementing":18343}
["jdbcsource",{"protocol":"1","table":"sample.shows"}] {"incrementing":732}
```
Per migrare i connettori di origine su Amazon MSK Connect, procedi come segue:
1. Crea un [plug-in personalizzato](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html) Amazon MSK Connect estraendo le librerie di connettori dal tuo cluster Kafka Connect locale o autogestito.
1. Crea [le proprietà dei lavoratori](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-config-provider.html#msk-connect-config-providers-create-custom-config) Amazon MSK Connect e imposta le proprietà `key.converter` e `offset.storage.topic` gli stessi valori impostati per il connettore Kafka in esecuzione nel tuo cluster Kafka Connect esistente. `value.converter`
1. Metti in pausa l'applicazione del connettore sul cluster esistente effettuando una `PUT /connectors/connector-name/pause` richiesta sul cluster Kafka Connect esistente.
1. Assicurati che tutte le attività dell'applicazione del connettore siano completamente interrotte. È possibile interrompere le attività effettuando una `GET /connectors/connector-name/status` richiesta sul cluster Kafka Connect esistente o consumando i messaggi dal nome dell'argomento impostato per la proprietà. `status.storage.topic`
1. Ottieni la configurazione del connettore dal cluster esistente. È possibile ottenere la configurazione del connettore effettuando una `GET /connectors/connector-name/config/` richiesta sul cluster esistente o utilizzando i messaggi dal nome dell'argomento impostato per la proprietà`config.storage.topic`.
1. Crea un nuovo [Amazon MSK Connector](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html) con lo stesso nome di un cluster esistente. Crea questo connettore utilizzando il plug-in personalizzato del connettore che hai creato nel passaggio 1, le proprietà del worker che hai creato nel passaggio 2 e la configurazione del connettore che hai estratto nel passaggio 5.
1. Quando lo stato di Amazon MSK Connector è `active` impostato su, visualizza i log per verificare che il connettore abbia iniziato a importare dati dal sistema di origine.
1. Elimina il connettore nel cluster esistente effettuando una richiesta. `DELETE /connectors/connector-name`
# Migrazione dei connettori sink ad Amazon MSK Connect
I connettori Sink sono applicazioni Apache Kafka Connect che esportano dati da Kafka a sistemi esterni. Questa sezione descrive il processo di migrazione delle applicazioni sink connector di Apache Kafka Connect che eseguono cluster Kafka Connect locali o autogestiti in esecuzione su Amazon MSK Connect. AWS
I connettori sink Kafka Connect utilizzano l'API Kafka Group Membership e memorizzano gli offset negli stessi `__consumer_offset` argomenti di una tipica applicazione consumer. Questo comportamento semplifica la migrazione del connettore sink da un cluster autogestito ad Amazon MSK Connect.
Per migrare i connettori sink su Amazon MSK Connect, procedi come segue:
1. Crea un [plug-in personalizzato](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html) Amazon MSK Connect estraendo le librerie di connettori dal tuo cluster Kafka Connect locale o autogestito.
1. Crea [le proprietà dei lavoratori](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-config-provider.html#msk-connect-config-providers-create-custom-config) Amazon MSK Connect e imposta le proprietà `key.converter` e `value.converter` gli stessi valori impostati per il connettore Kafka in esecuzione nel cluster Kafka Connect esistente.
1. Metti in pausa l'applicazione del connettore sul cluster esistente effettuando una `PUT /connectors/connector-name/pause` richiesta sul cluster Kafka Connect esistente.
1. Assicurati che tutte le attività dell'applicazione del connettore siano completamente interrotte. È possibile interrompere le attività effettuando una `GET /connectors/connector-name/status` richiesta sul cluster Kafka Connect esistente o consumando i messaggi dal nome dell'argomento impostato per la proprietà. `status.storage.topic`
1. Ottieni la configurazione del connettore dal cluster esistente. È possibile ottenere la configurazione del connettore effettuando una `GET /connectors/connector-name/config` richiesta sul cluster esistente o utilizzando i messaggi dal nome dell'argomento impostato per la proprietà`config.storage.topic`.
1. Crea un nuovo [Amazon MSK Connector](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html) con lo stesso nome del cluster esistente. Crea questo connettore utilizzando il plug-in personalizzato del connettore che hai creato nel passaggio 1, le proprietà del worker che hai creato nel passaggio 2 e la configurazione del connettore che hai estratto nel passaggio 5.
1. Quando lo stato di Amazon MSK Connector è `active` impostato su, visualizza i log per verificare che il connettore abbia iniziato a importare dati dal sistema di origine.
1. Elimina il connettore nel cluster esistente effettuando una richiesta. `DELETE /connectors/connector-name`
# Risolvi i problemi in Amazon MSK Connect
Le seguenti informazioni agevolano la risoluzione dei problemi che si potrebbero verificare durante l'utilizzo di MSK Connect. Puoi anche pubblicare il problema nel [AWS re:Post](https://repost.aws/).
**Il connettore non è in grado di accedere alle risorse ospitate sulla rete Internet pubblica**
Consulta la sezione [Abilitazione dell'accesso a Internet per Amazon MSK Connect](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-internet-access.html).
**Il numero di attività in esecuzione del connettore non corrisponde al numero di attività specificato in tasks.max**
Ecco alcuni motivi per cui un connettore può utilizzare un numero inferiore di attività rispetto alla configurazione tasks.max specificata:
+ Alcune implementazioni di connettori limitano il numero di attività che è possibile utilizzare. Ad esempio, il connettore Debezium per MySQL è limitato all'utilizzo di una singola attività.
+ Quando si utilizza la modalità di capacità con scalabilità automatica, Amazon MSK Connect sostituisce la proprietà tasks.max del connettore con un valore proporzionale al numero di lavoratori in esecuzione nel connettore e al numero di lavoratori per lavoratore. MCUs Se hai configurato il parametro opzionale`maxAutoscalingTaskCount`, il valore non supererà questo limite. `tasks.max` Per ulteriori informazioni, consulta [Comprendere il numero massimo di attività di scalabilità automatica.](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html#msk-connect-max-autoscaling-task-count)
+ Per i connettori sink, il livello di parallelismo (numero di attività) non può essere superiore al numero di partizioni di argomento. Sebbene sia possibile impostare tasks.max su un valore maggiore, una singola partizione non viene mai elaborata da più di una singola attività alla volta.
+ In Kafka Connect 2.7.x, l'assegnatore di partizioni dei consumatori predefinito è `RangeAssignor`. Il comportamento di questo assegnatore è quello di assegnare la prima partizione di ogni argomento a un singolo consumatore, la seconda partizione di ogni argomento a un singolo consumatore ecc. Ciò significa che il numero massimo di attività attive utilizzate da un connettore sink tramite `RangeAssignor` è uguale al numero massimo di partizioni utilizzate in ogni singolo argomento utilizzato. Se ciò non funziona per il tuo caso d'uso, dovresti [creare una configurazione del worker](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-create-custom-worker-config) in cui la proprietà `consumer.partition.assignment.strategy` sia impostata su un assegnatore di partizioni dei consumatori più adatto. *Vedi [Interfaccia Kafka 2.7 ConsumerPartitionAssignor:](https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.html) tutte le classi di implementazione conosciute.*