Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Membuat cluster Connect

Cluster Connect menyediakan lingkungan untuk konektor yang membantu memindahkan data dari deployment Kafka yang ada ke cluster Google Cloud Managed Service untuk Apache Kafka atau memindahkan data dari cluster Managed Service untuk Apache Kafka ke layanan Google Cloud lain atau cluster Kafka lain. Cluster Kafka sekunder dapat berupa cluster Google Cloud Managed Service untuk Apache Kafka lainnya, cluster yang dikelola sendiri, atau cluster lokal.

Sebelum memulai

Pastikan Anda telah membuat cluster Managed Service untuk Apache Kafka. Anda memerlukan nama cluster Managed Service untuk Apache Kafka yang akan dihubungkan ke cluster Connect.

Setiap cluster Connect dikaitkan dengan cluster Managed Service untuk Apache Kafka. Cluster ini menyimpan status konektor yang berjalan di cluster Connect.

Peran dan izin yang diperlukan untuk membuat cluster Connect

Untuk mendapatkan izin yang diperlukan guna membuat cluster Connect, minta administrator untuk memberi Anda peran IAM Managed Kafka Connect Cluster Editor (roles/managedkafka.connectClusterEditor) di project Anda. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk membuat cluster Connect. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat cluster Connect:

Berikan izin membuat cluster Connect di lokasi yang ditentukan: managedkafka.connectClusters.create

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Akun utama ACL yang diperlukan

Secara default, cluster Managed Service untuk Apache Kafka mengizinkan Connect Cluster mengakses resource jika tidak ada ACL yang dikonfigurasi. Hal ini dilakukan dengan menyetel allow.everyone.if.no.acl.found ke true, yang merupakan setelan default.

Namun, jika cluster Managed Service untuk Apache Kafka memiliki ACL yang dikonfigurasi, Connect Cluster tidak akan otomatis memiliki izin baca dan tulis ke resource. Anda harus memberikannya secara manual.

Akun layanan cluster Connect yang digunakan sebagai akun utama di ACL mengikuti format ini: User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com.

Jika Anda telah mengonfigurasi ACL di cluster Kafka, berikan izin baca dan izin tulis ke topik serta izin baca ke grup konsumen kepada cluster Connect menggunakan perintah berikut:

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --operation WRITE --topic *

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --group *

Untuk mengetahui informasi selengkapnya tentang perintah ini, lihat Mengonfigurasi ACL Apache Kafka untuk kontrol akses terperinci.

Membuat cluster Connect di project lain

Managed Service untuk Apache Kafka menggunakan agen layanan untuk mengakses resourceGoogle Cloud . Agen layanan dikaitkan dengan project tempat Anda membuat cluster.

Jika Anda membuat cluster Connect di project yang berbeda dengan cluster Managed Service untuk Apache Kafka, cluster Connect dan cluster Kafka akan menggunakan agen layanan yang terkait dengan project masing-masing. Dalam hal ini, agen layanan untuk cluster Connect memerlukan izin untuk mengakses resource Google Cloud di project cluster Kafka.

Untuk memberikan izin yang diperlukan, berikan peran Managed Kafka Service Agent kepada agen layanan cluster Connect di project cluster Kafka. Misalnya, jika Anda membuat cluster Kafka di project kafka-project dan cluster Connect di project connect-project, berikan peran Managed Kafka Service Agent di kafka-project kepada agen layanan yang terkait dengan connect-project.

Alamat email agen layanan memiliki format berikut: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, dengan PROJECT_NUMBER adalah nomor project. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Membuat dan memberikan peran kepada agen layanan.

Properti cluster Connect

Bagian ini menjelaskan properti cluster Connect.

Nama cluster Connect

Nama cluster Connect yang Anda buat. Untuk panduan tentang cara memberi nama cluster Connect, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka. Nama cluster tidak dapat diubah.

Cluster Kafka utama

Cluster Managed Service untuk Apache Kafka yang terkait dengan cluster Connect Anda. Cluster terkait ini (cluster utama) menyimpan status konektor yang berjalan di cluster Connect. Umumnya, cluster Managed Service untuk Apache Kafka utama juga berfungsi sebagai tujuan untuk semua konektor sumber dan input untuk semua konektor sink yang berjalan di cluster Connect.

