Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Memahami MSK Connect
MSK Connect adalah fitur Amazon MSK yang memudahkan pengembang untuk melakukan streaming data ke dan dari cluster Apache Kafka mereka. MSK Connect menggunakan Kafka Connect versi 2.7.1 atau 3.7.x, yang merupakan kerangka kerja sumber terbuka untuk menghubungkan kluster Apache Kafka dengan sistem eksternal seperti database, indeks pencarian, dan sistem file. Dengan MSK Connect, Anda dapat menggunakan konektor terkelola penuh yang dibuat untuk Kafka Connect yang memindahkan data ke atau menarik data dari penyimpanan data populer seperti Amazon S3 dan Amazon Service. OpenSearch Anda dapat menerapkan konektor yang dikembangkan oleh pihak ke-3 seperti Debezium untuk streaming log perubahan dari database ke cluster Apache Kafka, atau menyebarkan konektor yang ada tanpa perubahan kode. Konektor secara otomatis menskalakan untuk menyesuaikan perubahan beban dan Anda hanya membayar untuk sumber daya yang Anda gunakan.
Gunakan konektor sumber untuk mengimpor data dari sistem eksternal ke topik Anda. Dengan konektor wastafel, Anda dapat mengekspor data dari topik Anda ke sistem eksternal.
MSK Connect mendukung konektor untuk cluster Apache Kafka apa pun dengan konektivitas ke VPC Amazon, apakah itu cluster MSK atau cluster Apache Kafka yang dihosting secara independen.
MSK Connect terus memantau kesehatan konektor dan status pengiriman, menambal dan mengelola perangkat keras yang mendasarinya, dan menskalakan konektor secara otomatis agar sesuai dengan perubahan throughput.
Untuk mulai menggunakan MSK Connect, lihat[Memulai MSK Connect](msk-connect-getting-started.md).
Untuk mempelajari AWS sumber daya yang dapat Anda buat dengan MSK Connect, lihat[Memahami konektor](msk-connect-connectors.md),[Buat plugin kustom](msk-connect-plugins.md), dan[Memahami pekerja MSK Connect](msk-connect-workers.md).
Untuk informasi tentang MSK Connect API, lihat Referensi [API Amazon MSK Connect](https://docs.aws.amazon.com/MSKC/latest/mskc/Welcome.html).
## Manfaat menggunakan Amazon MSK Connect
Apache Kafka adalah salah satu platform streaming open source yang paling banyak diadopsi untuk menelan dan memproses aliran data real-time. Dengan Apache Kafka, Anda dapat memisahkan dan secara mandiri menskalakan aplikasi penghasil data dan penggunaan data Anda.
Kafka Connect adalah komponen penting dalam membangun dan menjalankan aplikasi streaming dengan Apache Kafka. Kafka Connect menyediakan cara standar untuk memindahkan data antara Kafka dan sistem eksternal. Kafka Connect sangat skalabel dan dapat menangani volume besar data Kafka Connect menyediakan serangkaian operasi dan alat API yang kuat untuk mengonfigurasi, menyebarkan, dan memantau konektor yang memindahkan data antara topik Kafka dan sistem eksternal. Anda dapat menggunakan alat ini untuk menyesuaikan dan memperluas fungsionalitas Kafka Connect untuk memenuhi kebutuhan spesifik aplikasi streaming Anda.
Anda mungkin menghadapi tantangan saat mengoperasikan kluster Apache Kafka Connect sendiri, atau ketika Anda mencoba memigrasi aplikasi Apache Kafka Connect open source ke. AWS Tantangan ini termasuk waktu yang diperlukan untuk menyiapkan infrastruktur dan penerapan aplikasi, hambatan rekayasa saat menyiapkan cluster Apache Kafka Connect yang dikelola sendiri, dan overhead operasional administratif.
Untuk mengatasi tantangan ini, sebaiknya gunakan Amazon Managed Streaming for Apache Kafka Connect (Amazon MSK Connect) untuk memigrasikan aplikasi Apache Kafka Connect open source Anda ke. AWS Amazon MSK Connect menyederhanakan penggunaan Kafka Connect untuk mengalirkan data ke dan dari antara cluster Apache Kafka dan sistem eksternal, seperti database, indeks pencarian, dan sistem file.
Berikut adalah beberapa manfaat untuk bermigrasi ke Amazon MSK Connect:
+ **Penghapusan overhead operasional** - Amazon MSK Connect menghilangkan beban operasional yang terkait dengan penambalan, penyediaan, dan penskalaan cluster Apache Kafka Connect. Amazon MSK Connect terus memantau kesehatan cluster Connect Anda dan mengotomatiskan patching dan upgrade versi tanpa menyebabkan gangguan pada beban kerja Anda.
+ **Memulai ulang tugas Connect secara otomatis** — Amazon MSK Connect dapat memulihkan tugas yang gagal secara otomatis untuk mengurangi gangguan produksi. Kegagalan tugas dapat disebabkan oleh kesalahan sementara, seperti melanggar batas koneksi TCP untuk Kafka, dan penyeimbangan kembali tugas ketika pekerja baru bergabung dengan grup konsumen untuk konektor wastafel.
+ **Penskalaan horizontal dan vertikal otomatis** — Amazon MSK Connect memungkinkan aplikasi konektor untuk menskalakan secara otomatis untuk mendukung throughput yang lebih tinggi. Amazon MSK Connect mengelola penskalaan untuk Anda. Anda hanya perlu menentukan jumlah pekerja dalam grup penskalaan otomatis dan ambang batas pemanfaatan. Anda dapat menggunakan operasi Amazon MSK Connect `UpdateConnector` API untuk meningkatkan atau menurunkan skala v secara vertikal CPUs antara 1 dan 8 v CPUs untuk mendukung throughput variabel.
+ **Konektivitas jaringan pribadi** — Amazon MSK Connect terhubung secara pribadi ke sistem sumber dan sink dengan menggunakan AWS PrivateLink dan nama DNS pribadi.
# Memulai MSK Connect
Ini adalah step-by-step tutorial yang menggunakan Konsol Manajemen AWS untuk membuat cluster MSK dan konektor sink yang mengirimkan data dari cluster ke bucket S3.
**Topics**
+ [Menyiapkan sumber daya yang diperlukan untuk MSK Connect](mkc-tutorial-setup.md)
+ [Buat plugin kustom](mkc-create-plugin.md)
+ [Buat mesin klien dan topik Apache Kafka](mkc-create-topic.md)
+ [Buat konektor](mkc-create-connector.md)
+ [Kirim data ke cluster MSK](mkc-send-data.md)
# Menyiapkan sumber daya yang diperlukan untuk MSK Connect
Pada langkah ini Anda membuat sumber daya berikut yang Anda butuhkan untuk skenario memulai ini:
+ Bucket Amazon S3 berfungsi sebagai tujuan yang menerima data dari konektor.
+ Cluster MSK tempat Anda akan mengirim data. Konektor kemudian akan membaca data dari cluster ini dan mengirimkannya ke bucket S3 tujuan.
+ Kebijakan IAM yang berisi izin untuk menulis ke bucket S3 tujuan.
+ Peran IAM yang memungkinkan konektor untuk menulis ke bucket S3 tujuan. Anda akan menambahkan kebijakan IAM yang Anda buat ke peran ini.
+ Titik akhir VPC Amazon untuk memungkinkan pengiriman data dari VPC Amazon yang memiliki cluster dan konektor ke Amazon S3.
**Untuk membuat bucket S3**
1. Masuk ke Konsol Manajemen AWS dan buka konsol Amazon S3 di. [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/)
1. Pilih **Buat bucket**.
1. Untuk nama bucket, masukkan nama deskriptif seperti`amzn-s3-demo-bucket-mkc-tutorial`.
1. Gulir ke bawah dan pilih **Buat ember**.
1. Dalam daftar ember, pilih bucket yang baru dibuat.
1. Pilih **Buat folder**.
1. Masukkan `tutorial` nama folder, lalu gulir ke bawah dan pilih **Buat folder**.
**Untuk membuat cluster**
1. Buka konsol MSK Amazon di [https://console.aws.amazon.com/msk/rumah? region=us-east-1\$1/home/](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/).
1. **Di panel kiri, di bawah **MSK Clusters, pilih Cluster**.**
1. Pilih **Buat klaster**.
1. Dalam **metode Creation**, pilih **Custom create**.
1. Untuk nama cluster masukkan**mkc-tutorial-cluster**.
1. Pada **tipe Cluster**, pilih **Provisioned**.
1. Pilih **Berikutnya**.
1. Di bawah **Networking**, pilih VPC Amazon. Kemudian pilih Availability Zones dan subnet yang ingin Anda gunakan. Ingat VPC Amazon dan subnet yang Anda pilih karena Anda membutuhkannya nanti dalam tutorial ini. IDs
1. Pilih **Berikutnya**.
1. Di bawah **Metode kontrol akses** memastikan bahwa hanya akses **Tidak Diautentikasi yang dipilih**.
1. Di bawah **Enkripsi** pastikan bahwa hanya **Plaintext** yang dipilih.
1. Lanjutkan melalui wizard dan kemudian pilih **Buat cluster**. Ini membawa Anda ke halaman detail untuk cluster. Pada halaman itu, di bawah **Grup keamanan diterapkan**, temukan ID grup keamanan. Ingat ID itu karena Anda membutuhkannya nanti dalam tutorial ini.
**Untuk membuat kebijakan IAM dengan izin untuk menulis ke bucket S3**
1. Buka konsol IAM di [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/).
1. Pada panel navigasi, pilih **Kebijakan**.
1. Pilih **Buat kebijakan**.
1. Di **Editor kebijakan**, pilih **JSON**, lalu ganti JSON di jendela editor dengan JSON berikut.
Dalam contoh berikut, ganti ** dengan nama bucket S3 Anda.
------
#### [ 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:::/*"
}
]
}
```
------
Untuk petunjuk tentang cara menulis kebijakan aman, lihat[Kontrol akses IAM](iam-access-control.md).
1. Pilih **Berikutnya**.
1. Pada halaman **Review dan create**, lakukan hal berikut:
1. Untuk **nama Kebijakan**, masukkan nama deskriptif, seperti**mkc-tutorial-policy**.
1. Di **Izin yang ditentukan dalam kebijakan ini**, tinjau and/or edit izin yang ditentukan dalam kebijakan Anda.
1. (Opsional) Untuk membantu mengidentifikasi, mengatur, atau mencari kebijakan, pilih **Tambahkan tag baru** untuk menambahkan tag sebagai pasangan nilai kunci. Misalnya, tambahkan tag ke kebijakan Anda dengan pasangan nilai kunci dan**Environment**. **Test**
Untuk informasi selengkapnya tentang penggunaan tag, lihat [Tag untuk AWS Identity and Access Management sumber daya](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_tags.html) di *Panduan Pengguna IAM*.
1. Pilih **Buat kebijakan**.
**Untuk membuat peran IAM yang dapat menulis ke bucket tujuan**
1. Pada panel navigasi konsol IAM, pilih **Peran, lalu pilih **Buat** peran**.
1. Pada halaman **Pilih entitas tepercaya**, lakukan hal berikut:
1. Untuk **jenis entitas Tepercaya**, pilih **Layanan AWS**.
1. Untuk **Service atau use case**, pilih **S3**.
1. Di bawah **Kasus penggunaan**, pilih **S3**.
1. Pilih **Berikutnya**.
1. Pada halaman **Tambahkan izin**, lakukan hal berikut:
1. Di kotak pencarian di bawah **Kebijakan izin**, masukkan nama kebijakan yang sebelumnya Anda buat untuk tutorial ini. Misalnya, **mkc-tutorial-policy**. Kemudian, pilih kotak di sebelah kiri nama kebijakan.
1. (Opsional) Tetapkan [batas izin](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_boundaries.html). Ini adalah fitur lanjutan yang tersedia untuk peran layanan, tetapi bukan peran tertaut layanan. *Untuk informasi tentang menyetel batas izin, lihat [Membuat peran dan melampirkan kebijakan (konsol)](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_job-functions_create-policies.html) di Panduan Pengguna IAM.*
1. Pilih **Berikutnya**.
1. Pada halaman **Nama, tinjau, dan buat**, lakukan hal berikut:
1. Untuk **nama Peran**, masukkan nama deskriptif, seperti**mkc-tutorial-role**.
**penting**
Saat Anda memberi nama peran, perhatikan hal berikut:
Nama peran harus unik di dalam diri Anda Akun AWS, dan tidak dapat dibuat unik berdasarkan kasus.
Misalnya, jangan membuat peran bernama keduanya **PRODROLE** dan**prodrole**. Ketika nama peran digunakan dalam kebijakan atau sebagai bagian dari ARN, nama peran tersebut peka huruf besar/kecil, namun ketika nama peran muncul kepada pelanggan di konsol, seperti selama proses masuk, nama peran tersebut tidak peka huruf besar/kecil.
Anda tidak dapat mengedit nama peran setelah dibuat karena entitas lain mungkin mereferensikan peran tersebut.
1. (Opsional) Untuk **Deskripsi**, masukkan deskripsi untuk peran tersebut.
1. **(Opsional) Untuk mengedit kasus penggunaan dan izin untuk peran, di **Langkah 1: Pilih entitas tepercaya** atau **Langkah 2: Tambahkan bagian izin**, pilih Edit.**
1. (Opsional) Untuk membantu mengidentifikasi, mengatur, atau mencari peran, pilih **Tambahkan tag baru** untuk menambahkan tag sebagai pasangan nilai kunci. Misalnya, tambahkan tag ke peran Anda dengan pasangan nilai kunci dan**ProductManager**. **John**
Untuk informasi selengkapnya tentang penggunaan tag, lihat [Tag untuk AWS Identity and Access Management sumber daya](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_tags.html) di *Panduan Pengguna IAM*.
1. Tinjau peran lalu pilih **Buat peran**.
**Untuk memungkinkan MSK Connect untuk mengambil peran**
1. **Di konsol IAM, di panel kiri, di bawah **Manajemen akses**, pilih Peran.**
1. Temukan `mkc-tutorial-role` dan pilih.
1. Di bawah **Ringkasan** peran, pilih tab **Trust relationship**.
1. Pilih **Edit trust relationship** (Edit Hubungan Kepercayaan).
1. Ganti kebijakan kepercayaan yang ada dengan JSON berikut.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "kafkaconnect.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
1. Pilih **Perbarui Kebijakan Kepercayaan**.
**Untuk membuat titik akhir VPC Amazon dari VPC cluster ke Amazon S3**
1. Buka konsol VPC Amazon di. [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/)
1. Di panel kiri, pilih **Endpoints**.
1. Pilih **Buat titik akhir**.
1. **Di bawah **Nama Layanan** pilih layanan **com.amazonaws.us-east-1.s3** dan tipe Gateway.**
1. Pilih VPC cluster dan kemudian pilih kotak di sebelah kiri tabel rute yang terkait dengan subnet cluster.
1. Pilih **Buat titik akhir**.
**Langkah Selanjutnya**
[Buat plugin kustom](mkc-create-plugin.md)
# Buat plugin kustom
Plugin berisi kode yang mendefinisikan logika konektor. Pada langkah ini Anda membuat plugin khusus yang memiliki kode untuk Lensa Konektor Sink Amazon S3. Pada langkah selanjutnya, saat Anda membuat konektor MSK, Anda menentukan bahwa kodenya ada di plugin khusus ini. Anda dapat menggunakan plugin yang sama untuk membuat beberapa konektor MSK dengan konfigurasi yang berbeda.
**Untuk membuat plugin kustom**
1. Unduh [konektor S3](https://www.confluent.io/hub/confluentinc/kafka-connect-s3).
1. Unggah file ZIP ke bucket S3 yang dapat Anda akses. Untuk informasi tentang cara mengunggah file ke Amazon S3, lihat [Mengunggah objek](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html) di panduan pengguna Amazon S3.
1. Buka konsol MSK Amazon di[https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Di panel kiri perluas **MSK Connect**, lalu pilih Plugin **kustom**.
1. Pilih **Buat plugin kustom**.
1. Pilih **Jelajahi S3**.
1. Dalam daftar bucket, temukan bucket tempat Anda mengunggah file ZIP, dan pilih bucket itu.
1. Dalam daftar objek di ember, pilih tombol radio di sebelah kiri file ZIP, lalu pilih tombol berlabel **Pilih**.
1. Masukkan `mkc-tutorial-plugin` untuk nama plugin kustom, lalu pilih **Buat plugin kustom**.
Mungkin perlu AWS beberapa menit untuk menyelesaikan pembuatan plugin khusus. Ketika proses pembuatan selesai, Anda melihat pesan berikut di spanduk di bagian atas jendela browser.
```
Custom plugin mkc-tutorial-plugin was successfully created
The custom plugin was created. You can now create a connector using this custom plugin.
```
**Langkah Selanjutnya**
[Buat mesin klien dan topik Apache Kafka](mkc-create-topic.md)
# Buat mesin klien dan topik Apache Kafka
Pada langkah ini Anda membuat instans Amazon EC2 untuk digunakan sebagai instans klien Apache Kafka. Anda kemudian menggunakan instance ini untuk membuat topik di cluster.
**Untuk membuat mesin klien**
1. Buka konsol Amazon EC2 di. [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/)
1. Pilih **Luncurkan Instans**.
1. Masukkan **Nama** untuk mesin klien Anda, seperti**mkc-tutorial-client**.
1. Tinggalkan **Amazon Linux 2 AMI (HVM) - Kernel 5.10, Jenis Volume SSD dipilih untuk jenis** **Amazon Machine Image (AMI)**.
1. Pilih jenis **instans t2.xlarge**.
1. Di bawah **Key pair (login)**, pilih **Create a new key pair**. Masukkan **mkc-tutorial-key-pair** **nama pasangan Kunci**, lalu pilih **Unduh Pasangan Kunci**. Alternatifnya, Anda dapat menggunakan pasangan kunci yang sudah ada.
1. Pilih **Luncurkan instans**.
1. Pilih **Lihat Instans.** Kemudian, di kolom **Grup Keamanan**, pilih grup keamanan yang terkait dengan instans baru Anda. Salin ID grup keamanan, dan simpan untuk nanti.
**Untuk memungkinkan klien yang baru dibuat mengirim data ke cluster**
1. Buka konsol VPC Amazon di. [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/)
1. Di panel kiri, di bawah **KEAMANAN**, pilih **Grup Keamanan**. Di kolom **ID grup Keamanan**, cari grup keamanan klaster. Anda menyimpan ID grup keamanan ini saat Anda membuat klaster di[Menyiapkan sumber daya yang diperlukan untuk MSK Connect](mkc-tutorial-setup.md). Pilih grup keamanan ini dengan memilih kotak di sebelah kiri barisnya. Pastikan tidak ada grup keamanan lain yang dipilih secara bersamaan.
1. Di bagian bawah layar, pilih tab **Aturan masuk**.
1. Pilih **Edit aturan masuk**.
1. Di kiri bawah layar, pilih **Tambahkan aturan**.
1. Dalam aturan baru, pilih **Semua lalu lintas** di kolom **Jenis**. Di bidang di sebelah kanan kolom **Sumber**, masukkan ID grup keamanan mesin klien. Ini adalah ID grup keamanan yang Anda simpan setelah Anda membuat mesin klien.
1. Pilih **Simpan aturan**. Cluster MSK Anda sekarang akan menerima semua lalu lintas dari klien yang Anda buat dalam prosedur sebelumnya.
**Cara membuat topik**
1. Buka konsol Amazon EC2 di. [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/)
1. Dalam tabel contoh pilih`mkc-tutorial-client`.
1. Di dekat bagian atas layar, pilih **Connect**, lalu ikuti petunjuk untuk terhubung ke instance.
1. Instal Java pada instance klien dengan menjalankan perintah berikut:
```
sudo yum install java-1.8.0
```
1. Jalankan perintah berikut untuk mengunduh Apache Kafka.
```
wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz
```
**catatan**
Jika Anda ingin menggunakan situs cermin selain yang digunakan dalam perintah ini, Anda dapat memilih yang berbeda di situs web [Apache](https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.1/kafka_2.11-1.1.1.tgz).
1. Jalankan perintah berikut di direktori tempat Anda mengunduh file TAR pada langkah sebelumnya.
```
tar -xzf kafka_2.12-2.2.1.tgz
```
1. Buka direktori **kafka\$12.12-2.2.1**.
1. Buka konsol MSK Amazon di [https://console.aws.amazon.com/msk/rumah? region=us-east-1\$1/home/](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/).
1. Di panel kiri pilih **Cluster**, lalu pilih namanya. `mkc-tutorial-cluster`
1. Pilih **Lihat informasi klien**.
1. Salin string koneksi **Plaintext**.
1. Pilih **Selesai**.
1. Jalankan perintah berikut pada instance klien (`mkc-tutorial-client`), ganti *bootstrapServerString* dengan nilai yang Anda simpan saat melihat informasi klien klaster.
```
/bin/kafka-topics.sh --create --bootstrap-server bootstrapServerString --replication-factor 2 --partitions 1 --topic mkc-tutorial-topic
```
Jika perintah berhasil, Anda melihat pesan berikut: `Created topic mkc-tutorial-topic.`
**Langkah Selanjutnya**
[Buat konektor](mkc-create-connector.md)
# Buat konektor
Prosedur ini menjelaskan cara membuat konektor menggunakan Konsol Manajemen AWS.
**Untuk membuat konektor**
1. Masuk ke Konsol Manajemen AWS, dan buka konsol MSK Amazon di [https://console.aws.amazon.com/msk/rumah? region=us-east-1\$1/home/](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/).
1. **Di panel kiri, perluas **MSK Connect**, lalu pilih Connectors.**
1. Pilih **Buat konektor**.
1. Dalam daftar plugin, pilih`mkc-tutorial-plugin`, lalu pilih **Berikutnya**.
1. Untuk nama konektor masukkan`mkc-tutorial-connector`.
1. Dalam daftar cluster, pilih`mkc-tutorial-cluster`.
1. Di bagian **Pengaturan jaringan konektor**, pilih salah satu dari berikut ini untuk jenis jaringan:
+ **IPv4**(default) - Untuk konektivitas ke tujuan IPv4 hanya
+ **Dual-stack** - Untuk konektivitas ke tujuan melalui keduanya IPv4 dan IPv6 (hanya tersedia jika subnet Anda memiliki IPv4 dan blok IPv6 CIDR yang terkait dengannya)
1. Salin konfigurasi berikut dan tempel ke bidang konfigurasi konektor.
Pastikan Anda mengganti wilayah dengan kode Wilayah AWS tempat Anda membuat konektor. Selain itu, ganti nama bucket Amazon S3 ** dengan nama bucket Anda dalam contoh berikut.
```
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. Di bawah **Izin akses** pilih`mkc-tutorial-role`.
1. Pilih **Berikutnya**. Pada halaman **Keamanan**, pilih **Berikutnya** lagi.
1. Pada halaman **Log** pilih **Berikutnya**.
1. Pada halaman **Tinjau dan buat**, tinjau konfigurasi konektor Anda dan pilih **Buat konektor**.
**Langkah Selanjutnya**
[Kirim data ke cluster MSK](mkc-send-data.md)
# Kirim data ke cluster MSK
Pada langkah ini Anda mengirim data ke topik Apache Kafka yang Anda buat sebelumnya, dan kemudian mencari data yang sama di bucket S3 tujuan.
**Untuk mengirim data ke cluster MSK**
1. Di `bin` folder instalasi Apache Kafka pada instance klien, buat file teks bernama `client.properties` dengan konten berikut.
```
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
```
1. Jalankan perintah berikut untuk membuat produser konsol. Ganti *BootstrapBrokerString* dengan nilai yang Anda peroleh saat menjalankan perintah sebelumnya.
```
/bin/kafka-console-producer.sh --broker-list BootstrapBrokerString --producer.config client.properties --topic mkc-tutorial-topic
```
1. Masukkan pesan apa pun yang Anda inginkan, dan tekan **Enter**. Ulangi langkah ini dua atau tiga kali. Setiap kali Anda memasukkan baris dan tekan **Enter**, baris itu dikirim ke cluster Apache Kafka Anda sebagai pesan terpisah.
1. Lihat di bucket Amazon S3 tujuan untuk menemukan pesan yang Anda kirim pada langkah sebelumnya.
# Memahami konektor
Konektor mengintegrasikan sistem eksternal dan layanan Amazon dengan Apache Kafka dengan terus menyalin data streaming dari sumber data ke cluster Apache Kafka Anda, atau terus menyalin data dari cluster Anda ke dalam data sink. Konektor juga dapat melakukan logika ringan seperti transformasi, konversi format, atau memfilter data sebelum mengirimkan data ke tujuan. Konektor sumber menarik data dari sumber data dan mendorong data ini ke dalam cluster, sementara konektor sink menarik data dari cluster dan mendorong data ini ke dalam sink data.
Diagram berikut menunjukkan arsitektur konektor. Seorang pekerja adalah proses mesin virtual Java (JVM) yang menjalankan logika konektor. Setiap pekerja membuat serangkaian tugas yang berjalan di thread paralel dan melakukan pekerjaan menyalin data. Tugas tidak menyimpan status, dan karenanya dapat dimulai, dihentikan, atau dimulai ulang kapan saja untuk menyediakan pipeline data yang tangguh dan dapat diskalakan.
![\[Diagram yang menunjukkan arsitektur cluster konektor.\]](http://docs.aws.amazon.com/id_id/msk/latest/developerguide/images/mkc-worker-architecture.png)
# Memahami kapasitas konektor
Kapasitas total konektor tergantung pada jumlah pekerja yang dimiliki konektor, serta pada jumlah MSK Connect Units (MCUs) per pekerja. Setiap MCU mewakili 1 vCPU komputasi dan 4 GiB memori. Memori MCU berkaitan dengan memori total instance pekerja dan bukan memori heap yang digunakan.
Pekerja MSK Connect menggunakan alamat IP di subnet yang disediakan pelanggan. Setiap pekerja menggunakan satu alamat IP dari salah satu subnet yang disediakan pelanggan. Anda harus memastikan bahwa Anda memiliki cukup alamat IP yang tersedia di subnet yang disediakan untuk CreateConnector permintaan untuk memperhitungkan kapasitas yang ditentukan, terutama ketika konektor penskalaan otomatis di mana jumlah pekerja dapat berfluktuasi.
Untuk membuat konektor, Anda harus memilih di antara salah satu dari dua mode kapasitas berikut.
+ *Disediakan* - Pilih mode ini jika Anda mengetahui persyaratan kapasitas untuk konektor Anda. Anda menentukan dua nilai:
+ Jumlah pekerja.
+ Jumlah MCUs per pekerja.
+ *Skala otomatis* - Pilih mode ini jika persyaratan kapasitas untuk konektor Anda bervariasi atau jika Anda tidak mengetahuinya sebelumnya. Saat Anda menggunakan mode skala otomatis, Amazon MSK Connect akan mengganti `tasks.max` properti konektor Anda dengan nilai yang sebanding dengan jumlah pekerja yang berjalan di konektor dan jumlah per pekerja. MCUs
Anda menentukan tiga set nilai:
+ Jumlah pekerja minimum dan maksimum.
+ Persentase scale-in dan scale-out untuk pemanfaatan CPU, yang ditentukan oleh metrik. `CpuUtilization` Ketika `CpuUtilization` metrik untuk konektor melebihi persentase scale-out, MSK Connect meningkatkan jumlah pekerja yang berjalan di konektor. Ketika `CpuUtilization` metrik berada di bawah persentase scale-in, MSK Connect mengurangi jumlah pekerja. Jumlah pekerja selalu tetap dalam angka minimum dan maksimum yang Anda tentukan saat Anda membuat konektor.
+ Jumlah MCUs per pekerja.
+ (Opsional) *Jumlah tugas penskalaan otomatis maksimum - Jumlah maksimum tugas* yang dialokasikan ke konektor selama operasi penskalaan otomatis. Parameter ini memungkinkan Anda untuk menetapkan batas atas pada pembuatan tugas, memberikan kontrol yang lebih besar atas pemanfaatan sumber daya dan paralelisme dalam kaitannya dengan partisi topik Kafka Anda.
Untuk informasi selengkapnya tentang pekerja, lihat[Memahami pekerja MSK Connect](msk-connect-workers.md), dan untuk informasi selengkapnya tentang jumlah tugas penskalaan otomatis maksimum, lihat. [Memahami jumlah tugas penskalaan otomatis maksimum](msk-connect-max-autoscaling-task-count.md) Untuk mempelajari metrik MSK Connect, lihat. [Memantau Amazon MSK Connect](mkc-monitoring-overview.md)
# Memahami jumlah tugas penskalaan otomatis maksimum
`maxAutoscalingTaskCount`Parameter ini adalah bidang kapasitas opsional yang tersedia untuk konektor penskalaan otomatis di Amazon MSK Connect. Parameter ini memungkinkan Anda untuk menetapkan batas atas pada jumlah maksimum tugas yang dapat dibuat selama operasi penskalaan otomatis konektor, memberikan kontrol yang lebih besar atas pemanfaatan dan kinerja sumber daya.
Saat Anda menggunakan mode kapasitas berskala otomatis, Amazon MSK Connect secara otomatis mengganti `tasks.max` properti konektor Anda dengan nilai yang sebanding dengan jumlah pekerja dan per pekerja. MCUs `maxAutoscalingTaskCount`Parameter menyediakan opsi tambahan yang dapat dikonfigurasi untuk membatasi jumlah maksimum tugas yang dibuat untuk konektor Anda.
Kemampuan ini sangat berguna ketika Anda ingin mengontrol tingkat paralelisme dalam kaitannya dengan jumlah partisi topik di cluster Kafka Anda. Dengan menetapkan batas ini, Anda dapat mengoptimalkan kinerja dan mencegah distribusi tugas yang tidak efisien yang mungkin terjadi ketika jumlah tugas yang dihitung secara otomatis melebihi persyaratan beban kerja Anda.
## Persyaratan konfigurasi
`maxAutoscalingTaskCount`Parameter harus memenuhi persyaratan berikut:
```
maxAutoscalingTaskCount ≥ maxWorkerCount
```
Persyaratan ini memastikan pemanfaatan sumber daya yang efisien dengan mempertahankan setidaknya satu tugas per pekerja. Sistem memberlakukan minimum ini untuk mengoptimalkan fungsionalitas konektor.
Saat Anda menentukan`maxAutoscalingTaskCount`, batas diterapkan segera setelah pembuatan konektor dan selama semua peristiwa penskalaan berikutnya. Karena jumlah pekerja meningkat atau berkurang selama operasi penskalaan otomatis, sistem terus menghormati batas ini. `tasks.max`Nilai menyesuaikan secara proporsional dengan jumlah pekerja dan MCUs per pekerja tetapi tidak pernah melebihi nilai yang dikonfigurasi`maxAutoscalingTaskCount`.
Jika Anda tidak menentukan parameter ini, konektor menggunakan perhitungan standar tanpa batas: `tasks.max = workerCount × mcuCount × tasksPerMcu` (di tasksPerMcu mana 2).
## Kapan menggunakan maxAutoscalingTask Count
Pertimbangkan untuk menggunakan `maxAutoscalingTaskCount` dalam skenario berikut:
+ *Jumlah partisi terbatas*: Ketika topik Kafka Anda memiliki jumlah partisi tetap yang lebih rendah dari jumlah tugas yang dihitung secara otomatis, menetapkan batas mencegah pembuatan tugas idle tanpa pekerjaan yang harus dilakukan.
+ *Optimalisasi kinerja*: Jika Anda telah mengidentifikasi bahwa jumlah tugas tertentu memberikan throughput optimal untuk beban kerja Anda, Anda dapat membatasi tugas maksimum untuk mempertahankan kinerja yang konsisten.
+ *Manajemen sumber daya*: Ketika Anda ingin mengontrol paralelisme maksimum dan konsumsi sumber daya konektor Anda terlepas dari berapa banyak pekerja yang berjalan.
## Contoh
Untuk konektor dengan konfigurasi berikut:
```
minWorkerCount: 1
maxWorkerCount: 4
mcuCount: 8
maxAutoscalingTaskCount: 15
```
Tanpa`maxAutoscalingTaskCount`, ketika diskalakan ke 4 pekerja, konektor akan membuat 64 tugas (4 pekerja × 8 MCUs × 2 tugas per MCU). Dengan `maxAutoscalingTaskCount` diatur ke 15, konektor hanya membuat 15 tugas, yang mungkin lebih tepat jika topik Kafka Anda memiliki 15 atau lebih sedikit partisi.
# Konfigurasikan tipe jaringan dual-stack
Amazon MSK Connect mendukung tipe jaringan dual-stack untuk konektor baru. Dengan jaringan dual-stack, konektor Anda dapat terhubung ke tujuan melalui keduanya dan IPv4 . IPv6 Perhatikan bahwa IPv6 konektivitas hanya tersedia dalam mode dual-stack (IPv4 \$1 IPv6) - IPv6 -hanya jaringan tidak didukung.
Secara default, konektor baru menggunakan tipe IPv4 jaringan. Untuk membuat konektor dengan tipe jaringan dual-stack, pastikan Anda telah memenuhi prasyarat yang dijelaskan di bagian berikut. Perhatikan bahwa, setelah Anda membuat konektor menggunakan tipe jaringan dual-stack, Anda tidak dapat memodifikasi jenis jaringannya. Untuk mengubah jenis jaringan, Anda harus menghapus dan membuat ulang konektor.
Amazon MSK Connect juga mendukung konektivitas titik akhir API layanan melalui keduanya dan IPv6 . IPv4 Untuk menggunakan IPv6 konektivitas untuk panggilan API, Anda perlu menggunakan titik akhir dual-stack. Untuk informasi selengkapnya tentang titik akhir layanan MSK Connect, lihat titik akhir dan [kuota Amazon MSK Connect](https://docs.aws.amazon.com/general/latest/gr/msk-connect.html).
## Prasyarat untuk menggunakan tipe jaringan dual-stack
Sebelum Anda mengonfigurasi tipe jaringan dual-stack untuk konektor Anda, pastikan bahwa semua subnet yang Anda berikan selama pembuatan konektor memiliki keduanya IPv6 dan blok IPv4 CIDR yang ditetapkan.
## Pertimbangan untuk menggunakan tipe jaringan dual-stack
+ IPv6 dukungan saat ini hanya tersedia dalam mode dual-stack (IPv4 \$1 IPv6), bukan sebagai -only IPv6
+ Konektor dengan dual-stack diaktifkan dapat terhubung melalui keduanya IPv4 dan IPv6 ke MSK dan Sink atau Source sistem data
+ Jenis jaringan tidak dapat dimodifikasi setelah pembuatan konektor - Anda harus menghapus dan membuat ulang konektor untuk mengubah jenis jaringan
+ Semua subnet yang ditentukan selama pembuatan konektor harus mendukung dual-stack agar pembuatan konektor berhasil dengan tipe jaringan dual-stack
+ Jika menggunakan subnet dual-stack tetapi tidak ada tipe jaringan yang ditentukan, konektor akan default ke IPv4 -only untuk kompatibilitas mundur
+ Untuk konektor yang ada, Anda tidak dapat memperbarui jenis jaringan - Anda harus menghapus dan membuat ulang konektor untuk mengubah jenis jaringan
+ Menggunakan jaringan dual-stack tidak menimbulkan biaya tambahan
# Buat konektor
Prosedur ini menjelaskan cara membuat konektor menggunakan Konsol Manajemen AWS.
**Membuat konektor menggunakan Konsol Manajemen AWS**
1. Buka konsol MSK Amazon di[https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. **Di panel kiri, di bawah **MSK Connect**, pilih Konektor.**
1. Pilih **Buat konektor**.
1. Anda dapat memilih antara menggunakan plugin khusus yang ada untuk membuat konektor, atau membuat plugin khusus baru terlebih dahulu. Untuk informasi tentang plugin khusus dan cara membuatnya, lihat[Buat plugin kustom](msk-connect-plugins.md). Dalam prosedur ini, mari kita asumsikan Anda memiliki plugin khusus yang ingin Anda gunakan. Dalam daftar plugin khusus, temukan salah satu yang ingin Anda gunakan, dan pilih kotak di sebelah kirinya, lalu pilih **Berikutnya**.
1. Masukkan nama dan, secara opsional, deskripsi.
1. Pilih cluster yang ingin Anda sambungkan.
1. Di bagian **Pengaturan jaringan konektor**, pilih salah satu dari berikut ini untuk jenis jaringan:
+ **IPv4**(default) - Untuk konektivitas ke tujuan IPv4 hanya
+ **Dual-stack** - Untuk konektivitas ke tujuan melalui keduanya IPv4 dan IPv6 (hanya tersedia jika subnet Anda memiliki IPv4 dan blok IPv6 CIDR yang terkait dengannya)
1. Tentukan konfigurasi konektor. Parameter konfigurasi yang perlu Anda tentukan bergantung pada jenis konektor yang ingin Anda buat. Namun, beberapa parameter umum untuk semua konektor, misalnya, `connector.class` dan `tasks.max` parameter. Berikut ini adalah contoh konfigurasi untuk [Confluent Amazon S3 Sink Connector](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. Selanjutnya, Anda mengonfigurasi kapasitas konektor Anda. Anda dapat memilih di antara dua mode kapasitas: provisioned dan auto scaled. Untuk informasi tentang dua opsi ini, lihat[Memahami kapasitas konektor](msk-connect-capacity.md).
1. (Opsional) Di bagian **Hitungan Tugas Penskalaan Otomatis Maksimum, gunakan bidang Hitung Tugas** Penskalaan Otomatis Maksimum untuk memasukkan jumlah tugas maksimum yang ingin Anda alokasikan ke konektor selama operasi penskalaan otomatis. Nilai harus setidaknya sama dengan jumlah pekerja maksimum Anda. Jika Anda tidak menentukan nilai, konektor menggunakan perhitungan standar tanpa batas apa pun. Untuk informasi selengkapnya, lihat [Memahami jumlah tugas penskalaan otomatis maksimum](msk-connect-max-autoscaling-task-count.md).
1. Pilih konfigurasi pekerja default atau konfigurasi pekerja khusus. Untuk informasi tentang membuat konfigurasi pekerja kustom, lihat[Memahami pekerja MSK Connect](msk-connect-workers.md).
1. Selanjutnya, Anda menentukan peran eksekusi layanan. Ini harus menjadi peran IAM yang dapat diasumsikan MSK Connect, dan yang memberikan konektor semua izin yang diperlukan untuk mengakses sumber daya yang diperlukan. AWS Izin tersebut tergantung pada logika konektor. Untuk informasi tentang cara membuat peran ini, lihat[Memahami peran eksekusi layanan](msk-connect-service-execution-role.md).
1. Pilih **Berikutnya**, tinjau informasi keamanan, lalu pilih **Berikutnya** lagi.
1. Tentukan opsi logging yang Anda inginkan, lalu pilih **Berikutnya**. Untuk informasi tentang pencatatan, lihat[Logging untuk MSK Connect](msk-connect-logging.md).
1. Pada halaman **Tinjau dan buat**, tinjau konfigurasi konektor Anda dan pilih **Buat konektor**.
Untuk menggunakan MSK Connect API untuk membuat konektor, lihat [CreateConnector](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateConnector.html).
Anda dapat menggunakan `UpdateConnector` API untuk memodifikasi konfigurasi konektor. Untuk informasi selengkapnya, lihat [Perbarui konektor](mkc-update-connector.md).
# Perbarui konektor
Prosedur ini menjelaskan cara memperbarui konfigurasi konektor MSK Connect yang ada menggunakan. Konsol Manajemen AWS
**Memperbarui konfigurasi konektor menggunakan Konsol Manajemen AWS**
1. Buka konsol MSK Amazon di[https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. **Di panel kiri, di bawah **MSK Connect**, pilih Konektor.**
1. Pilih konektor existig.
1. Pilih **Edit konfigurasi konektor**.
1. Perbarui konfigurasi konektor. Anda tidak dapat mengganti `connector.class` menggunakan UpdateConnector. Contoh berikut menunjukkan konfigurasi contoh untuk konektor 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. Pilih **Kirim**.
1. Anda kemudian dapat memantau keadaan operasi saat ini di tab **Operasi** konektor.
Untuk menggunakan MSK Connect API untuk memperbarui konfigurasi konektor, lihat [UpdateConnector](https://docs.aws.amazon.com/MSKC/latest/mskc/API_UpdateConnector.html).
# Menghubungkan dari konektor
Praktik terbaik berikut dapat meningkatkan kinerja konektivitas Anda ke Amazon MSK Connect.
## Jangan tumpang tindih IPs untuk peering Amazon VPC atau Transit Gateway
Jika Anda menggunakan peering VPC Amazon atau Transit Gateway dengan Amazon MSK Connect, jangan konfigurasikan konektor Anda untuk menjangkau sumber daya VPC peered dengan rentang CIDR: IPs
+ “10.99.0.0/16"
+ “192.168.0.0/16"
+ “172.21.0.0/16"
# Buat plugin kustom
Plugin adalah AWS sumber daya yang berisi kode yang mendefinisikan logika konektor Anda. Anda mengunggah file JAR (atau file ZIP yang berisi satu atau beberapa file JAR) ke bucket S3, dan tentukan lokasi bucket saat Anda membuat plugin. Ketika plugin dibuat, MSK Connect menyalin konten objek S3 pada saat itu. Itu tidak mempertahankan tautan ke objek S3, sehingga modifikasi selanjutnya pada objek tidak akan mempengaruhi plugin atau konektornya. Saat Anda membuat konektor, Anda menentukan plugin yang ingin digunakan MSK Connect untuk itu. Hubungan plugin dengan konektor adalah one-to-many: Anda dapat membuat satu atau lebih konektor dari plugin yang sama.
**catatan**
Plugin khusus tidak dapat diperbarui di tempat. Untuk menggunakan versi baru kode plugin Anda, hapus semua konektor yang mereferensikan plugin, hapus plugin, lalu buat ulang.
**Kemasan ketergantungan untuk plugin khusus**
Kami menyarankan Anda menyertakan semua file JAR yang diperlukan dan dependensi untuk plugin Anda. Package konektor Anda sebagai salah satu dari berikut ini:
File ZIP yang berisi semua file JAR yang diperlukan dan dependensi untuk plugin.
Uber JAR tunggal yang berisi semua file kelas untuk plugin dan dependensinya.
Tidak menggabungkan dependensi plugin Anda dapat memengaruhi ketersediaan atau kompatibilitas di lingkungan runtime dan menyebabkan kesalahan yang tidak terduga.
Untuk informasi tentang cara mengembangkan kode untuk konektor, lihat [Panduan Pengembangan Konektor di dokumentasi](https://kafka.apache.org/documentation/#connect_development) Apache Kafka.
**Membuat plugin khusus menggunakan Konsol Manajemen AWS**
1. Buka konsol MSK Amazon di[https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Di panel kiri, di bawah **MSK Connect**, pilih Plugin **kustom**.
1. Pilih **Buat plugin kustom**.
1. Pilih **Jelajahi S3**.
1. Dalam daftar bucket S3, pilih bucket yang memiliki file JAR atau ZIP untuk plugin.
1. Dalam daftar objek, pilih kotak di sebelah kiri file JAR atau ZIP untuk plugin, lalu pilih **Pilih**.
1. Pilih **Buat plugin kustom**.
Untuk menggunakan MSK Connect API untuk membuat plugin khusus, lihat [CreateCustomPlugin](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateCustomPlugin.html).
# Memahami pekerja MSK Connect
Seorang pekerja adalah proses mesin virtual Java (JVM) yang menjalankan logika konektor. Setiap pekerja membuat serangkaian tugas yang berjalan di thread paralel dan melakukan pekerjaan menyalin data. Tugas tidak menyimpan status, dan karenanya dapat dimulai, dihentikan, atau dimulai ulang kapan saja untuk menyediakan pipeline data yang tangguh dan dapat diskalakan. Perubahan jumlah pekerja, baik karena peristiwa penskalaan atau karena kegagalan yang tidak terduga, secara otomatis terdeteksi oleh pekerja yang tersisa. Mereka berkoordinasi untuk menyeimbangkan kembali tugas di seluruh set pekerja yang tersisa. Connect workers menggunakan kelompok konsumen Apache Kafka untuk berkoordinasi dan menyeimbangkan kembali.
Jika persyaratan kapasitas konektor Anda bervariasi atau sulit diperkirakan, Anda dapat membiarkan MSK Connect menskalakan jumlah pekerja sesuai kebutuhan antara batas bawah dan batas atas yang Anda tentukan. Atau, Anda dapat menentukan jumlah pasti pekerja yang ingin Anda jalankan logika konektor Anda. Untuk informasi selengkapnya, lihat [Memahami kapasitas konektor](msk-connect-capacity.md).
**Pekerja MSK Connect menggunakan alamat IP**
Pekerja MSK Connect menggunakan alamat IP di subnet yang disediakan pelanggan. Setiap pekerja menggunakan satu alamat IP dari salah satu subnet yang disediakan pelanggan. Anda harus memastikan bahwa Anda memiliki cukup alamat IP yang tersedia di subnet yang disediakan untuk CreateConnector permintaan untuk memperhitungkan kapasitas yang ditentukan, terutama ketika konektor penskalaan otomatis di mana jumlah pekerja dapat berfluktuasi.
## Konfigurasi pekerja default
MSK Connect menyediakan konfigurasi pekerja default berikut:
```
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
```
# Properti konfigurasi pekerja yang didukung
MSK Connect menyediakan konfigurasi pekerja default. Anda juga memiliki opsi untuk membuat konfigurasi pekerja khusus untuk digunakan dengan konektor Anda. Daftar berikut mencakup informasi tentang properti konfigurasi pekerja yang didukung atau tidak didukung Amazon MSK Connect.
+ `value.converter`Properti `key.converter` dan diperlukan.
+ MSK Connect mendukung properti `producer.` konfigurasi berikut.
```
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 mendukung properti `consumer.` konfigurasi berikut.
```
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
```
+ Semua properti konfigurasi lain yang tidak dimulai dengan `consumer.` awalan `producer.` atau didukung *kecuali* untuk properti berikut.
```
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
```
Untuk informasi selengkapnya tentang properti konfigurasi pekerja dan apa yang diwakilinya, lihat Konfigurasi [Kafka Connect di dokumentasi](https://kafka.apache.org/documentation/#connectconfigs) Apache Kafka.
# Buat konfigurasi pekerja khusus
Prosedur ini menjelaskan cara membuat konfigurasi pekerja khusus menggunakan Konsol Manajemen AWS.
**Membuat konfigurasi pekerja khusus menggunakan Konsol Manajemen AWS**
1. Buka konsol MSK Amazon di[https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/).
1. Di panel kiri, di bawah **MSK Connect**, pilih konfigurasi **Worker**.
1. Pilih **Buat konfigurasi pekerja**.
1. Masukkan nama dan deskripsi opsional, lalu tambahkan properti dan nilai yang ingin Anda atur.
1. Pilih **Buat konfigurasi pekerja**.
Untuk menggunakan MSK Connect API untuk membuat konfigurasi pekerja, lihat [CreateWorkerConfiguration](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateWorkerConfiguration.html).
# Kelola offset konektor sumber menggunakan `offset.storage.topic`
Bagian ini memberikan informasi untuk membantu Anda mengelola offset konektor sumber menggunakan topik *penyimpanan offset*. Topik penyimpanan offset adalah topik internal yang digunakan Kafka Connect untuk menyimpan konektor dan offset konfigurasi tugas.
## Pertimbangan-pertimbangan
Pertimbangkan hal berikut ketika Anda mengelola offset konektor sumber.
+ Untuk menentukan topik penyimpanan offset, berikan nama topik Kafka tempat offset konektor disimpan sebagai nilai untuk konfigurasi pekerja `offset.storage.topic` Anda.
+ Berhati-hatilah saat Anda membuat perubahan pada konfigurasi konektor. Mengubah nilai konfigurasi dapat mengakibatkan perilaku konektor yang tidak diinginkan jika konektor sumber menggunakan nilai dari konfigurasi ke catatan offset kunci. Kami menyarankan Anda merujuk ke dokumentasi plugin Anda untuk panduan.
+ **Sesuaikan jumlah partisi default** — Selain menyesuaikan konfigurasi pekerja dengan menambahkan`offset.storage.topic`, Anda dapat menyesuaikan jumlah partisi untuk topik penyimpanan offset dan status. Partisi default untuk topik internal adalah sebagai berikut.
+ `config.storage.topic`: 1, tidak dapat dikonfigurasi, harus topik partisi tunggal
+ `offset.storage.topic`: 25, dapat dikonfigurasi dengan menyediakan `offset.storage.partitions`
+ `status.storage.topic`: 5, dapat dikonfigurasi dengan menyediakan `status.storage.partitions`
+ **Menghapus topik secara manual** - Amazon MSK Connect membuat topik internal Kafka connect baru (nama topik dimulai dengan`__amazon_msk_connect`) pada setiap penyebaran konektor. Topik lama yang dilampirkan ke konektor yang dihapus tidak dihapus secara otomatis karena topik internal, seperti`offset.storage.topic`, dapat digunakan kembali di antara konektor. Namun, Anda dapat secara manual menghapus topik internal yang tidak digunakan yang dibuat oleh MSK Connect. Topik internal diberi nama mengikuti format`__amazon_msk_connect__connector_name_connector_id`.
Ekspresi reguler `__amazon_msk_connect__connector_name_connector_id` dapat digunakan untuk menghapus topik internal. Anda tidak boleh menghapus topik internal yang saat ini digunakan oleh konektor yang sedang berjalan.
+ **Menggunakan nama yang sama untuk topik internal yang dibuat oleh MSK Connect** — Jika Anda ingin menggunakan kembali topik penyimpanan offset untuk menggunakan offset dari konektor yang dibuat sebelumnya, Anda harus memberikan konektor baru nama yang sama dengan konektor lama. `offset.storage.topic`Properti dapat diatur menggunakan konfigurasi pekerja untuk menetapkan nama yang sama ke `offset.storage.topic` dan digunakan kembali di antara konektor yang berbeda. Konfigurasi ini dijelaskan dalam [Mengelola offset konektor](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-create-custom-worker-config). MSK Connect tidak mengizinkan konektor yang berbeda untuk berbagi `config.storage.topic` dan`status.storage.topic`. Topik-topik tersebut dibuat setiap kali Anda membuat konektor baru di MSKC. Mereka secara otomatis dinamai mengikuti format`__amazon_msk_connect__connector_name_connector_id`, dan begitu juga berbeda di berbagai konektor yang Anda buat.
# Gunakan topik penyimpanan offset default
Secara default, Amazon MSK Connect menghasilkan topik penyimpanan offset baru di cluster Kafka Anda untuk setiap konektor yang Anda buat. MSK membangun nama topik default menggunakan bagian dari konektor ARN. Misalnya, `__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2`.
# Gunakan topik penyimpanan offset khusus
Untuk memberikan kontinuitas offset antara konektor sumber, Anda dapat menggunakan topik penyimpanan offset pilihan Anda alih-alih topik default. Menentukan topik penyimpanan offset membantu Anda menyelesaikan tugas seperti membuat konektor sumber yang melanjutkan pembacaan dari offset terakhir konektor sebelumnya.
Untuk menentukan topik penyimpanan offset, Anda memberikan nilai untuk `offset.storage.topic` properti dalam konfigurasi pekerja sebelum membuat konektor. Jika Anda ingin menggunakan kembali topik penyimpanan offset untuk menggunakan offset dari konektor yang dibuat sebelumnya, Anda harus memberi konektor baru nama yang sama dengan konektor lama. Jika Anda membuat topik penyimpanan offset kustom, Anda harus mengatur [https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy](https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy)ke `compact` dalam konfigurasi topik Anda.
**catatan**
Jika Anda menentukan topik penyimpanan offset saat membuat konektor *sink*, MSK Connect akan membuat topik jika belum ada. Namun, topik tersebut tidak akan digunakan untuk menyimpan offset konektor.
Offset konektor sink malah dikelola menggunakan protokol grup konsumen Kafka. Setiap konektor wastafel membuat grup bernama`connect-{CONNECTOR_NAME}`. Selama grup konsumen ada, konektor wastafel berturut-turut yang Anda buat dengan `CONNECTOR_NAME` nilai yang sama akan berlanjut dari offset komitmen terakhir.
**penting**
Jika Anda ingin memperbarui konfigurasi konektor yang ada sambil mempertahankan kontinuitas offset, gunakan API. UpdateConnector Untuk informasi selengkapnya, lihat [Perbarui konektor](mkc-update-connector.md).
**Example : Menentukan topik penyimpanan offset saat membuat ulang konektor sumber**
Jika Anda perlu menghapus dan membuat ulang konektor sambil mempertahankan kontinuitas offset, Anda dapat menentukan topik penyimpanan offset dalam konfigurasi pekerja Anda. Misalnya, Anda memiliki konektor change data capture (CDC) dan Anda ingin membuatnya kembali tanpa kehilangan tempat Anda di aliran CDC. Langkah-langkah berikut menunjukkan bagaimana menyelesaikan tugas ini.
1. Pada mesin klien Anda, jalankan perintah berikut untuk menemukan nama topik penyimpanan offset konektor Anda. Ganti `` dengan string broker bootstrap cluster Anda. Untuk petunjuk tentang mendapatkan string broker bootstrap Anda, lihat[Dapatkan broker bootstrap untuk cluster MSK Amazon](msk-get-bootstrap-brokers.md).
```
/bin/kafka-topics.sh --list --bootstrap-server
```
Output berikut menunjukkan daftar semua topik cluster, termasuk topik konektor internal default. Dalam contoh ini, konektor CDC yang ada menggunakan [topik penyimpanan offset default](msk-connect-default-offset-storage-topic.md) yang dibuat oleh MSK Connect. Inilah sebabnya mengapa topik penyimpanan offset disebut`__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. Buka konsol MSK Amazon di [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk).
1. Pilih konektor Anda dari daftar **Konektor**. Salin dan simpan konten bidang **konfigurasi Konektor** sehingga Anda dapat memodifikasinya dan menggunakannya untuk membuat konektor baru.
1. Pilih **Hapus** untuk menghapus konektor. Kemudian masukkan nama konektor di bidang input teks untuk mengonfirmasi penghapusan.
1. Buat konfigurasi pekerja khusus dengan nilai yang sesuai dengan skenario Anda. Untuk petunjuk, lihat [Buat konfigurasi pekerja khusus](msk-connect-create-custom-worker-config.md).
Dalam konfigurasi pekerja Anda, Anda harus menentukan nama topik penyimpanan offset yang sebelumnya Anda ambil sebagai nilai untuk `offset.storage.topic` like dalam konfigurasi berikut.
```
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.
**penting**
Anda harus memberikan konektor baru Anda nama yang sama dengan konektor lama.
Buat konektor baru menggunakan konfigurasi pekerja yang Anda atur di langkah sebelumnya. Untuk petunjuk, lihat [Buat konektor](mkc-create-connector-intro.md).
# Tutorial: Eksternalisasi informasi sensitif menggunakan penyedia konfigurasi
Contoh ini menunjukkan cara mengeksternalisasi informasi sensitif untuk Amazon MSK Connect menggunakan penyedia konfigurasi open source. Penyedia konfigurasi memungkinkan Anda menentukan variabel alih-alih teks biasa dalam konfigurasi konektor atau pekerja, dan pekerja yang berjalan di konektor Anda menyelesaikan variabel ini saat runtime. Ini mencegah kredensi dan rahasia lainnya disimpan dalam teks biasa. Penyedia konfigurasi dalam contoh mendukung pengambilan parameter konfigurasi dari AWS Secrets Manager, Amazon S3 dan Systems Manager (SSM). Pada [Langkah 2](#msk-connect-config-providers), Anda dapat melihat cara mengatur penyimpanan dan pengambilan informasi sensitif untuk layanan yang ingin Anda konfigurasi.
## Pertimbangan-pertimbangan
Pertimbangkan hal berikut saat menggunakan penyedia konfigurasi MSK dengan Amazon MSK Connect:
+ Tetapkan izin yang sesuai saat menggunakan penyedia konfigurasi ke Peran Eksekusi Layanan IAM.
+ Tentukan penyedia konfigurasi dalam konfigurasi pekerja dan implementasinya dalam konfigurasi konektor.
+ Nilai konfigurasi sensitif dapat muncul di log konektor jika plugin tidak mendefinisikan nilai-nilai tersebut sebagai rahasia. Kafka Connect memperlakukan nilai konfigurasi yang tidak ditentukan sama dengan nilai plaintext lainnya. Untuk mempelajari selengkapnya, lihat [Mencegah rahasia muncul di log konektor](msk-connect-logging.md#msk-connect-logging-secrets).
+ Secara default, MSK Connect sering me-restart konektor saat konektor menggunakan penyedia konfigurasi. Untuk menonaktifkan perilaku restart ini, Anda dapat mengatur `config.action.reload` nilai ke `none` dalam konfigurasi konektor Anda.
## Buat plugin khusus dan unggah ke S3
Untuk membuat plugin khusus, buat file zip yang berisi konektor dan msk-config-provider dengan menjalankan perintah berikut di mesin lokal Anda.
**Untuk membuat plugin kustom menggunakan jendela terminal dan Debezium sebagai konektor**
Gunakan AWS CLI untuk menjalankan perintah sebagai pengguna super dengan kredenal yang memungkinkan Anda mengakses bucket S3 Anda. AWS *Untuk informasi tentang menginstal dan menyiapkan AWS CLI, lihat [Memulai AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html) di Panduan Pengguna.AWS Command Line Interface * *Untuk informasi tentang penggunaan AWS CLI dengan Amazon S3, lihat Menggunakan [Amazon S3 dengan AWS CLI di Panduan Pengguna](https://docs.aws.amazon.com/cli/latest/userguide/cli-services-s3.html).AWS Command Line Interface *
1. Di jendela terminal, buat folder bernama `custom-plugin` di ruang kerja Anda menggunakan perintah berikut.
```
mkdir custom-plugin && cd custom-plugin
```
1. Unduh rilis stabil terbaru dari **MySQL Connector** Plug-in dari situs [Debezium menggunakan](https://debezium.io/releases/) perintah berikut.
```
wget https://repo1.maven.org/maven2/io/debezium/debezium-connectormysql/
2.2.0.Final/debezium-connector-mysql-2.2.0.Final-plugin.tar.gz
```
Ekstrak file gzip yang diunduh di `custom-plugin` folder menggunakan perintah berikut.
```
tar xzf debezium-connector-mysql-2.2.0.Final-plugin.tar.gz
```
1. Unduh [file zip penyedia konfigurasi MSK](https://github.com/aws-samples/msk-config-providers/releases/download/r0.4.0/msk-config-providers-0.4.0-with-dependencies.zip) menggunakan perintah berikut.
```
wget https://github.com/aws-samples/msk-config-providers/releases/download/r0.4.0/msk-config-providers-0.4.0-with-dependencies.zip
```
Ekstrak file zip yang diunduh di `custom-plugin` folder menggunakan perintah follwoing.
```
unzip msk-config-providers-0.4.0-with-dependencies.zip
```
1. Zip konten penyedia konfigurasi MSK dari langkah di atas dan konektor khusus ke dalam satu file bernama. `custom-plugin.zip`
```
zip -r ../custom-plugin.zip *
```
1. Unggah file ke S3 untuk direferensikan nanti.
```
aws s3 cp ../custom-plugin.zip s3:
```
1. Di konsol MSK Amazon, di bawah bagian **MSK Connect**, pilih **Plugin Kustom, lalu pilih Buat plugin** **khusus** dan telusuri ember **s3: < *S3\$1URI\$1BUCKET\$1LOCATION* >** S3 untuk memilih file ZIP plugin khusus yang baru saja Anda unggah.
![\[Amazon S3 bucket interface showing a single custom-plugin.zip file in the debezium folder.\]](http://docs.aws.amazon.com/id_id/msk/latest/developerguide/images/s3-object-browser.png)
1. Masukkan **debezium-custom-plugin** untuk nama plugin. Secara opsional, masukkan deskripsi dan pilih **Buat Plugin Kustom**.
![\[Amazon S3 bucket interface showing a single custom-plugin.zip file in the debezium folder.\]](http://docs.aws.amazon.com/id_id/msk/latest/developerguide/images/create-custom-plugin.png)
## Konfigurasikan parameter dan izin untuk penyedia yang berbeda
Anda dapat mengonfigurasi nilai parameter dalam tiga layanan ini:
+ Secrets Manager
+ Penyimpanan Parameter Systems Manager
+ S3 - Layanan Penyimpanan Sederhana
Pilih salah satu tab di bawah ini untuk petunjuk tentang pengaturan parameter dan izin yang relevan untuk layanan tersebut.
------
#### [ Configure in Secrets Manager ]
**Untuk mengkonfigurasi nilai parameter di Secrets Manager**
1. Buka [konsol Secrets Manager](https://console.aws.amazon.com/secretsmanager/).
1. Buat rahasia baru untuk menyimpan kredensi atau rahasia Anda. Untuk petunjuk, lihat [Membuat AWS Secrets Manager rahasia](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html) di *Panduan AWS Secrets Manager Pengguna*.
1. Salin ARN rahasia Anda.
1. Tambahkan izin Secrets Manager dari kebijakan contoh berikut ke [peran eksekusi Layanan](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html) Anda. Ganti contoh ARN,`arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234`, dengan ARN rahasia Anda.
1. Tambahkan konfigurasi pekerja dan instruksi konektor.
****
```
{
"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. Untuk menggunakan penyedia konfigurasi Secrets Manager, salin baris kode berikut ke kotak teks konfigurasi pekerja di Langkah 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. Untuk penyedia konfigurasi manajer rahasia, salin baris kode berikut dalam konfigurasi konektor di Langkah 4.
```
#Example implementation for secrets manager variable
database.user=${secretsmanager:MSKAuroraDBCredentials:username}
database.password=${secretsmanager:MSKAuroraDBCredentials:password}
```
Anda juga dapat menggunakan langkah di atas dengan lebih banyak penyedia konfigurasi.
------
#### [ Configure in Systems Manager Parameter Store ]
**Untuk mengkonfigurasi nilai parameter di Systems Manager Parameter Store**
1. Buka [konsol System Manager](https://console.aws.amazon.com/systems-manager/).
1. Di panel navigasi, pilih **Penyimpanan Parameter**.
1. Buat parameter baru untuk disimpan di Systems Manager. Untuk petunjuknya, lihat [Membuat parameter Systems Manager (konsol)](https://docs.aws.amazon.com/systems-manager/latest/userguide/parameter-create-console.html) di Panduan AWS Systems Manager Pengguna.
1. Salin ARN parameter Anda.
1. Tambahkan izin Systems Manager dari kebijakan contoh berikut ke [peran eksekusi Layanan](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html) Anda. Ganti ** dengan ARN parameter Anda.
****
```
{
"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. Untuk menggunakan penyedia konfigurasi penyimpanan parameter, salin baris kode berikut ke kotak teks konfigurasi pekerja di Langkah 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. Untuk penyedia konfigurasi penyimpanan parameter, salin baris kode berikut dalam konfigurasi konektor di Langkah 5.
```
#Example implementation for parameter store variable
schema.history.internal.kafka.bootstrap.servers=${ssm::MSKBootstrapServerAddress}
```
Anda juga dapat menggabungkan dua langkah di atas dengan lebih banyak penyedia konfigurasi.
------
#### [ Configure in Amazon S3 ]
**Untuk mengkonfigurasi objects/files di Amazon S3**
1. Buka [konsol Amazon S3](https://console.aws.amazon.com/s3/).
1. Unggah objek Anda ke ember di S3. Untuk petunjuk, lihat [Mengunggah objek](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html).
1. Salin ARN objek Anda.
1. Tambahkan izin Baca Objek Amazon S3 dari kebijakan contoh berikut ke peran eksekusi [Layanan](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html) Anda. Ganti contoh ARN,`arn:aws:s3:::`, dengan ARN objek Anda.
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": "s3:GetObject",
"Resource": "arn:aws:s3:::"
}
]
}
```
1. Untuk menggunakan penyedia konfigurasi Amazon S3, salin baris kode berikut ke kotak teks konfigurasi pekerja di Langkah 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. Untuk penyedia konfigurasi Amazon S3, salin baris kode berikut ke konfigurasi konektor di Langkah 4.
```
#Example implementation for S3 object
database.ssl.truststore.location = ${s3import:us-west-2:my_cert_bucket/path/to/trustore_unique_filename.jks}
```
Anda juga dapat menggabungkan dua langkah di atas dengan lebih banyak penyedia konfigurasi.
------
## Buat konfigurasi pekerja khusus dengan informasi tentang penyedia konfigurasi Anda
1. Pilih **Konfigurasi pekerja** di bawah bagian **Amazon MSK Connect**.
1. Pilih **Buat konfigurasi pekerja**.
1. Masukkan `SourceDebeziumCustomConfig` di kotak teks Nama Konfigurasi Pekerja. Deskripsi adalah opsional.
1. Salin kode konfigurasi yang relevan berdasarkan penyedia yang diinginkan, dan tempelkan di kotak teks **konfigurasi Pekerja**.
1. Ini adalah contoh konfigurasi pekerja untuk ketiga penyedia:
```
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. Klik Buat konfigurasi pekerja.
## Buat konektor
1. Buat konektor baru menggunakan instruksi di [Buat konektor baru](https://docs.aws.amazon.com/msk/latest/developerguide/mkc-create-connector.html).
1. Pilih `custom-plugin.zip` file yang Anda unggah ke bucket S3 Anda [Buat plugin khusus dan unggah ke S3](#msk-connect-config-providers-create-custom-plugin) sebagai sumber untuk plugin khusus.
1. Salin kode konfigurasi yang relevan berdasarkan penyedia yang diinginkan, dan tempel di bidang konfigurasi Konektor.
1. Ini adalah contoh untuk konfigurasi konektor untuk ketiga penyedia:
```
#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. Pilih **Gunakan konfigurasi khusus** dan pilih **SourceDebeziumCustomConfig**dari tarik-turun **Konfigurasi Pekerja**.
1. Ikuti langkah-langkah yang tersisa dari instruksi di [Buat konektor](https://docs.aws.amazon.com/msk/latest/developerguide/mkc-create-connector.html).
# Peran dan kebijakan IAM untuk MSK Connect
Bagian ini membantu Anda menyiapkan kebijakan dan peran IAM yang sesuai untuk menerapkan dan mengelola Amazon MSK Connect dengan aman di lingkungan Anda. AWS Bagian berikut menjelaskan peran eksekusi layanan yang harus digunakan dengan MSK Connect, termasuk kebijakan kepercayaan yang diperlukan dan izin tambahan yang diperlukan saat menghubungkan ke kluster MSK yang diautentikasi oleh IAM. Halaman ini juga menyediakan contoh kebijakan IAM komprehensif untuk memberikan akses penuh ke fungsionalitas MSK Connect, serta detail tentang kebijakan AWS terkelola yang tersedia untuk layanan.
**Topics**
+ [Memahami peran eksekusi layanan](msk-connect-service-execution-role.md)
+ [Contoh kebijakan IAM untuk MSK Connect](mkc-iam-policy-examples.md)
+ [Mencegah masalah wakil lintas layanan yang membingungkan](cross-service-confused-deputy-prevention.md)
+ [AWS kebijakan terkelola untuk MSK Connect](mkc-security-iam-awsmanpol.md)
+ [Menggunakan peran terkait layanan untuk MSK Connect](mkc-using-service-linked-roles.md)
# Memahami peran eksekusi layanan
**catatan**
Amazon MSK Connect tidak mendukung penggunaan peran [Service-linked sebagai peran](mkc-using-service-linked-roles.md) eksekusi layanan. Anda harus membuat peran eksekusi layanan terpisah. Untuk petunjuk tentang cara membuat peran IAM kustom, lihat [Membuat peran untuk mendelegasikan izin ke AWS layanan di Panduan](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-service.html) Pengguna *IAM*.
Saat membuat konektor dengan MSK Connect, Anda harus menentukan peran AWS Identity and Access Management (IAM) yang akan digunakan dengannya. Peran eksekusi layanan Anda harus memiliki kebijakan kepercayaan berikut sehingga MSK Connect dapat menerapkannya. Untuk informasi tentang kunci konteks kondisi dalam kebijakan ini, lihat[Mencegah masalah wakil lintas layanan yang membingungkan](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"
}
}
}
]
}
```
------
Jika klaster MSK Amazon yang ingin Anda gunakan dengan konektor adalah klaster yang menggunakan autentikasi IAM, maka Anda harus menambahkan kebijakan izin berikut ke peran eksekusi layanan konektor. Untuk informasi tentang cara menemukan UUID klaster Anda dan cara membuat topik ARNs, lihat. [Sumber daya kebijakan otorisasi](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-*"
]
}
]
}
```
------
Bergantung pada jenis konektornya, Anda mungkin juga perlu melampirkan kebijakan izin ke peran eksekusi layanan yang memungkinkannya mengakses AWS sumber daya. Misalnya, jika konektor Anda perlu mengirim data ke bucket S3, maka peran eksekusi layanan harus memiliki kebijakan izin yang memberikan izin untuk menulis ke bucket tersebut. Untuk tujuan pengujian, Anda dapat menggunakan salah satu kebijakan IAM pra-bangun yang memberikan akses penuh, seperti. `arn:aws:iam::aws:policy/AmazonS3FullAccess` Namun, untuk tujuan keamanan, kami menyarankan Anda menggunakan kebijakan paling ketat yang memungkinkan konektor Anda membaca dari AWS sumber atau menulis ke AWS wastafel.
# Contoh kebijakan IAM untuk MSK Connect
Untuk memberi pengguna non-admin akses penuh ke semua fungsionalitas MSK Connect, lampirkan kebijakan seperti berikut ini ke peran IAM pengguna.
------
#### [ 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/*"
]
}
]
}
```
------
# Mencegah masalah wakil lintas layanan yang membingungkan
Masalah "confused deputy" adalah masalah keamanan di mana entitas yang tidak memiliki izin untuk melakukan tindakan dapat memengaruhi entitas yang memiliki hak akses lebih tinggi untuk melakukan tindakan. Pada tahun AWS, peniruan lintas layanan dapat mengakibatkan masalah wakil yang membingungkan. Peniruan identitas lintas layanan dapat terjadi ketika satu layanan (*layanan yang dipanggil*) memanggil layanan lain (*layanan yang dipanggil*). Layanan pemanggilan dapat dimanipulasi menggunakan izinnya untuk bertindak pada sumber daya pelanggan lain dengan cara yang seharusnya tidak dilakukannya kecuali bila memiliki izin untuk mengakses. Untuk mencegah hal ini, AWS menyediakan alat yang membantu Anda melindungi data untuk semua layanan dengan principal layanan yang telah diberi akses ke sumber daya di akun Anda.
Sebaiknya gunakan kunci konteks kondisi [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)global [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)dan global dalam kebijakan sumber daya untuk membatasi izin yang diberikan MSK Connect kepada layanan lain ke sumber daya. Jika `aws:SourceArn` nilai tidak berisi ID akun (misalnya, ARN bucket Amazon S3 tidak berisi ID akun), Anda harus menggunakan kedua kunci konteks kondisi global untuk membatasi izin. Jika Anda menggunakan kunci konteks kondisi global dan nilai `aws:SourceArn` berisi ID akun, nilai `aws:SourceAccount` dan akun dalam nilai `aws:SourceArn` harus menggunakan ID akun yang sama saat digunakan dalam pernyataan kebijakan yang sama. Gunakan `aws:SourceArn` jika Anda ingin hanya satu sumber daya yang akan dikaitkan dengan akses lintas layanan. Gunakan `aws:SourceAccount` jika Anda ingin mengizinkan sumber daya apa pun di akun tersebut dikaitkan dengan penggunaan lintas layanan.
Dalam kasus MSK Connect, nilai `aws:SourceArn` harus berupa konektor MSK.
Cara paling efektif untuk melindungi dari masalah "confused deputy" adalah dengan menggunakan kunci konteks kondisi global `aws:SourceArn` dengan ARN lengkap sumber daya. Jika Anda tidak mengetahui ARN lengkap sumber daya atau jika Anda menentukan beberapa sumber daya, gunakan kunci kondisi konteks `aws:SourceArn` global dengan wildcard (`*`) untuk bagian ARN yang tidak diketahui. Misalnya, *arn:aws:kafkaconnect:us-east-1:123456789012:connector/\$1* mewakili semua konektor yang termasuk dalam akun dengan ID 123456789012 di Wilayah AS Timur (Virginia N.).
Contoh berikut menunjukkan bagaimana Anda dapat menggunakan `aws:SourceArn` dan kunci konteks kondisi `aws:SourceAccount` global di MSK Connect untuk mencegah masalah wakil bingung. Ganti *123456789012* dan arn:aws:kafkaconnect: ::connector//dengan informasi Anda dan *us-east-1* konektor*123456789012*. *my-S3-Sink-Connector* *abcd1234-5678-90ab-cdef-1234567890ab* Akun 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 kebijakan terkelola untuk MSK Connect
Kebijakan AWS terkelola adalah kebijakan mandiri yang dibuat dan dikelola oleh AWS. AWS Kebijakan terkelola dirancang untuk memberikan izin bagi banyak kasus penggunaan umum sehingga Anda dapat mulai menetapkan izin kepada pengguna, grup, dan peran.
Perlu diingat bahwa kebijakan AWS terkelola mungkin tidak memberikan izin hak istimewa paling sedikit untuk kasus penggunaan spesifik Anda karena tersedia untuk digunakan semua pelanggan. AWS Kami menyarankan Anda untuk mengurangi izin lebih lanjut dengan menentukan [kebijakan yang dikelola pelanggan](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#customer-managed-policies) yang khusus untuk kasus penggunaan Anda.
Anda tidak dapat mengubah izin yang ditentukan dalam kebijakan AWS terkelola. Jika AWS memperbarui izin yang ditentukan dalam kebijakan AWS terkelola, pembaruan akan memengaruhi semua identitas utama (pengguna, grup, dan peran) yang dilampirkan kebijakan tersebut. AWS kemungkinan besar akan memperbarui kebijakan AWS terkelola saat baru Layanan AWS diluncurkan atau operasi API baru tersedia untuk layanan yang ada.
Untuk informasi selengkapnya, lihat [Kebijakan terkelola AWS](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#aws-managed-policies) dalam *Panduan Pengguna IAM*.
## AWS kebijakan terkelola: Amazon MSKConnect ReadOnlyAccess
Kebijakan ini memberi pengguna izin yang diperlukan untuk mencantumkan dan menjelaskan sumber daya MSK Connect.
Anda dapat melampirkan kebijakan `AmazonMSKConnectReadOnlyAccess` ke identitas IAM Anda.
------
#### [ 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 kebijakan terkelola: KafkaConnectServiceRolePolicy
Kebijakan ini memberi layanan MSK Connect izin yang diperlukan untuk membuat dan mengelola antarmuka jaringan yang memiliki tag. `AmazonMSKConnectManaged:true` Antarmuka jaringan ini memberikan akses jaringan MSK Connect ke sumber daya di VPC Amazon Anda, seperti cluster Apache Kafka atau sumber atau wastafel.
Anda tidak dapat melampirkan KafkaConnectServiceRolePolicy ke entitas IAM Anda. Kebijakan ini dilampirkan pada peran terkait layanan yang memungkinkan MSK Connect melakukan tindakan atas nama Anda.
------
#### [ 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 memperbarui kebijakan AWS terkelola
Lihat detail tentang pembaruan kebijakan AWS terkelola untuk MSK Connect sejak layanan ini mulai melacak perubahan ini.
| Ubah | Deskripsi | Date |
| --- | --- | --- |
| MSK Connect memperbarui kebijakan hanya-baca | MSK Connect memperbarui MSKConnect ReadOnlyAccess kebijakan Amazon untuk menghapus pembatasan operasi listing. | 13 Oktober 2021 |
| MSK Connect mulai melacak perubahan | MSK Connect mulai melacak perubahan untuk kebijakan AWS terkelolanya. | 14 September 2021 |
# Menggunakan peran terkait layanan untuk MSK Connect
Amazon MSK Connect menggunakan peran AWS Identity and Access Management terkait [layanan](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_terms-and-concepts.html#iam-term-service-linked-role) (IAM). Peran terkait layanan adalah jenis peran IAM unik yang ditautkan langsung ke MSK Connect. Peran terkait layanan telah ditentukan sebelumnya oleh MSK Connect dan mencakup semua izin yang diperlukan layanan untuk memanggil layanan lain AWS atas nama Anda.
Peran terkait layanan membuat pengaturan MSK Connect menjadi lebih mudah karena Anda tidak perlu menambahkan izin yang diperlukan secara manual. MSK Connect mendefinisikan izin dari peran terkait layanan, dan kecuali ditentukan lain, hanya MSK Connect yang dapat mengambil perannya. Izin yang ditentukan mencakup kebijakan kepercayaan dan kebijakan izin, serta bahwa kebijakan izin tidak dapat dilampirkan ke entitas IAM lainnya.
Untuk informasi tentang layanan lain yang mendukung peran terkait layanan, lihat [Layanan AWS yang Berfungsi dengan IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_aws-services-that-work-with-iam.html) dan cari layanan yang memiliki **Ya** di kolom **Peran Terkait Layanan**. Pilih **Ya** dengan sebuah tautan untuk melihat dokumentasi peran terkait layanan untuk layanan tersebut.
## Izin peran terkait layanan untuk MSK Connect
MSK Connect menggunakan peran terkait layanan bernama — **AWSServiceRoleForKafkaConnect**Memungkinkan Amazon MSK Connect mengakses sumber daya Amazon atas nama Anda.
Peran AWSService RoleForKafkaConnect terkait layanan mempercayai `kafkaconnect.amazonaws.com` layanan untuk mengambil peran tersebut.
Untuk informasi tentang kebijakan izin yang digunakan peran, lihat[AWS kebijakan terkelola: KafkaConnectServiceRolePolicy](mkc-security-iam-awsmanpol.md#security-iam-awsmanpol-KafkaConnectServiceRolePolicy).
Anda harus mengonfigurasi izin untuk mengizinkan entitas IAM (seperti pengguna, grup, atau peran) untuk membuat, mengedit, atau menghapus peran terkait layanan. Untuk informasi selengkapnya, silakan lihat [Izin Peran Tertaut Layanan](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#service-linked-role-permissions) di *Panduan Pengguna IAM*.
## Membuat peran terkait layanan untuk MSK Connect
Anda tidak perlu membuat peran terkait layanan secara manual. Saat Anda membuat konektor di, API Konsol Manajemen AWS, atau AWS API AWS CLI, MSK Connect membuat peran terkait layanan untuk Anda.
Jika Anda menghapus peran terkait layanan ini, dan ingin membuatnya lagi, Anda dapat mengulangi proses yang sama untuk membuat kembali peran tersebut di akun Anda. Saat Anda membuat konektor, MSK Connect membuat peran terkait layanan untuk Anda lagi.
## Mengedit peran terkait layanan untuk MSK Connect
MSK Connect tidak mengizinkan Anda mengedit peran AWSService RoleForKafkaConnect terkait layanan. Setelah membuat peran terkait layanan, Anda tidak dapat mengubah nama peran karena berbagai entitas mungkin mereferensikan peran tersebut. Namun, Anda dapat menyunting penjelasan peran menggunakan IAM. Untuk informasi selengkapnya, lihat [Mengedit Peran Tertaut Layanan](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#edit-service-linked-role) dalam *Panduan Pengguna IAM*.
## Menghapus peran terkait layanan untuk MSK Connect
Anda dapat menggunakan konsol IAM, AWS CLI atau AWS API untuk menghapus peran terkait layanan secara manual. Untuk melakukan ini, Anda harus terlebih dahulu menghapus semua konektor MSK Connect Anda secara manual, dan kemudian Anda dapat menghapus peran secara manual. Untuk informasi selengkapnya, silakan lihat [Menghapus Peran Terkait Layanan](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#delete-service-linked-role) di *Panduan Pengguna IAM*.
## Wilayah yang Didukung untuk peran terkait layanan MSK Connect
MSK Connect mendukung penggunaan peran terkait layanan di semua wilayah tempat layanan tersedia. Untuk informasi lebih lanjut, lihat [Wilayah dan Titik Akhir AWS](https://docs.aws.amazon.com/general/latest/gr/rande.html).
# Aktifkan akses internet untuk Amazon MSK Connect
Jika konektor Anda untuk Amazon MSK Connect memerlukan akses ke internet, kami sarankan Anda menggunakan pengaturan Amazon Virtual Private Cloud (VPC) berikut untuk mengaktifkan akses tersebut.
+ Konfigurasikan konektor Anda dengan subnet pribadi.
+ Buat [gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) publik atau [instans NAT](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_NAT_Instance.html) untuk VPC Anda di subnet publik. Untuk informasi selengkapnya, lihat halaman [Connect subnet ke internet atau lainnya VPCs menggunakan perangkat NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) di *Amazon Virtual Private Cloud**Panduan Pengguna*.
+ Izinkan lalu lintas keluar dari subnet pribadi Anda ke gateway atau instance NAT Anda.
# Siapkan gateway NAT untuk Amazon MSK Connect
Langkah-langkah berikut menunjukkan cara mengatur gateway NAT untuk mengaktifkan akses internet untuk konektor. Anda harus menyelesaikan langkah-langkah ini sebelum membuat konektor di subnet pribadi.
## Prasyarat lengkap untuk menyiapkan gateway NAT
Pastikan Anda memiliki item berikut.
+ ID dari Amazon Virtual Private Cloud (VPC) yang terkait dengan cluster Anda. Misalnya, *vpc-123456ab*.
+ Subnet pribadi di VPC Anda. IDs *Misalnya, *subnet-a1b2c3de, subnet-f4g5h6ij*, dll.* Anda harus mengkonfigurasi konektor Anda dengan subnet pribadi.
## Langkah-langkah untuk mengaktifkan akses internet untuk konektor Anda
**Untuk mengaktifkan akses internet untuk konektor Anda**
1. Buka Amazon Virtual Private Cloud konsol di [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
1. Buat subnet publik untuk gateway NAT Anda dengan nama deskriptif, dan catat ID subnet. Untuk petunjuk terperinci, lihat [Membuat subnet di VPC Anda](https://docs.aws.amazon.com/vpc/latest/userguide/working-with-vpcs.html#AddaSubnet).
1. Buat gateway internet sehingga VPC Anda dapat berkomunikasi dengan internet, dan catat ID gateway. Lampirkan gateway internet ke VPC Anda. Untuk petunjuk, lihat [Membuat dan melampirkan gateway internet](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Attach_Gateway).
1. Menyediakan gateway NAT publik sehingga host di subnet pribadi Anda dapat mencapai subnet publik Anda. Saat Anda membuat gateway NAT, pilih subnet publik yang Anda buat sebelumnya. Untuk petunjuk, silakan lihat [Membuat gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html#nat-gateway-creating).
1. Konfigurasikan tabel rute Anda. Anda harus memiliki dua tabel rute secara total untuk menyelesaikan pengaturan ini. Anda seharusnya sudah memiliki tabel rute utama yang dibuat secara otomatis bersamaan dengan VPC Anda. Pada langkah ini Anda membuat tabel rute tambahan untuk subnet publik Anda.
1. Gunakan pengaturan berikut untuk memodifikasi tabel rute utama VPC Anda sehingga subnet pribadi Anda merutekan lalu lintas ke gateway NAT Anda. Untuk petunjuk, lihat [Bekerja dengan tabel rute](https://docs.aws.amazon.com/vpc/latest/userguide/WorkWithRouteTables.html) di *Panduan *Amazon Virtual Private Cloud*Pengguna*.
**Tabel rute MSKC pribadi**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/msk/latest/developerguide/msk-connect-internet-access-private-subnets-example.html)
1. Ikuti petunjuk di [Buat tabel rute kustom](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Routing) untuk membuat tabel rute untuk subnet publik Anda. Saat Anda membuat tabel, masukkan nama deskriptif di kolom **Tag nama** untuk membantu Anda mengidentifikasi subnet mana yang terkait dengan tabel tersebut. Misalnya, **MSKC Publik**.
1. Konfigurasikan tabel rute **MSKC Publik** Anda menggunakan pengaturan berikut.
****
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/msk/latest/developerguide/msk-connect-internet-access-private-subnets-example.html)
# Memahami nama host DNS pribadi
Dengan dukungan nama host DNS Pribadi di MSK Connect, Anda dapat mengonfigurasi konektor untuk mereferensikan nama domain publik atau pribadi. Support tergantung pada server DNS yang ditentukan dalam set opsi *VPC DHCP*.
Set opsi DHCP adalah sekelompok konfigurasi jaringan yang digunakan instans EC2 dalam VPC untuk berkomunikasi melalui jaringan VPC. Setiap VPC memiliki set opsi DHCP default, tetapi Anda dapat membuat set opsi DHCP khusus jika Anda ingin instance di VPC menggunakan server DNS yang berbeda untuk resolusi nama domain, bukan server DNS yang disediakan Amazon. Lihat [set opsi DHCP di Amazon VPC](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_DHCP_Options.html).
Sebelum resolusi capability/feature DNS Pribadi disertakan dengan MSK Connect, konektor menggunakan resolver DNS VPC layanan untuk kueri DNS dari konektor pelanggan. Konektor tidak menggunakan server DNS yang ditentukan dalam set opsi VPC DHCP pelanggan untuk resolusi DNS.
Konektor hanya dapat mereferensikan nama host dalam konfigurasi konektor pelanggan atau plugin yang dapat diselesaikan secara publik. Mereka tidak dapat menyelesaikan nama host pribadi yang ditentukan di zona yang dihosting secara pribadi atau menggunakan server DNS di jaringan pelanggan lain.
Tanpa DNS Pribadi, pelanggan yang memilih untuk membuat database, gudang data, dan sistem seperti Secrets Manager di VPC mereka sendiri tidak dapat diakses oleh internet, tidak dapat bekerja dengan konektor MSK. Pelanggan sering menggunakan nama host DNS pribadi untuk mematuhi postur keamanan perusahaan.
# Konfigurasikan opsi VPC DHCP yang ditetapkan untuk konektor Anda
Konektor secara otomatis menggunakan server DNS yang ditentukan dalam opsi VPC DHCP mereka yang ditetapkan ketika konektor dibuat. Sebelum Anda membuat konektor, pastikan Anda mengonfigurasi opsi VPC DHCP yang ditetapkan untuk persyaratan resolusi nama host DNS konektor Anda.
Konektor yang Anda buat sebelum fitur hostname DNS Pribadi tersedia di MSK Connect terus menggunakan konfigurasi resolusi DNS sebelumnya tanpa perlu modifikasi.
Jika Anda hanya memerlukan resolusi nama host DNS yang dapat diselesaikan secara publik di konektor Anda, untuk pengaturan yang lebih mudah, kami sarankan menggunakan VPC default akun Anda saat Anda membuat konektor. Lihat [Server DNS Amazon](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#AmazonDNS) di Panduan *Pengguna Amazon VPC* untuk informasi selengkapnya tentang server DNS yang disediakan Amazon atau Resolver Amazon Route 53.
Jika Anda perlu menyelesaikan nama host DNS pribadi, pastikan VPC yang diteruskan selama pembuatan konektor memiliki opsi DHCP yang diatur dengan benar. Untuk informasi selengkapnya, lihat [Bekerja dengan set opsi DHCP](https://docs.aws.amazon.com/vpc/latest/userguide/DHCPOptionSet.html) di Panduan *Pengguna Amazon VPC*.
Saat Anda mengonfigurasi opsi DHCP yang ditetapkan untuk resolusi nama host DNS pribadi, pastikan konektor dapat mencapai server DNS khusus yang Anda konfigurasikan dalam set opsi DHCP. Jika tidak, pembuatan konektor Anda akan gagal.
Setelah Anda menyesuaikan set opsi VPC DHCP, konektor yang kemudian dibuat di VPC tersebut menggunakan server DNS yang Anda tentukan dalam set opsi. Jika Anda mengubah set opsi setelah Anda membuat konektor, konektor mengadopsi pengaturan dalam opsi baru yang ditetapkan dalam beberapa menit.
# Konfigurasikan atribut DNS untuk VPC Anda
[*Pastikan Anda memiliki atribut DNS VPC yang dikonfigurasi dengan benar seperti yang dijelaskan dalam atribut DNS di nama host [VPC dan DNS Anda di Panduan Pengguna Amazon VPC](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#vpc-dns-support)*.](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#vpc-dns-hostnames)
Lihat [Menyelesaikan kueri DNS antara VPCs dan jaringan Anda](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/resolver.html) di *Panduan Pengembang Amazon Route 53* untuk informasi tentang penggunaan titik akhir resolver masuk dan keluar guna menghubungkan jaringan lain ke VPC agar berfungsi dengan konektor Anda.
# Tangani kegagalan pembuatan konektor
Bagian ini menjelaskan kemungkinan kegagalan pembuatan konektor yang terkait dengan resolusi DNS dan tindakan yang disarankan untuk menyelesaikan masalah.
| Kegagalan | Tindakan yang disarankan |
| --- | --- |
| Pembuatan konektor gagal jika kueri resolusi DNS gagal, atau jika server DNS tidak dapat dijangkau dari konektor. | Anda dapat melihat kegagalan pembuatan konektor karena kueri resolusi DNS yang gagal di CloudWatch log Anda, jika Anda telah mengonfigurasi log ini untuk konektor Anda. Periksa konfigurasi server DNS dan pastikan konektivitas jaringan ke server DNS dari konektor. |
| Jika Anda mengubah konfigurasi server DNS di opsi VPC DHCP yang disetel saat konektor berjalan, kueri resolusi DNS dari konektor dapat gagal. Jika resolusi DNS gagal, beberapa tugas konektor dapat memasuki status gagal. | Anda dapat melihat kegagalan pembuatan konektor karena kueri resolusi DNS yang gagal di CloudWatch log Anda, jika Anda telah mengonfigurasi log ini untuk konektor Anda. Tugas yang gagal harus dimulai ulang secara otomatis untuk mengembalikan konektor. Jika itu tidak terjadi, Anda dapat menghubungi dukungan untuk memulai kembali tugas yang gagal untuk konektor mereka atau Anda dapat membuat ulang konektor. |
# Keamanan untuk MSK Connect
Anda dapat menggunakan Endpoint VPC Antarmuka, yang didukung oleh AWS PrivateLink, untuk mencegah lalu lintas antara Amazon VPC dan Amazon MSK-Connect yang kompatibel dari meninggalkan jaringan Amazon. APIs Endpoint VPC antarmuka tidak memerlukan gateway internet, perangkat NAT, koneksi VPN, atau koneksi. AWS Direct Connect Lihat informasi yang lebih lengkap di [Gunakan Amazon MSK APIs dengan Titik Akhir VPC Antarmuka](privatelink-vpc-endpoints.md).
# Logging untuk MSK Connect
MSK Connect dapat menulis peristiwa log yang dapat Anda gunakan untuk men-debug konektor Anda. Saat Anda membuat konektor, Anda dapat menentukan nol atau lebih dari tujuan log berikut:
+ Amazon CloudWatch Logs: Anda menentukan grup log yang Anda inginkan MSK Connect untuk mengirim peristiwa log konektor Anda. Untuk informasi tentang cara membuat grup log, lihat [Membuat grup CloudWatch log](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html#Create-Log-Group) *di Panduan Pengguna Log*.
+ Amazon S3: Anda menentukan bucket S3 yang Anda inginkan MSK Connect untuk mengirim peristiwa log konektor Anda. Untuk informasi tentang cara membuat bucket S3, lihat [Membuat bucket di Panduan](https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket-overview.html) Pengguna *Amazon S3*.
+ Amazon Data Firehose: Anda menentukan aliran pengiriman yang Anda inginkan MSK Connect untuk mengirim peristiwa log konektor Anda. Untuk informasi tentang cara membuat aliran pengiriman, lihat [Membuat aliran pengiriman Amazon Data Firehose di Panduan](https://docs.aws.amazon.com/firehose/latest/dev/basic-create.html) Pengguna *Firehose*.
Untuk mempelajari lebih lanjut tentang mengatur logging, lihat [Mengaktifkan logging dari AWS layanan tertentu](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/AWS-logs-and-resource-policy.html) di *Panduan Amazon CloudWatch Logs Pengguna*.
MSK Connect memancarkan jenis peristiwa log berikut:
****
| Tingkat | Deskripsi |
| --- | --- |
| INFO | Acara runtime yang menarik saat startup dan shutdown. |
| WARN | Situasi runtime yang bukan kesalahan tetapi tidak diinginkan atau tidak terduga. |
| FATAL | Kesalahan parah yang menyebabkan penghentian prematur. |
| ERROR | Kondisi tak terduga dan kesalahan runtime yang tidak fatal. |
Berikut ini adalah contoh peristiwa log yang dikirim ke CloudWatch Log:
```
[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)
```
## Mencegah rahasia muncul di log konektor
**catatan**
Nilai konfigurasi sensitif dapat muncul di log konektor jika plugin tidak mendefinisikan nilai-nilai tersebut sebagai rahasia. Kafka Connect memperlakukan nilai konfigurasi yang tidak ditentukan sama dengan nilai plaintext lainnya.
Jika plugin Anda mendefinisikan properti sebagai rahasia, Kafka Connect menyunting nilai properti dari log konektor. Misalnya, log konektor berikut menunjukkan bahwa jika plugin mendefinisikan `aws.secret.key` sebagai `PASSWORD` tipe, maka nilainya diganti dengan`[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)
```
Untuk mencegah rahasia muncul di file log konektor, pengembang plugin harus menggunakan konstanta enum Kafka Connect [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)untuk menentukan properti sensitif. Ketika properti adalah tipe`ConfigDef.Type.PASSWORD`, Kafka Connect mengecualikan nilainya dari log konektor bahkan jika nilai dikirim sebagai plaintext.
# Memantau Amazon MSK Connect
Pemantauan merupakan bagian penting dalam menjaga keandalan, ketersediaan, dan kinerja MSK Connect dan AWS solusi Anda yang lain. Amazon CloudWatch memantau AWS sumber daya Anda dan aplikasi yang Anda jalankan AWS secara real-time. Anda dapat mengumpulkan dan melacak metrik, membuat dasbor yang disesuaikan, dan mengatur alarm yang memberi tahu Anda atau mengambil tindakan saat metrik tertentu mencapai ambang batas yang ditentukan. Misalnya, Anda dapat CloudWatch melacak penggunaan CPU atau metrik lain dari konektor Anda, sehingga Anda dapat meningkatkan kapasitasnya jika diperlukan. Untuk informasi selengkapnya, lihat [Panduan CloudWatch Pengguna Amazon](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/).
Anda dapat menggunakan operasi API berikut:
+ `DescribeConnectorOperation`: Pantau status operasi pembaruan konektor.
+ `ListConnectorOperations`: Lacak pembaruan sebelumnya yang berjalan di konektor Anda.
Tabel berikut menunjukkan metrik yang MSK Connect kirimkan ke CloudWatch bawah dimensi. `ConnectorName` MSK Connect memberikan metrik ini secara default dan tanpa biaya tambahan. CloudWatch menyimpan metrik ini selama 15 bulan, sehingga Anda dapat mengakses informasi historis dan mendapatkan perspektif yang lebih baik tentang kinerja konektor Anda. Anda juga dapat mengatur alarm yang memperhatikan ambang batas tertentu dan mengirim notifikasi atau mengambil tindakan saat ambang batas tersebut terpenuhi. Untuk informasi selengkapnya, lihat [Panduan CloudWatch Pengguna Amazon](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/).
| Nama metrik | Deskripsi |
| --- | --- |
| CpuUtilization | Persentase konsumsi CPU oleh sistem dan pengguna. |
| ErroredTaskCount | Jumlah tugas yang telah salah. |
| MemoryUtilization | Persentase total memori pada instance pekerja, bukan hanya memori heap Java virtual machine (JVM) yang saat ini digunakan. JVM biasanya tidak melepaskan memori kembali ke sistem operasional. Jadi, ukuran tumpukan JVM (MemoryUtilization) biasanya dimulai dengan ukuran tumpukan minimum yang secara bertahap meningkat hingga maksimum stabil sekitar 80-90%. Penggunaan heap JVM mungkin meningkat atau berkurang karena penggunaan memori aktual konektor berubah. |
| RebalanceCompletedTotal | Jumlah total rebalances yang diselesaikan oleh konektor ini. |
| RebalanceTimeAvg | Waktu rata-rata dalam milidetik yang dihabiskan oleh konektor untuk menyeimbangkan kembali. |
| RebalanceTimeMax | Waktu maksimum dalam milidetik yang dihabiskan oleh konektor untuk menyeimbangkan kembali. |
| RebalanceTimeSinceLast | Waktu dalam milidetik sejak konektor ini menyelesaikan penyeimbangan ulang terbaru. |
| RunningTaskCount | Jumlah tugas yang berjalan di konektor. |
| SinkConsumerByteRate | Jumlah rata-rata byte yang dikonsumsi per detik oleh konsumen Sink kerangka kerja Kafka Connect sebelum transformasi apa pun diterapkan pada data. |
| SinkRecordReadRate | Rata-rata jumlah catatan per detik dibaca dari kluster Apache Kafka atau Amazon MSK. |
| SinkRecordSendRate | Rata-rata jumlah rekaman per detik yang dihasilkan dari transformasi dan dikirim ke tujuan. Nomor ini tidak termasuk catatan yang difilter. |
| SourceRecordPollRate | Jumlah rata-rata per detik catatan yang diproduksi atau disurvei. |
| SourceProducerByteRate | Jumlah rata-rata byte yang dihasilkan per detik oleh produsen Source kerangka kerja Kafka Connect setelah transformasi apa pun diterapkan pada data. |
| SourceRecordWriteRate | Rata-rata jumlah rekaman per detik output dari transformasi dan ditulis ke Apache Kafka atau Amazon MSK cluster. |
| TaskStartupAttemptsTotal | Jumlah total startup tugas yang telah dicoba konektor. Anda dapat menggunakan metrik ini untuk mengidentifikasi anomali dalam upaya memulai tugas. |
| TaskStartupSuccessPercentage | Persentase rata-rata tugas yang berhasil dimulai untuk konektor. Anda dapat menggunakan metrik ini untuk mengidentifikasi anomali dalam upaya memulai tugas. |
| WorkerCount | Jumlah pekerja yang berjalan di konektor. |
| BytesInPerSec | Metadata byte ditransfer ke kerangka kerja Kafka Connect untuk komunikasi antar pekerja. |
| BytesOutPerSec | Metadata byte ditransfer dari kerangka kerja Kafka Connect untuk komunikasi antar pekerja. |
# Contoh untuk menyiapkan sumber daya Amazon MSK Connect
Bagian ini mencakup contoh untuk membantu Anda menyiapkan sumber daya Amazon MSK Connect seperti konektor pihak ketiga umum dan penyedia konfigurasi.
**Topics**
+ [Siapkan konektor wastafel Amazon S3](mkc-S3sink-connector-example.md)
+ [Siapkan konektor wastafel EventBridge Kafka untuk MSK Connect](mkc-eventbridge-kafka-connector.md)
+ [Gunakan konektor sumber Debezium dengan penyedia konfigurasi](mkc-debeziumsource-connector-example.md)
# Siapkan konektor wastafel Amazon S3
Contoh ini menunjukkan cara menggunakan konektor wastafel [Amazon S3 Confluent dan untuk membuat konektor wastafel](https://www.confluent.io/hub/confluentinc/kafka-connect-s3) Amazon S3 AWS CLI di MSK Connect.
1. Salin JSON berikut dan tempel di file baru. Ganti string placeholder dengan nilai yang sesuai dengan string koneksi server bootstrap Amazon MSK cluster Anda dan subnet dan grup keamanan cluster. IDs Untuk informasi tentang cara menyiapkan peran eksekusi layanan, lihat[Peran dan kebijakan IAM untuk 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. Jalankan AWS CLI perintah berikut di folder tempat Anda menyimpan file JSON di langkah sebelumnya.
```
aws kafkaconnect create-connector --cli-input-json file://connector-info.json
```
Berikut ini adalah contoh output yang Anda dapatkan ketika Anda menjalankan perintah dengan sukses.
```
{
"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"
}
```
# Siapkan konektor wastafel EventBridge Kafka untuk MSK Connect
Topik ini menunjukkan cara mengatur [konektor sink EventBridge Kafka](https://github.com/awslabs/eventbridge-kafka-connector) untuk MSK Connect. Konektor ini memungkinkan Anda mengirim acara dari cluster MSK Anda ke [bus EventBridge acara](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-bus.html). Topik ini menjelaskan proses pembuatan sumber daya yang diperlukan dan mengonfigurasi konektor untuk memungkinkan aliran data yang mulus antara Kafka dan. EventBridge
**Topics**
+ [Prasyarat](#mkc-eb-kafka-prerequisites)
+ [Siapkan sumber daya yang diperlukan untuk MSK Connect](#mkc-eb-kafka-set-up-resources)
+ [Buat konektor](#mkc-eb-kafka-create-connector)
+ [Kirim pesan kepada Kafka](#mkc-eb-kafka-send-json-encoded-messages)
## Prasyarat
Sebelum menggunakan konektor, pastikan Anda memiliki sumber daya berikut:
+ **Amazon MSK cluster: Cluster** MSK aktif untuk memproduksi dan mengkonsumsi pesan Kafka.
+ **Bus EventBridge acara Amazon: Bus** EventBridge acara untuk menerima acara dari topik Kafka.
+ **Peran IAM**: Buat peran IAM dengan izin yang diperlukan untuk MSK Connect dan konektor. EventBridge
+ [Akses ke internet publik](msk-connect-internet-access.md) dari MSK Connect atau titik akhir [antarmuka VPC](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-related-service-vpc.html) EventBridge untuk dibuat di VPC dan subnet cluster MSK Anda. Ini membantu Anda menghindari melintasi internet publik dan tanpa memerlukan gateway NAT.
+ [Mesin klien](create-serverless-cluster-client.md), seperti instans Amazon EC2 atau [AWS CloudShell](https://aws.amazon.com/cloudshell/), untuk membuat topik dan mengirim catatan ke Kafka.
## Siapkan sumber daya yang diperlukan untuk MSK Connect
Anda membuat peran IAM untuk konektor, dan kemudian Anda membuat konektor. Anda juga membuat EventBridge aturan untuk memfilter acara Kafka yang dikirim ke bus EventBridge acara.
**Topics**
+ [Peran IAM untuk konektor](#mkc-eb-kafka-iam-role-connector)
+ [EventBridge Aturan untuk acara yang masuk](#mkc-eb-kafka-create-rule)
### Peran IAM untuk konektor
Peran IAM yang Anda kaitkan dengan konektor harus memiliki [PutEvents](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-permissions-reference.html)izin untuk mengizinkan pengiriman acara. EventBridge Contoh kebijakan IAM berikut memberi Anda izin untuk mengirim acara ke bus acara bernama. `example-event-bus` Pastikan Anda mengganti sumber daya ARN pada contoh berikut dengan ARN bus acara Anda.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"events:PutEvents"
],
"Resource": "arn:aws:events:us-east-1:123456789012:event-bus/example-event-bus"
}
]
}
```
------
Selain itu, Anda harus memastikan bahwa peran IAM Anda untuk konektor berisi kebijakan kepercayaan berikut.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "kafkaconnect.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
### EventBridge Aturan untuk acara yang masuk
Anda membuat [aturan](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-rules.html) yang cocok dengan peristiwa yang masuk dengan kriteria data peristiwa, yang dikenal sebagai [https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html). [Dengan pola peristiwa, Anda dapat menentukan kriteria untuk memfilter peristiwa yang masuk, dan menentukan peristiwa mana yang harus memicu aturan tertentu dan selanjutnya diarahkan ke target yang ditentukan.](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-targets.html) Contoh pola acara berikut cocok dengan acara Kafka yang dikirim ke bus EventBridge acara.
```
{
"detail": {
"topic": ["msk-eventbridge-tutorial"]
}
}
```
Berikut ini adalah contoh acara yang dikirim dari Kafka untuk EventBridge menggunakan konektor wastafel Kafka.
```
{
"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"
}
}
}
```
Di EventBridge konsol, [buat aturan](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-create-rule.html) di bus acara menggunakan pola contoh ini dan tentukan target, seperti grup CloudWatch Log. EventBridge Konsol akan secara otomatis mengonfigurasi kebijakan akses yang diperlukan untuk grup CloudWatch Log.
## Buat konektor
Di bagian berikut, Anda membuat dan menggunakan [konektor wastafel EventBridge Kafka](https://github.com/awslabs/eventbridge-kafka-connector) menggunakan. Konsol Manajemen AWS
**Topics**
+ [Langkah 1: Unduh konektornya](#mkc-eb-kafka-download-connector)
+ [Langkah 2: Buat ember Amazon S3](#mkc-eb-kafka-s3-bucket-create)
+ [Langkah 3: Buat plugin di MSK Connect](#mkc-eb-kafka-create-plugin)
+ [Langkah 4: Buat konektor](#mkc-eb-kafka-create-connector)
### Langkah 1: Unduh konektornya
Unduh EventBridge konektor wastafel JAR terbaru dari [halaman GitHub rilis](https://github.com/awslabs/eventbridge-kafka-connector/releases) untuk konektor EventBridge Kafka. Misalnya, untuk mengunduh versi v1.4.1, pilih tautan file JAR,`kafka-eventbridge-sink-with-dependencies.jar`, untuk mengunduh konektor. Kemudian, simpan file ke lokasi yang diinginkan di mesin Anda.
### Langkah 2: Buat ember Amazon S3
1. Untuk menyimpan file JAR di Amazon S3 untuk digunakan dengan MSK Connect, buka, lalu Konsol Manajemen AWS pilih Amazon S3.
1. Di konsol Amazon S3, pilih **Buat bucket**, dan masukkan nama bucket unik. Misalnya, **amzn-s3-demo-bucket1-eb-connector**.
1. Pilih Wilayah yang sesuai untuk bucket Amazon S3 Anda. Pastikan bahwa itu cocok dengan Wilayah tempat klaster MSK Anda digunakan.
1. Untuk **pengaturan Bucket**, pertahankan pilihan default atau sesuaikan sesuai kebutuhan.
1. Pilih **Buat ember**
1. Unggah file JAR ke bucket Amazon S3.
### Langkah 3: Buat plugin di MSK Connect
1. Buka Konsol Manajemen AWS, lalu arahkan ke **MSK Connect**.
1. Di panel navigasi kiri, pilih **Plugin khusus**.
1. Pilih **Buat plugin**, lalu masukkan **nama Plugin**. Misalnya, **eventbridge-sink-plugin**.
1. Untuk **lokasi plugin Kustom**, tempel **URL objek S3**.
1. Tambahkan deskripsi opsional untuk plugin.
1. Pilih **Buat plugin**.
Setelah plugin dibuat, Anda dapat menggunakannya untuk mengkonfigurasi dan menyebarkan konektor EventBridge Kafka di MSK Connect.
### Langkah 4: Buat konektor
Sebelum membuat konektor, kami sarankan untuk membuat topik Kafka yang diperlukan untuk menghindari kesalahan konektor. Untuk membuat topik, gunakan mesin klien Anda.
1. Di panel kiri konsol MSK, pilih **Konektor, lalu pilih **Buat** konektor**.
1. Dalam daftar plugin, pilih **eventbridge-sink-plugin**, lalu pilih **Berikutnya**.
1. Untuk nama konektor, masukkan**EventBridgeSink**.
1. Dalam daftar cluster, pilih cluster MSK Anda.
1. Salin konfigurasi berikut untuk konektor dan tempelkan ke bidang **konfigurasi Konektor**
Ganti placeholder dalam konfigurasi berikut, sesuai kebutuhan.
+ Hapus `aws.eventbridge.endpoint.uri` jika klaster MSK Anda memiliki akses internet publik.
+ Jika Anda menggunakan PrivateLink untuk terhubung dengan aman dari MSK ke EventBridge, ganti bagian DNS setelahnya `https://` dengan nama DNS pribadi yang benar dari titik akhir antarmuka VPC (opsional) untuk yang Anda buat sebelumnya. EventBridge
+ Ganti bus EventBridge acara ARN dalam konfigurasi berikut dengan ARN bus acara Anda.
+ Perbarui nilai spesifik Wilayah apa pun.
```
{
"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"
}
```
Untuk informasi selengkapnya tentang konfigurasi konektor, lihat [eventbridge-kafka-connector](https://github.com/awslabs/eventbridge-kafka-connector).
Jika perlu, ubah pengaturan untuk pekerja dan penskalaan otomatis. Kami juga merekomendasikan untuk menggunakan versi Apache Kafka Connect terbaru yang tersedia (disarankan) dari dropdown. Di bawah **Izin akses**, gunakan peran yang dibuat sebelumnya. Kami juga merekomendasikan untuk mengaktifkan logging CloudWatch untuk observabilitas dan pemecahan masalah. Sesuaikan pengaturan opsional lainnya, seperti tag, berdasarkan kebutuhan Anda. Kemudian, gunakan konektor dan tunggu status masuk status Running.
## Kirim pesan kepada Kafka
Anda dapat mengonfigurasi pengkodean pesan, seperti Apache Avro dan JSON, dengan menentukan konverter yang berbeda menggunakan dan, secara opsional, pengaturan yang tersedia di `value.converter` Kafka Connect. `key.converter`
[connector example](#connector-ex)Dalam topik ini dikonfigurasi untuk bekerja dengan pesan yang dikodekan JSON, seperti yang ditunjukkan oleh penggunaan for. `org.apache.kafka.connect.json.JsonConverter` `value converter` Saat konektor dalam status Running, kirim catatan ke topik `msk-eventbridge-tutorial` Kafka dari mesin klien Anda.
# Gunakan konektor sumber Debezium dengan penyedia konfigurasi
[Contoh ini menunjukkan cara menggunakan plugin konektor Debezium MySQL dengan database Amazon Aurora yang kompatibel dengan MySQL sebagai sumbernya.](https://aws.amazon.com/rds/aurora/) Dalam contoh ini, kami juga menyiapkan Penyedia [Config AWS Secrets Manager](https://github.com/jcustenborder/kafka-config-provider-aws) open-source untuk mengeksternalisasi kredenal database di. AWS Secrets Manager Untuk mempelajari lebih lanjut tentang penyedia konfigurasi, lihat[Tutorial: Eksternalisasi informasi sensitif menggunakan penyedia konfigurasi](msk-connect-config-provider.md).
**penting**
Plugin konektor Debezium MySQL [hanya mendukung satu tugas](https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-property-tasks-max) dan tidak berfungsi dengan mode kapasitas skala otomatis untuk Amazon MSK Connect. Sebagai gantinya, Anda harus menggunakan mode kapasitas yang disediakan dan menyetel `workerCount` sama dengan satu di konfigurasi konektor Anda. Untuk mempelajari lebih lanjut tentang mode kapasitas MSK Connect, lihat[Memahami kapasitas konektor](msk-connect-capacity.md).
# Prasyarat lengkap untuk menggunakan konektor sumber Debezium
Konektor Anda harus dapat mengakses internet sehingga dapat berinteraksi dengan layanan seperti AWS Secrets Manager yang berada di luar Anda Amazon Virtual Private Cloud. Langkah-langkah di bagian ini membantu Anda menyelesaikan tugas-tugas berikut untuk mengaktifkan akses internet.
+ Siapkan subnet publik yang menghosting gateway NAT dan merutekan lalu lintas ke gateway internet di VPC Anda.
+ Buat rute default yang mengarahkan lalu lintas subnet pribadi Anda ke gateway NAT Anda.
Untuk informasi selengkapnya, lihat [Aktifkan akses internet untuk Amazon MSK Connect](msk-connect-internet-access.md).
**Prasyarat**
Sebelum Anda dapat mengaktifkan akses internet, Anda memerlukan item berikut:
+ ID dari Amazon Virtual Private Cloud (VPC) yang terkait dengan cluster Anda. Misalnya, *vpc-123456ab*.
+ Subnet pribadi di VPC Anda. IDs *Misalnya, *subnet-a1b2c3de, subnet-f4g5h6ij*, dll.* Anda harus mengkonfigurasi konektor Anda dengan subnet pribadi.
**Untuk mengaktifkan akses internet untuk konektor Anda**
1. Buka Amazon Virtual Private Cloud konsol di [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
1. Buat subnet publik untuk gateway NAT Anda dengan nama deskriptif, dan catat ID subnet. Untuk petunjuk terperinci, lihat [Membuat subnet di VPC Anda](https://docs.aws.amazon.com/vpc/latest/userguide/working-with-vpcs.html#AddaSubnet).
1. Buat gateway internet sehingga VPC Anda dapat berkomunikasi dengan internet, dan catat ID gateway. Pasang gateway internet ke VPC Anda. Untuk petunjuk, lihat [Membuat dan melampirkan gateway internet](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Attach_Gateway).
1. Menyediakan gateway NAT publik sehingga host di subnet pribadi Anda dapat mencapai subnet publik Anda. Saat Anda membuat gateway NAT, pilih subnet publik yang Anda buat sebelumnya. Untuk petunjuk, silakan lihat [Membuat gateway NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html#nat-gateway-creating).
1. Konfigurasikan tabel rute Anda. Anda harus memiliki dua tabel rute secara total untuk menyelesaikan pengaturan ini. Anda seharusnya sudah memiliki tabel rute utama yang dibuat secara otomatis bersamaan dengan VPC Anda. Pada langkah ini Anda membuat tabel rute tambahan untuk subnet publik Anda.
1. Gunakan pengaturan berikut untuk memodifikasi tabel rute utama VPC Anda sehingga subnet pribadi Anda merutekan lalu lintas ke gateway NAT Anda. Untuk petunjuk, lihat [Bekerja dengan tabel rute](https://docs.aws.amazon.com/vpc/latest/userguide/WorkWithRouteTables.html) di *Panduan *Amazon Virtual Private Cloud*Pengguna*.
**Tabel rute MSKC pribadi**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html)
1. Ikuti petunjuk di [Buat tabel rute kustom](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Routing) untuk membuat tabel rute untuk subnet publik Anda. Saat Anda membuat tabel, masukkan nama deskriptif di kolom **Tag nama** untuk membantu Anda mengidentifikasi subnet mana yang terkait dengan tabel tersebut. Misalnya, **MSKC Publik**.
1. Konfigurasikan tabel rute **MSKC Publik** Anda menggunakan pengaturan berikut.
****
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html)
Sekarang setelah Anda mengaktifkan akses internet untuk Amazon MSK Connect, Anda siap membuat konektor.
# Buat konektor sumber Debezium
Prosedur ini menjelaskan cara membuat konektor sumber Debezium.
1.
**Buat plugin khusus**
1. [Unduh plugin konektor MySQL untuk rilis stabil terbaru dari situs Debezium.](https://debezium.io/releases/) Catat versi rilis Debezium yang Anda unduh (versi 2.x, atau seri 1.x yang lebih lama). Kemudian dalam prosedur ini, Anda akan membuat konektor berdasarkan versi Debezium Anda.
1. Unduh dan ekstrak Penyedia [Config AWS Secrets Manager](https://www.confluent.io/hub/jcustenborder/kafka-config-provider-aws).
1. Tempatkan arsip berikut ke dalam direktori yang sama:
+ `debezium-connector-mysql`Folder
+ `jcusten-border-kafka-config-provider-aws-0.1.1`Folder
1. Kompres direktori yang Anda buat pada langkah sebelumnya ke dalam file ZIP dan kemudian unggah file ZIP ke bucket S3. Untuk petunjuk, lihat [Mengunggah objek](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html) di Panduan Pengguna *Amazon S3*.
1. Salin JSON berikut dan tempel dalam file. Misalnya, `debezium-source-custom-plugin.json`. Ganti ** dengan nama yang Anda inginkan plugin, ** dengan ARN bucket Amazon S3 tempat Anda mengunggah file ZIP, `` dan dengan kunci file objek ZIP yang Anda unggah ke S3.
```
{
"name": "",
"contentType": "ZIP",
"location": {
"s3Location": {
"bucketArn": "",
"fileKey": ""
}
}
}
```
1. Jalankan AWS CLI perintah berikut dari folder tempat Anda menyimpan file JSON untuk membuat plugin.
```
aws kafkaconnect create-custom-plugin --cli-input-json file://
```
Anda akan melihat output yang mirip dengan contoh berikut.
```
{
"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. Jalankan perintah berikut untuk memeriksa status plugin. Negara harus berubah dari `CREATING` ke`ACTIVE`. Ganti placeholder ARN dengan ARN yang Anda dapatkan di output dari perintah sebelumnya.
```
aws kafkaconnect describe-custom-plugin --custom-plugin-arn ""
```
1.
**Konfigurasikan AWS Secrets Manager dan buat rahasia untuk kredensi database Anda**
1. Buka konsol Secrets Manager di [https://console.aws.amazon.com/secretsmanager/](https://console.aws.amazon.com/secretsmanager/).
1. Buat rahasia baru untuk menyimpan kredensi login database Anda. Untuk petunjuk, lihat [Membuat rahasia](https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_create-basic-secret.html) di Panduan *AWS Secrets Manager*Pengguna.
1. Salin ARN rahasia Anda.
1. Tambahkan izin Secrets Manager dari kebijakan contoh berikut ke kebijakan Anda[Memahami peran eksekusi layanan](msk-connect-service-execution-role.md). Ganti ** dengan ARN rahasia Anda.
------
#### [ 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"
]
}
]
}
```
------
*Untuk petunjuk tentang cara menambahkan izin IAM, lihat [Menambahkan dan menghapus izin identitas IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) di Panduan Pengguna IAM.*
1.
**Buat konfigurasi pekerja khusus dengan informasi tentang penyedia konfigurasi Anda**
1. Salin properti konfigurasi pekerja berikut ke dalam file, ganti string placeholder dengan nilai yang sesuai dengan skenario Anda. Untuk mempelajari lebih lanjut tentang properti konfigurasi untuk Penyedia Config AWS Secrets Manager, lihat [SecretsManagerConfigProvider](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-config-provider-aws/configProviders/SecretsManagerConfigProvider.html)di dokumentasi 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. Jalankan AWS CLI perintah berikut untuk membuat konfigurasi pekerja kustom Anda.
Ganti nilai-nilai berikut:
+ **- nama deskriptif untuk konfigurasi pekerja kustom Anda
+ **- versi base64 yang dikodekan dari properti plaintext yang Anda salin pada langkah sebelumnya
```
aws kafkaconnect create-worker-configuration --name --properties-file-content
```
1.
**Buat konektor**
1. Salin JSON berikut yang sesuai dengan versi Debezium Anda (2.x atau 1.x) dan tempel di file baru. Ganti `` string dengan nilai yang sesuai dengan skenario Anda. Untuk informasi tentang cara menyiapkan peran eksekusi layanan, lihat[Peran dan kebijakan IAM untuk MSK Connect](msk-connect-iam.md).
Perhatikan bahwa konfigurasi menggunakan variabel seperti `${secretManager:MySecret-1234:dbusername}` bukan plaintext untuk menentukan kredenal database. Ganti `MySecret-1234` dengan nama rahasia Anda dan kemudian sertakan nama kunci yang ingin Anda ambil. Anda juga harus mengganti `` dengan ARN konfigurasi pekerja kustom Anda.
------
#### [ Debezium 2.x ]
Untuk versi Debezium 2.x, salin JSON berikut dan tempel di file baru. Ganti ** string dengan nilai yang sesuai dengan skenario Anda.
```
{
"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 ]
Untuk versi Debezium 1.x, salin JSON berikut dan tempel di file baru. Ganti ** string dengan nilai yang sesuai dengan skenario Anda.
```
{
"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. Jalankan AWS CLI perintah berikut di folder tempat Anda menyimpan file JSON di langkah sebelumnya.
```
aws kafkaconnect create-connector --cli-input-json file://connector-info.json
```
Berikut ini adalah contoh output yang Anda dapatkan ketika Anda menjalankan perintah dengan sukses.
```
{
"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"
}
```
# Perbarui konfigurasi konektor Debezium
Untuk memperbarui konfigurasi konektor Debezium, ikuti langkah-langkah ini:
1. Salin JSON berikut dan tempel ke file baru. Ganti `` string dengan nilai yang sesuai dengan skenario Anda.
```
{
"connectorArn": ,
"connectorConfiguration": ,
"currentVersion":
}
```
1. Jalankan AWS CLI perintah berikut di folder tempat Anda menyimpan file JSON di langkah sebelumnya.
```
aws kafkaconnect update-connector --cli-input-json file://connector-info.json
```
Berikut ini adalah contoh output ketika Anda menjalankan perintah berhasil.
```
{
"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. Anda sekarang dapat menjalankan perintah berikut untuk memantau keadaan operasi saat ini:
```
aws kafkaconnect describe-connector-operation --connector-operation-arn
```
Untuk contoh konektor Debezium dengan langkah-langkah terperinci, lihat Memperkenalkan [Amazon MSK Connect - Streaming Data ke dan dari Cluster Apache Kafka Anda Menggunakan](https://aws.amazon.com/blogs/aws/introducing-amazon-msk-connect-stream-data-to-and-from-your-apache-kafka-clusters-using-managed-connectors/) Konektor Terkelola.
# Migrasi ke Amazon MSK Connect
Bagian ini menjelaskan cara memigrasikan aplikasi konektor Apache Kafka Anda ke Amazon Managed Streaming for Apache Kafka Connect (Amazon MSK Connect). Untuk mengetahui lebih lanjut tentang manfaat migrasi ke Amazon MSK Connect, lihat. [Manfaat menggunakan Amazon MSK Connect](msk-connect.md#msk-connect-benefits)
Bagian ini juga menjelaskan topik manajemen negara yang digunakan oleh Kafka Connect dan Amazon MSK Connect dan mencakup prosedur untuk memigrasi konektor sumber dan sink.
# Memahami topik internal yang digunakan oleh Kafka Connect
Aplikasi Apache Kafka Connect yang berjalan dalam mode terdistribusi menyimpan statusnya dengan menggunakan topik internal di cluster Kafka dan keanggotaan grup. Berikut ini adalah nilai konfigurasi yang sesuai dengan topik internal yang digunakan untuk aplikasi Kafka Connect:
+ Topik konfigurasi, ditentukan melalui `config.storage.topic`
Dalam topik konfigurasi, Kafka Connect menyimpan konfigurasi semua konektor dan tugas yang telah dimulai oleh pengguna. Setiap kali pengguna memperbarui konfigurasi konektor atau ketika konektor meminta konfigurasi ulang (misalnya, konektor mendeteksi bahwa ia dapat memulai lebih banyak tugas), catatan dipancarkan ke topik ini. Topik ini diaktifkan pemadatan, sehingga selalu menyimpan status terakhir untuk setiap entitas.
+ Topik offset, ditentukan melalui `offset.storage.topic`
Dalam topik offset, Kafka Connect menyimpan offset konektor sumber. Seperti topik konfigurasi, topik offset diaktifkan pemadatan. Topik ini digunakan untuk menulis posisi sumber hanya untuk konektor sumber yang menghasilkan data ke Kafka dari sistem eksternal. Konektor sink, yang membaca data dari Kafka dan mengirim ke sistem eksternal, menyimpan offset konsumen mereka dengan menggunakan kelompok konsumen Kafka biasa.
+ Topik status, ditentukan melalui `status.storage.topic`
Dalam topik status, Kafka Connect menyimpan kondisi konektor dan tugas saat ini. Topik ini digunakan sebagai tempat sentral untuk data yang ditanyakan oleh pengguna REST API. Topik ini memungkinkan pengguna untuk menanyakan pekerja mana pun dan masih mendapatkan status semua plugin yang sedang berjalan. Seperti topik konfigurasi dan offset, topik status juga diaktifkan pemadatan.
Selain topik-topik ini, Kafka Connect memanfaatkan API keanggotaan grup Kafka secara ekstensif. Grup diberi nama setelah nama konektor. Misalnya, untuk konektor bernama file-sink, grup diberi nama. connect-file-sink Setiap konsumen dalam grup memberikan catatan untuk satu tugas. Kelompok-kelompok ini dan offsetnya dapat diambil dengan menggunakan alat kelompok konsumen biasa, seperti. `Kafka-consumer-group.sh` Untuk setiap konektor sink, runtime Connect menjalankan grup konsumen reguler yang mengekstrak catatan dari Kafka.
# Manajemen negara bagian aplikasi Amazon MSK Connect
Secara default, Amazon MSK Connect membuat tiga topik terpisah di cluster Kafka untuk setiap Konektor MSK Amazon untuk menyimpan konfigurasi, offset, dan status konektor. Nama topik default disusun sebagai berikut:
+ *connector-name*\$1\$1msk\$1connect\$1configs\$1 \$1 *connector-id*
+ *connector-name*\$1\$1msk\$1connect\$1status\$1 \$1 *connector-id*
+ *connector-name*\$1\$1msk\$1connect\$1offsets\$1 \$1 *connector-id*
**catatan**
Untuk memberikan kontinuitas offset antara konektor sumber, Anda dapat menggunakan topik penyimpanan offset pilihan Anda, bukan topik default. Menentukan topik penyimpanan offset membantu Anda menyelesaikan tugas seperti membuat konektor sumber yang melanjutkan pembacaan dari offset terakhir konektor sebelumnya. Untuk menentukan topik penyimpanan offset, berikan nilai untuk [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)properti dalam konfigurasi pekerja Amazon MSK Connect sebelum membuat konektor.
# Migrasikan konektor sumber ke Amazon MSK Connect
Konektor sumber adalah aplikasi Apache Kafka Connect yang mengimpor catatan dari sistem eksternal ke Kafka. Bagian ini menjelaskan proses migrasi aplikasi konektor sumber Apache Kafka Connect yang menjalankan kluster Kafka Connect lokal atau dikelola sendiri yang berjalan ke Amazon MSK Connect. AWS
Aplikasi konektor sumber Kafka Connect menyimpan offset dalam topik yang diberi nama dengan nilai yang ditetapkan untuk properti config. `offset.storage.topic` Berikut ini adalah contoh pesan offset untuk konektor JDBC yang menjalankan dua tugas yang mengimpor data dari dua tabel berbeda bernama dan. `movies` `shows` Baris terbaru yang diimpor dari film tabel memiliki ID utama`18343`. Baris terbaru yang diimpor dari tabel show memiliki ID utama`732`.
```
["jdbcsource",{"protocol":"1","table":"sample.movies"}] {"incrementing":18343}
["jdbcsource",{"protocol":"1","table":"sample.shows"}] {"incrementing":732}
```
Untuk memigrasikan konektor sumber ke Amazon MSK Connect, lakukan hal berikut:
1. Buat [plugin khusus](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html) Amazon MSK Connect dengan menarik pustaka konektor dari kluster Kafka Connect lokal atau yang dikelola sendiri.
1. Buat [properti pekerja](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-config-provider.html#msk-connect-config-providers-create-custom-config) Amazon MSK Connect dan atur properti `key.converter``value.converter`, dan `offset.storage.topic` ke nilai yang sama yang ditetapkan untuk konektor Kafka yang berjalan di cluster Kafka Connect yang ada.
1. Jeda aplikasi konektor pada cluster yang ada dengan membuat `PUT /connectors/connector-name/pause` permintaan pada cluster Kafka Connect yang ada.
1. Pastikan bahwa semua tugas aplikasi konektor benar-benar dihentikan. Anda dapat menghentikan tugas baik dengan membuat `GET /connectors/connector-name/status` permintaan pada klaster Kafka Connect yang ada atau dengan menggunakan pesan dari nama topik yang ditetapkan untuk properti`status.storage.topic`.
1. Dapatkan konfigurasi konektor dari cluster yang ada. Anda bisa mendapatkan konfigurasi konektor baik dengan membuat `GET /connectors/connector-name/config/` permintaan pada klaster yang ada atau dengan menggunakan pesan dari nama topik yang ditetapkan untuk properti`config.storage.topic`.
1. Buat [Konektor MSK Amazon](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html) baru dengan nama yang sama dengan cluster yang ada. Buat konektor ini dengan menggunakan plugin kustom konektor yang Anda buat di langkah 1, properti pekerja yang Anda buat di langkah 2, dan konfigurasi konektor yang Anda ekstrak pada langkah 5.
1. Saat status Konektor MSK Amazon`active`, lihat log untuk memverifikasi bahwa konektor telah mulai mengimpor data dari sistem sumber.
1. Hapus konektor di cluster yang ada dengan membuat `DELETE /connectors/connector-name` permintaan.
# Migrasikan konektor wastafel ke Amazon MSK Connect
Konektor sink adalah aplikasi Apache Kafka Connect yang mengekspor data dari Kafka ke sistem eksternal. Bagian ini menjelaskan proses migrasi aplikasi konektor sink Apache Kafka Connect yang menjalankan kluster Kafka Connect lokal atau dikelola sendiri yang berjalan ke Amazon MSK Connect. AWS
Konektor sink Kafka Connect menggunakan API keanggotaan grup Kafka dan menyimpan offset dalam `__consumer_offset` topik yang sama dengan aplikasi konsumen biasa. Perilaku ini menyederhanakan migrasi konektor sink dari cluster yang dikelola sendiri ke Amazon MSK Connect.
Untuk memigrasikan konektor sink ke Amazon MSK Connect, lakukan hal berikut:
1. Buat [plugin khusus](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html) Amazon MSK Connect dengan menarik pustaka konektor dari kluster Kafka Connect lokal atau yang dikelola sendiri.
1. Buat [properti pekerja](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-config-provider.html#msk-connect-config-providers-create-custom-config) Amazon MSK Connect dan atur properti `key.converter` dan `value.converter` ke nilai yang sama yang ditetapkan untuk konektor Kafka yang berjalan di cluster Kafka Connect yang ada.
1. Jeda aplikasi konektor pada cluster Anda yang ada dengan membuat `PUT /connectors/connector-name/pause` permintaan pada cluster Kafka Connect yang ada.
1. Pastikan bahwa semua tugas aplikasi konektor benar-benar dihentikan. Anda dapat menghentikan tugas baik dengan membuat `GET /connectors/connector-name/status` permintaan pada klaster Kafka Connect yang ada, atau dengan menggunakan pesan dari nama topik yang ditetapkan untuk properti`status.storage.topic`.
1. Dapatkan konfigurasi konektor dari cluster yang ada. Anda bisa mendapatkan konfigurasi konektor baik dengan membuat `GET /connectors/connector-name/config` permintaan pada klaster yang ada, atau dengan menggunakan pesan dari nama topik yang ditetapkan untuk properti`config.storage.topic`.
1. Buat [Konektor MSK Amazon](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html) baru dengan nama yang sama dengan cluster yang ada. Buat konektor ini dengan menggunakan plugin kustom konektor yang Anda buat di langkah 1, properti pekerja yang Anda buat di langkah 2, dan konfigurasi konektor yang Anda ekstrak pada langkah 5.
1. Saat status Konektor MSK Amazon`active`, lihat log untuk memverifikasi bahwa konektor telah mulai mengimpor data dari sistem sumber.
1. Hapus konektor di cluster yang ada dengan membuat `DELETE /connectors/connector-name` permintaan.
# Memecahkan masalah di Amazon MSK Connect
Informasi berikut dapat membantu Anda memecahkan masalah yang mungkin Anda alami saat menggunakan MSK Connect. Anda juga dapat memposting masalah Anda ke [AWS re:Post](https://repost.aws/).
**Konektor tidak dapat mengakses sumber daya yang dihosting di internet publik**
Lihat [Mengaktifkan akses internet untuk Amazon MSK Connect](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-internet-access.html).
**Jumlah tugas yang berjalan konektor tidak sama dengan jumlah tugas yang ditentukan dalam tasks.max**
Berikut adalah beberapa alasan konektor mungkin menggunakan lebih sedikit tugas daripada konfigurasi tasks.max yang ditentukan:
+ Beberapa implementasi konektor membatasi jumlah tugas yang dapat digunakan. Misalnya, konektor Debezium untuk MySQL terbatas untuk menggunakan satu tugas.
+ Saat menggunakan mode kapasitas berskala otomatis, Amazon MSK Connect mengganti properti tasks.max konektor dengan nilai yang sebanding dengan jumlah pekerja yang berjalan di konektor dan jumlah per pekerja. MCUs Jika Anda telah mengonfigurasi `maxAutoscalingTaskCount` parameter opsional, `tasks.max` nilainya tidak akan melebihi batas ini. Untuk informasi selengkapnya, lihat [Memahami jumlah tugas penskalaan otomatis maksimum](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html#msk-connect-max-autoscaling-task-count).
+ Untuk konektor wastafel, tingkat paralelisme (jumlah tugas) tidak boleh lebih dari jumlah partisi topik. Meskipun Anda dapat mengatur tasks.max lebih besar dari itu, satu partisi tidak pernah diproses oleh lebih dari satu tugas pada satu waktu.
+ Di Kafka Connect 2.7.x, penetapan partisi konsumen default adalah. `RangeAssignor` Perilaku pemberi tugas ini adalah memberikan partisi pertama dari setiap topik kepada satu konsumen, partisi kedua dari setiap topik kepada satu konsumen, dll. Ini berarti bahwa jumlah maksimum tugas aktif untuk konektor wastafel menggunakan `RangeAssignor` sama dengan jumlah maksimum partisi dalam setiap topik yang dikonsumsi. Jika ini tidak berfungsi untuk kasus penggunaan Anda, Anda harus [membuat Konfigurasi Pekerja](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-create-custom-worker-config) di mana `consumer.partition.assignment.strategy` properti disetel ke penugasan partisi konsumen yang lebih sesuai. Lihat [Antarmuka Kafka 2.7 ConsumerPartitionAssignor: *Semua Kelas Penerapan yang Dikenal*](https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.html).