Satu cluster Managed Service untuk Apache Kafka dapat memiliki beberapa cluster Connect. Jika Anda memilih cluster Managed Service untuk Apache Kafka di project yang berbeda, pastikan izin yang sesuai telah dikonfigurasi.

Anda tidak dapat mengupdate ke cluster Kafka yang berbeda setelah membuat cluster Connect.

Manfaat kolokasi region untuk latensi dan biaya jaringan

Menempatkan cluster Managed Service untuk Apache Kafka dan Connect di region yang sama akan mengurangi latensi dan biaya jaringan. Misalnya, asumsikan cluster Managed Service untuk Apache Kafka Anda berada di region-a dan Anda menggunakan konektor sink untuk menulis data dari cluster Managed Service untuk Apache Kafka (sumber) ini ke Tabel BigQuery (sink) yang juga berada di region-a. Jika Anda men-deploy cluster Connect di region-a, pilihan deployment ini akan meminimalkan latensi untuk operasi tulis BigQuery dan menghilangkan biaya transfer jaringan antar-region antara cluster Managed Service untuk Apache Kafka dan cluster Connect.

Pertimbangan latensi dan biaya multisistem

Kafka Connect menggunakan konektor untuk memindahkan data antar-sistem. Satu sisi konektor selalu berinteraksi dengan cluster Managed Service untuk Apache Kafka. Satu cluster Kafka Connect dapat menjalankan beberapa konektor, yang masing-masing bertindak sebagai sumber (menarik data dari sistem) atau sink (mendorong data ke sistem).

Meskipun cluster Connect di region yang sama dengan cluster Managed Service untuk Apache Kafka mendapatkan manfaat dari latensi komunikasi yang lebih rendah di antara keduanya, setiap konektor juga berinteraksi dengan sistem lain, seperti tabel BigQuery atau cluster Kafka lain. Meskipun cluster Connect dan cluster Managed Service untuk Apache Kafka berada di lokasi yang sama, sistem lain tersebut mungkin berada di region yang berbeda. Hal ini menyebabkan latensi dan biaya yang lebih tinggi. Latensi pipeline secara keseluruhan bergantung pada lokasi ketiga sistem: cluster Managed Service untuk Apache Kafka, cluster Connect, dan sistem sumber atau sink.

Misalnya, jika cluster Managed Service untuk Apache Kafka Anda berada di region-a, cluster Connect Anda di region-b, dan Anda menggunakan konektor Cloud Storage untuk bucket di region-c, Anda akan dikenai biaya untuk dua hop jaringan (region-a ke region-b, lalu region-b ke region-c, atau sebaliknya, bergantung pada arah konektor).

Pertimbangkan dengan cermat semua region yang terlibat saat merencanakan penempatan cluster Connect Anda untuk mengoptimalkan latensi dan biaya.

Konfigurasi kapasitas

Konfigurasi kapasitas mengharuskan Anda mengonfigurasi jumlah vCPU dan jumlah memori untuk setiap vCPU untuk cluster Connect Anda. Anda dapat memperbarui kapasitas cluster Connect setelah membuatnya. Berikut adalah properti untuk konfigurasi kapasitas:

vCPU: Jumlah vCPU yang ditetapkan ke cluster Connect. Nilai minimumnya adalah 3 vCPU.
Memori: Jumlah memori yang ditetapkan untuk setiap vCPU. Anda harus menyediakan antara 1 GiB dan 8 GiB per vCPU. Jumlah memori dapat ditambah atau dikurangi dalam batas ini setelah cluster dibuat.

Misalnya, jika Anda membuat cluster dengan 6 vCPU, memori minimum yang dapat Anda alokasikan ke cluster adalah 6 GiB (1 GiB per vCPU), dan maksimumnya adalah 48 GiB (8 GiB per vCPU).

vCPU dan memori yang dialokasikan untuk setiap worker di cluster Connect memiliki dampak signifikan pada performa, kapasitas, dan biaya cluster. Berikut adalah penjelasan tentang pengaruh vCPU dan memori terhadap cluster Connect.

Jumlah vCPU

Kafka Connect membagi pekerjaan konektor menjadi beberapa tugas. Setiap tugas dapat memproses data secara paralel. Lebih banyak vCPU berarti lebih banyak tugas yang dapat dijalankan secara bersamaan, sehingga menghasilkan throughput yang lebih tinggi.
Lebih banyak vCPU akan meningkatkan biaya untuk cluster Connect Anda.

Memori

Kafka Connect menggunakan memori untuk melakukan buffering data saat data mengalir antara konektor dan Managed Service untuk Apache Kafka. Memori yang lebih besar memungkinkan buffer yang lebih besar. Memori yang besar dapat meningkatkan throughput, terutama untuk aliran data bervolume tinggi. Konektor yang menangani pesan atau data yang sangat besar memerlukan memori yang cukup untuk memprosesnya tanpa mengalami pengecualian OutOfMemoryError.
Memori yang lebih besar akan meningkatkan biaya cluster Connect Anda.
Jika menggunakan logika transformasi yang berat, Anda memerlukan alokasi memori yang lebih besar.

Tujuan Anda adalah memilih konfigurasi kapasitas yang tepat untuk cluster Connect Anda. Untuk melakukannya, Anda harus memahami throughput yang dapat ditangani oleh cluster Connect Anda.

Subnet worker (utama)

Subnet worker, yang juga dikenal sebagai subnet utama, menghubungkan jaringan VPC Anda ke cluster Connect. Subnet ini memungkinkan worker cluster menjangkau endpoint sumber dan sink di jaringan konsumen, seperti cluster Managed Service untuk Apache Kafka atau cluster Kafka yang dihosting sendiri.

Berikut beberapa persyaratan untuk mengonfigurasi subnet worker:

Subnet worker diperlukan.
Subnet harus berada di region yang sama dengan cluster Connect.
Subnet harus berada di VPC induk yang sama dengan salah satu daftar subnet yang terhubung dari cluster Kafka utama.
Rentang CIDR subnet harus memiliki ukuran minimum /22 (1024 alamat).

Worker cluster diberi alamat IP di subnet worker, menggunakan antarmuka Private Service Connect. Worker dapat menjangkau tujuan jaringan apa pun yang dapat diakses dari jaringan VPC subnet, dengan persyaratan berikut:

Endpoint tidak boleh berada dalam rentang CIDR 172.16.0.0/14. Rentang ini dicadangkan untuk penggunaan internal Managed Service untuk Apache Kafka Connect.
Aturan firewall harus mengizinkan traffic. Lihat Mengonfigurasi keamanan untuk lampiran jaringan.
Untuk traffic internet, Anda harus mengonfigurasi Cloud NAT. Misalnya, Cloud NAT diperlukan agar konektor MirrorMaker dapat mereplikasi data dari cluster Kafka yang dapat diakses melalui internet.
Untuk mengakses endpoint Private Service Connect yang berada di VPC yang berbeda dengan VPC subnet worker, Anda harus memastikan bahwa Anda menggunakan konfigurasi konsumen yang didukung (misalnya, NCC). Untuk mengetahui informasi selengkapnya, lihat Tentang mengakses layanan yang dipublikasikan melalui endpoint.

Domain DNS yang dapat di-resolve

Domain DNS yang dapat di-resolve, juga dikenal sebagai nama domain DNS, memungkinkan alamat DNS di jaringan VPC konsumen tersedia untuk VPC tenant. Hal ini memungkinkan cluster Connect me-resolve nama DNS ke alamat IP, sehingga memfasilitasi komunikasi dengan layanan lain, termasuk cluster Kafka lain untuk konektor MirrorMaker.

Untuk domain DNS yang dapat di-resolve, Anda dapat memilih cluster Managed Service untuk Apache Kafka. Anda tidak perlu mengonfigurasi nama domain DNS untuk cluster Managed Service untuk Apache Kafka utama, karena alamat bootstrap-nya otomatis disertakan dalam daftar domain DNS yang dapat di-resolve.

Namun, Anda juga dapat menentukan domain DNS secara manual, yang diperlukan jika Anda memilih cluster Kafka eksternal. Domain DNS cluster Managed Service untuk Apache Kafka utama otomatis disertakan. Cluster Kafka lainnya masih memerlukan konfigurasi domain DNS.

Resource Secret Manager

Beberapa konektor memerlukan data sensitif, seperti sandi, sebagai bagian dari konfigurasinya. Untuk mengelola jenis data ini dengan aman, Anda dapat menyimpan data di Secret Manager dan memberi cluster Connect akses ke secret.

Untuk menggunakan secret Secret Manager dengan Kafka Connect, lakukan hal berikut:

Berikan peran IAM berikut ke Akun Layanan Managed Kafka. Peran ini memungkinkan cluster Connect Anda melihat secret dan mengakses payload-nya.
- Secret Manager Viewer (roles/secretmanager.viewer)
- Secret Manager Secret Accessor (roles/secretmanager.secretAccessor)
Buat secret yang berisi data sensitif. Untuk mengetahui informasi selengkapnya, lihat Membuat secret.
Saat Anda membuat atau mengupdate cluster Connect, tentukan secret yang dapat diakses oleh cluster. Anda dapat menentukan hingga 32 secret per cluster Connect.

Secret dipasang sebagai file di worker cluster. Konektor memiliki akses hanya baca ke file ini. Saat Anda membuat konektor, properti konfigurasi konektor dapat mereferensikan secret.

Untuk mereferensikan jalur ke file secret, gunakan format berikut:
```
/var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION
```
Contoh: ssl.truststore.location=/var/secrets/project1-truststore-1
Untuk menggunakan value secret sebagai nilai konfigurasi (misalnya, sandi), gunakan format berikut:
```
${directory:/var/secrets:PROJECT_NAME-SECRET_NAME-SECRET_VERSION}
```
Contoh: password=${directory:/var/secrets:project1-database_password-3}

Ganti kode berikut:

PROJECT_NAME: Nama project Google Cloud .
SECRET_NAME: Nama secret.
SECRET_VERSION: Versi secret.

Label

Label adalah pasangan nilai kunci yang membantu Anda dalam pengaturan dan identifikasi. Fitur ini membantu Anda mengatur cluster Connect. Anda dapat melampirkan label ke setiap cluster Connect, lalu memfilter resource berdasarkan labelnya. Contoh label adalah environment:prod, application:web-app.

Membuat cluster Connect

Sebelum membuat cluster, tinjau dokumentasi untuk Properti cluster Connect.

Pembuatan cluster Connect memerlukan waktu 20 hingga 30 menit.

Konsol

Di konsol Google Cloud , buka halaman Connect Clusters.

Buka Connect Clusters
Klik Create .
Di kolom Connect cluster name, masukkan nama untuk cluster Connect. Untuk mengetahui informasi selengkapnya, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.
Dalam daftar Primary Kafka cluster, pilih cluster Managed Service untuk Apache Kafka. Untuk mengetahui informasi selengkapnya, lihat Cluster Kafka utama.
Dalam daftar Region, pilih lokasi untuk cluster Connect. Untuk mengetahui informasi selengkapnya tentang cara memilih lokasi, lihat Cluster Kafka utama.
Di bagian Capacity configuration, masukkan nilai untuk kolom berikut atau pertahankan nilai default.
- Di kolom vCPUs, masukkan jumlah CPU virtual untuk cluster.
- Di kolom Memory, masukkan jumlah memori per CPU, dalam GiB. Nilai tidak boleh melebihi 8 GiB per CPU.
Untuk mengetahui informasi selengkapnya, lihat Konfigurasi kapasitas.
Di bagian Network configuration, pilih jaringan VPC dari daftar Network atau biarkan nilai default. Daftar ini diisi saat Anda memilih cluster Kafka utama.
Di bagian Worker subnet, pilih subnet dari daftar Subnet atau biarkan nilai default. Untuk mengetahui informasi selengkapnya, lihat Subnet worker. Kolom Subnet URI path akan otomatis diisi saat Anda memilih subnet.
Opsional: Tambahkan domain DNS yang dapat di-resolve. Domain DNS cluster Kafka utama otomatis ditambahkan sebagai domain DNS yang dapat di-resolve. Untuk menentukan domain DNS tambahan, lakukan hal berikut:
1. Luaskan bagian Resolvable DNS domains.
2. Klik Add DNS domain.
3. Untuk menambahkan domain DNS dari cluster Managed Service untuk Apache Kafka yang ada, pilih cluster dari daftar Kafka cluster. Atau, masukkan domain DNS di kolom DNS domain.
4. Klik Done.
Opsional: Untuk menambahkan resource Secret Manager, lakukan hal berikut:
1. Luaskan bagian Secret manager resources.
2. Klik Add secret resource.
3. Dalam daftar Secret, pilih secret.
4. Dalam daftar Secret version, pilih versi secret.
5. Klik Done.
Opsional: Tambahkan label untuk mengatur project Anda. Untuk menambahkan label, lakukan langkah-langkah berikut:
1. Luaskan bagian Label.
2. Klik Tambahkan label.
3. Di kolom Key, masukkan kunci label.
4. Di kolom Value, masukkan nilai label.
Klik Create.

gcloud

Di konsol Google Cloud , aktifkan Cloud Shell.

Aktifkan Cloud Shell

Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.
Jalankan perintah gcloud managed-kafka connect-clusters create:
```
gcloud managed-kafka connect-clusters create CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU \
    --memory=MEMORY \
    --primary-subnet=WORKER_SUBNET \
    --kafka-cluster=KAFKA_CLUSTER \
    [--project=PROJECT_ID] \
    [--secret=SECRET] \
    [--dns-name=DNS_DOMAIN_NAME] \
    [--config-file=CONFIG_FILE] \
    [--labels=LABELS]
    [--async]
```
Ganti kode berikut:
- CONNECT_CLUSTER_ID: ID atau nama cluster Connect. Untuk panduan tentang cara memberi nama cluster Connect, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka. Nama cluster Connect tidak dapat diubah.
- LOCATION: Lokasi tempat Anda membuat cluster Connect. Region ini harus berupa region Google Cloudyang didukung. Anda tidak dapat mengubah lokasi cluster Connect setelah dibuat. Untuk mengetahui daftar lokasi yang tersedia, lihat Lokasi Managed Service untuk Apache Kafka. Untuk mengetahui informasi selengkapnya tentang rekomendasi lokasi, lihat Cluster Kafka utama.
- CPU: Jumlah vCPU untuk cluster Connect. Nilai minimumnya adalah 3 vCPU. Lihat jumlah vCPU.
- MEMORY: Jumlah memori untuk cluster Connect. Gunakan unit "MB", "MiB", "GB", "GiB", "TB", atau "TiB". Misalnya, "3GiB". Anda harus menyediakan antara 1 GiB dan 8 GiB per vCPU. Lihat Memori.
- WORKER_SUBNET: Subnet worker untuk cluster Connect.
  
  Format subnet adalah projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.
  
  Subnet worker harus berada di region yang sama dengan cluster Connect.
- PROJECT_ID: (Opsional) ID projectGoogle Cloud . Jika tidak diberikan, project saat ini akan digunakan.
- KAFKA_CLUSTER: ID atau nama yang sepenuhnya memenuhi syarat dari cluster Managed Service untuk Apache Kafka utama yang terkait dengan cluster Connect. Lihat Cluster Kafka. Format cluster Kafka adalah projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID.
  
  Anda tidak dapat mengupdate ke cluster Kafka yang berbeda setelah membuat cluster Connect.
- SECRET: (Opsional) Secret yang akan dimuat ke worker. Versi Secret yang tepat dari Secret Manager harus disediakan, alias tidak didukung. Hingga 32 secret dapat dimuat ke dalam satu cluster. Format: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID
- DNS_DOMAIN_NAME: (Opsional) Nama domain DNS dari subnet yang akan ditampilkan ke Connect Cluster. Cluster Connect dapat mengakses resource menggunakan nama domain, bukan mengandalkan alamat IP. Lihat Peering DNS.
- LABELS: (Opsional) Label yang akan dikaitkan dengan cluster. Untuk mengetahui informasi selengkapnya tentang format label, lihat Label. Daftar pasangan KEY=VALUE label yang akan ditambahkan. Kunci harus diawali dengan huruf kecil dan hanya berisi tanda hubung (-), garis bawah (_), huruf kecil, dan angka. Nilai hanya boleh berisi tanda hubung (-), garis bawah (_), huruf kecil, dan angka.
- CONFIG_FILE: (Opsional) Jalur ke file JSON atau YAML yang berisi konfigurasi yang diganti dari default cluster atau konektor. File ini juga mendukung JSON atau YAML inline.
- --async: (Opsional) Segera kembali, tanpa menunggu operasi yang sedang berlangsung selesai. Dengan flag --async, Anda dapat melanjutkan tugas lain saat pembuatan cluster terjadi di latar belakang. Jika Anda tidak menggunakan flag, sistem akan menunggu operasi selesai sebelum menampilkan respons. Anda harus menunggu hingga cluster diupdate sepenuhnya sebelum dapat melanjutkan tugas lainnya.
Anda akan mendapatkan respons seperti berikut:
```
Create request issued for: [sample-connectcluster]
Check operation [projects/test-project/locations/us-east1/operations/operation-1753590328249-63ae19098cc06-64300a0a-06512d02] for status.
```
Simpan OPERATION_ID untuk melacak progres. Misalnya, nilai di sini adalah operation-1753590328249-63ae19098cc06-64300a0a-06512d02.

Terraform

Anda dapat menggunakan resource Terraform untuk membuat cluster Connect.

resource "google_managed_kafka_connect_cluster" "default" {
  provider           = google-beta
  project            = data.google_project.default.project_id
  connect_cluster_id = "my-connect-cluster-id"
  location           = "us-central1"
  kafka_cluster      = google_managed_kafka_cluster.default.id
  capacity_config {
    vcpu_count   = 12
    memory_bytes = 12884901888 # 12 GiB
  }
  gcp_config {
    access_config {
      network_configs {
        primary_subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

Untuk mempelajari cara menerapkan atau menghapus konfigurasi Terraform, lihat Perintah dasar Terraform.

Go

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Go API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi(ADC). Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func createConnectCluster(w io.Writer, projectID, region, clusterID, kafkaCluster string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-connect-cluster"
	// kafkaCluster := "projects/my-project-id/locations/us-central1/clusters/my-kafka-cluster"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/connectClusters/%s", locationPath, clusterID)

	// Capacity configuration with 12 vCPU and 12 GiB RAM
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   12,
		MemoryBytes: 12884901888, // 12 GiB in bytes
	}

	// Optionally, you can also specify accessible subnets and resolvable DNS
	// domains as part of your network configuration. For example:
	// networkConfigs := []*managedkafkapb.ConnectNetworkConfig{
	// 	{
	// 		PrimarySubnet:      primarySubnet,
	// 		AdditionalSubnets:  []string{"subnet-1", "subnet-2"},
	// 		DnsDomainNames:     []string{"domain-1", "domain-2"},
	// 	},
	// }

	connectCluster := &managedkafkapb.ConnectCluster{
		Name:           clusterPath,
		KafkaCluster:   kafkaCluster,
		CapacityConfig: capacityConfig,
	}

	req := &managedkafkapb.CreateConnectClusterRequest{
		Parent:           locationPath,
		ConnectClusterId: clusterID,
		ConnectCluster:   connectCluster,
	}
	op, err := client.CreateConnectCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnectCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 5-15 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created connect cluster: %s\n", resp.Name)
	return nil
}

Java

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Java API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.


import com.google.api.gax.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.ConnectAccessConfig;
import com.google.cloud.managedkafka.v1.ConnectCluster;
import com.google.cloud.managedkafka.v1.ConnectGcpConfig;
import com.google.cloud.managedkafka.v1.ConnectNetworkConfig;
import com.google.cloud.managedkafka.v1.CreateConnectClusterRequest;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateConnectCluster {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String clusterId = "my-connect-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    String kafkaCluster = "my-kafka-cluster"; // The Kafka cluster to connect to
    int cpu = 12;
    long memoryBytes = 12884901888L; // 12 GiB
    createConnectCluster(projectId, region, clusterId, subnet, kafkaCluster, cpu, memoryBytes);
  }

  public static void createConnectCluster(
      String projectId,
      String region,
      String clusterId,
      String subnet,
      String kafkaCluster,
      int cpu,
      long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setVcpuCount(cpu)
        .setMemoryBytes(memoryBytes).build();
    ConnectNetworkConfig networkConfig = ConnectNetworkConfig.newBuilder()
        .setPrimarySubnet(subnet)
        .build();
    // Optionally, you can also specify additional accessible subnets and resolvable
    // DNS domains as part of your network configuration. For example:
    // .addAllAdditionalSubnets(List.of("subnet-1", "subnet-2"))
    // .addAllDnsDomainNames(List.of("dns-1", "dns-2"))
    ConnectGcpConfig gcpConfig = ConnectGcpConfig.newBuilder()
        .setAccessConfig(ConnectAccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
        .build();
    ConnectCluster connectCluster = ConnectCluster.newBuilder()
        .setCapacityConfig(capacityConfig)
        .setGcpConfig(gcpConfig)
        .setKafkaCluster(kafkaCluster)
        .build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.createConnectClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient
        .create(settingsBuilder.build())) {
      CreateConnectClusterRequest request = CreateConnectClusterRequest.newBuilder()
          .setParent(LocationName.of(projectId, region).toString())
          .setConnectClusterId(clusterId)
          .setConnectCluster(connectCluster)
          .build();

      // The duration of this operation can vary considerably, typically taking
      // between 10-30 minutes.
      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
          .createConnectClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf(
          "Connect cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(), operation.isDone(), future.getMetadata().get().toString());

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone
      // = True)
      ConnectCluster response = future.get();
      System.out.printf("Created connect cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaConnectClient.createConnectCluster got err: %s\n", 
          e.getMessage());
      throw e;
    }
  }
}

Python

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Python API Managed Service untuk Apache Kafka.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.cloud.managedkafka_v1.services.managed_kafka_connect import ManagedKafkaConnectClient
from google.cloud.managedkafka_v1.types import ConnectCluster, CreateConnectClusterRequest, ConnectNetworkConfig

# TODO(developer): Update with your values.
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"
# kafka_cluster_id = "my-kafka-cluster"
# primary_subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 12
# memory_bytes = 12884901888  # 12 GiB

connect_client = ManagedKafkaConnectClient()
kafka_client = managedkafka_v1.ManagedKafkaClient()

parent = connect_client.common_location_path(project_id, region)
kafka_cluster_path = kafka_client.cluster_path(project_id, region, kafka_cluster_id)

connect_cluster = ConnectCluster()
connect_cluster.name = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)
connect_cluster.kafka_cluster = kafka_cluster_path
connect_cluster.capacity_config.vcpu_count = cpu
connect_cluster.capacity_config.memory_bytes = memory_bytes
connect_cluster.gcp_config.access_config.network_configs = [ConnectNetworkConfig(primary_subnet=primary_subnet)]
# Optionally, you can also specify accessible subnets and resolvable DNS domains as part of your network configuration.
# For example:
# connect_cluster.gcp_config.access_config.network_configs = [
#     ConnectNetworkConfig(
#         primary_subnet=primary_subnet,
#         additional_subnets=additional_subnets,
#         dns_domain_names=dns_domain_names,
#     )
# ]

request = CreateConnectClusterRequest(
    parent=parent,
    connect_cluster_id=connect_cluster_id,
    connect_cluster=connect_cluster,
)

try:
    operation = connect_client.create_connect_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # Creating a Connect cluster can take 10-40 minutes.
    response = operation.result(timeout=3000)
    print("Created Connect cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

Memantau operasi pembuatan cluster

Anda dapat menjalankan perintah berikut hanya jika Anda menjalankan gcloud CLI untuk membuat cluster Connect.

Pembuatan cluster Connect biasanya memerlukan waktu 20-30 menit. Untuk melacak progres pembuatan cluster, perintah gcloud managed-kafka connect-clusters create menggunakan operasi yang berjalan lama (LRO), yang dapat Anda pantau menggunakan perintah berikut:
```
gcloud managed-kafka operations describe OPERATION_ID \
    --location=LOCATION
```
Ganti kode berikut:
- OPERATION_ID dengan nilai ID operasi dari bagian sebelumnya.
- LOCATION dengan nilai lokasi dari bagian sebelumnya.

Apa langkah selanjutnya?

Apache Kafka® adalah merek dagang terdaftar dari Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lainnya.

Membuat cluster Connect Tetap teratur dengan koleksi Simpan dan kategorikan konten berdasarkan preferensi Anda.

Sebelum memulai

Peran dan izin yang diperlukan untuk membuat cluster Connect

Izin yang diperlukan

Akun utama ACL yang diperlukan

Membuat cluster Connect di project lain

Properti cluster Connect

Nama cluster Connect

Cluster Kafka utama

Manfaat kolokasi region untuk latensi dan biaya jaringan

Pertimbangan latensi dan biaya multisistem

Konfigurasi kapasitas

Jumlah vCPU

Memori

Subnet worker (utama)

Domain DNS yang dapat di-resolve

Resource Secret Manager

Label

Membuat cluster Connect

Konsol

gcloud

Terraform

Go

Java

Python

Memantau operasi pembuatan cluster

Apa langkah selanjutnya?

Membuat cluster Connect