Membuat cluster Google Cloud Managed Service for Apache Kafka

Cluster Managed Service untuk Apache Kafka menyediakan lingkungan untuk menyimpan dan memproses aliran pesan yang disusun ke dalam topik.

Untuk membuat cluster, Anda dapat menggunakan konsol Google Cloud , Google Cloud CLI, library klien, atau Managed Kafka API. Anda tidak dapat menggunakan API Apache Kafka open source untuk membuat cluster.

Sebelum memulai

Pastikan Anda memahami hal-hal berikut:

• Cara kerja Managed Service untuk Apache Kafka.

• Konsep jaringan yang terkait dengan Managed Service untuk Apache Kafka.

Peran dan izin yang diperlukan untuk membuat cluster

Untuk mendapatkan izin yang diperlukan guna membuat cluster, minta administrator untuk memberi Anda peran IAM Managed Kafka Cluster Editor (roles/managedkafka.clusterEditor) 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. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat cluster:

• Buat cluster: managedkafka.clusters.create

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

Properti cluster Managed Service for Apache Kafka

Saat membuat atau memperbarui cluster Managed Service untuk Apache Kafka, Anda harus menentukan properti berikut.

Nama cluster

Nama atau ID cluster Managed Service untuk Apache Kafka yang Anda buat. Untuk pedoman tentang cara memberi nama cluster, lihat Pedoman untuk memberi nama resource Managed Service untuk Apache Kafka. Nama cluster tidak dapat diubah.

Lokasi

Lokasi tempat Anda membuat cluster. Lokasi harus berada di salah satu region Google Cloud yang didukung. Lokasi cluster tidak dapat diubah nanti. Untuk mengetahui daftar lokasi yang tersedia, lihat lokasi Managed Service for Apache Kafka.

Konfigurasi kapasitas

Kapasitas komputasi, memori, dan penyimpanan cluster Kafka. Saat membuat cluster, Anda harus menentukan nilai berikut:

• vCPUs: Jumlah vCPU dalam cluster. Diperlukan minimal 3 vCPU per cluster.

• Memori: Jumlah memori yang ditetapkan ke cluster. Anda harus menyediakan antara 1 GiB dan 8 GiB per vCPU.

  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).

Layanan ini otomatis menyediakan broker, termasuk volume Persistent Disk untuk broker. Secara opsional, Anda dapat mengaktifkan Penyimpanan lokal kustom, yang memungkinkan Anda menentukan kapasitas disk untuk setiap broker. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi ukuran disk broker.

Untuk mengetahui panduan tentang memperkirakan kapasitas yang dibutuhkan cluster Anda, lihat Merencanakan ukuran cluster Kafka.

Konfigurasi jaringan

Konfigurasi jaringan adalah daftar subnet VPC tempat cluster dapat diakses. Untuk membuat atau menggunakan pesan, klien harus dapat menjangkau salah satu subnet ini.

Berikut adalah beberapa panduan untuk konfigurasi jaringan Anda:

• Minimal 1 subnet diperlukan untuk cluster. Maksimumnya adalah 10.

• Hanya 1 subnet per jaringan yang diizinkan untuk cluster tertentu.

• Setiap subnet harus berada di region yang sama dengan cluster. Project dan jaringan dapat berbeda.

• Alamat IP untuk broker dan server bootstrap dialokasikan secara otomatis di setiap subnet. Selain itu, entri DNS untuk alamat IP ini dibuat di jaringan VPC yang sesuai.

• Jika menambahkan subnet dari project lain, Anda harus memberikan izin ke akun layanan yang dikelola Google yang terkait dengan cluster. Untuk informasi selengkapnya, lihat Menghubungkan cluster di seluruh project.

Setelah membuat cluster, Anda dapat memperbarui daftar subnet. Untuk mengetahui informasi selengkapnya tentang jaringan, lihat Mengonfigurasi jaringan untuk Managed Service untuk Apache Kafka.

Cluster publik

Secara opsional, Anda dapat mengonfigurasi cluster sebagai cluster publik untuk memungkinkan aplikasi klien terhubung ke cluster Anda dan menjalankan operasi Kafka (termasuk mengirim dan menerima pesan) melalui internet publik. Untuk mengaktifkan akses publik, Anda harus menentukan satu atau beberapa rentang IP sumber yang diizinkan.

Saat Anda mengonfigurasi rentang IP sumber yang diizinkan, aturan berikut berlaku:

• Semua rentang IP sumber yang diizinkan harus ditentukan dalam notasi CIDR IPv4 (misalnya, 203.0.113.0/24 atau 198.51.100.5/32).

• Setiap ukuran subnet CIDR harus antara /16 dan /32.

• Rentang CIDR tidak boleh tumpang-tindih.

• Semua rentang IP sumber yang diizinkan harus dapat dirutekan secara publik. Rentang IP pribadi, seperti 10.0.0.0/8 (RFC 1918), ditolak.

• Anda dapat menentukan maksimum 500 rentang IP sumber yang diizinkan.

• Alamat IPv6 tidak didukung.

Perhatian: Untuk melindungi cluster Anda, izinkan akses hanya dari rentang IP eksternal tepercaya. Jangan mengekspos cluster Anda ke rentang yang tidak tepercaya.

Sebelum mengaktifkan akses publik untuk cluster, sebaiknya baca artikel Menghubungkan klien ke cluster publik untuk mempelajari cara penerapan fitur akses publik.

Label

Label adalah pasangan nilai kunci yang membantu Anda dalam mengatur dan mengidentifikasi. Label memungkinkan Anda mengategorikan resource berdasarkan lingkungan. Contohnya adalah "env:production" dan "owner:data-engineering".

Anda dapat memfilter dan menelusuri resource berdasarkan labelnya. Misalnya, asumsikan Anda memiliki beberapa cluster Managed Service untuk Apache Kafka bagi departemen yang berbeda. Anda dapat mengonfigurasi dan menelusuri cluster dengan label "department:marketing" untuk menemukan cluster yang relevan.

Konfigurasi penyeimbangan ulang

Setelan ini menentukan apakah layanan secara otomatis menyeimbangkan ulang replika partisi di seluruh broker.

Mode yang tersedia adalah:

• Penyeimbangan ulang otomatis saat penskalaan: Jika opsi ini diaktifkan, layanan akan otomatis memicu penyeimbangan ulang replika saat Anda melakukan penskalaan cluster. Mode ini membantu mempertahankan distribusi beban yang merata, tetapi mungkin mempengaruhi performa untuk sementara selama operasi penyeimbangan ulang.

• No rebalance: Jika opsi ini diaktifkan, layanan tidak akan menyeimbangkan ulang replika secara otomatis.

Untuk mengetahui informasi selengkapnya, lihat Menyeimbangkan ulang partisi broker.

Enkripsi

Managed Service for Apache Kafka dapat mengenkripsi pesan dengan Google-owned and Google-managed encryption keys (default) atau Kunci enkripsi yang dikelola pelanggan (CMEK). Setiap pesan dienkripsi saat dalam penyimpanan dan pengiriman. Jenis enkripsi untuk cluster tidak dapat diubah.

Google-owned and Google-managed encryption keys digunakan secara default. Kunci ini dibuat, dikelola, dan disimpan sepenuhnya oleh Google Cloud dalam infrastrukturnya.

CMEK adalah kunci enkripsi yang Anda kelola menggunakan Cloud Key Management Service. Fitur ini memungkinkan Anda memiliki kontrol yang lebih besar atas kunci yang digunakan untuk mengenkripsi data dalam penyimpanan di layanan Google Cloud yang didukung. Penggunaan CMEK akan menimbulkan biaya tambahan terkait Cloud Key Management Service. Untuk penggunaan CMEK, key ring Anda harus berada di lokasi yang sama dengan resource yang Anda gunakan. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi enkripsi pesan.

Konfigurasi mTLS

Secara opsional, Anda dapat mengonfigurasi mTLS sebagai metode autentikasi alternatif yang menggunakan sertifikat klien. Konfigurasi ini mencakup hal berikut:

• Kumpulan CA: Daftar 1 hingga 10 kumpulan Certificate Authority Service (CAS) yang dipercaya oleh cluster untuk autentikasi klien.

• Aturan pemetaan Principal SSL: Properti broker ssl.principal.mapping.rules yang bersifat opsional, tetapi direkomendasikan untuk menyederhanakan nama principal sertifikat yang panjang untuk digunakan dalam ACL Kafka.

Untuk mengetahui informasi selengkapnya tentang mTLS, lihat Mengonfigurasi autentikasi mTLS.

Membuat cluster

Sebelum membuat cluster, baca dokumentasi tentang properti cluster.

Pembuatan cluster biasanya memerlukan waktu 20-30 menit.

Untuk membuat cluster, ikuti langkah-langkah berikut:

Konsol

1. Di konsol Google Cloud , buka halaman Clusters.

   Buka Cluster

2. Pilih Create.

   Halaman Create Kafka cluster akan terbuka.

3. Untuk Cluster name, masukkan string.

   Untuk mengetahui informasi selengkapnya tentang cara memberi nama cluster, lihat Pedoman untuk memberi nama resource Managed Service for Apache Kafka.

4. Untuk Location, masukkan lokasi yang didukung.

   Untuk mengetahui informasi selengkapnya tentang lokasi yang didukung, lihat Lokasi Managed Service untuk Apache Kafka yang didukung.

5. Untuk Konfigurasi kapasitas, masukkan nilai untuk Memori dan vCPU.

   Untuk mengetahui informasi selengkapnya tentang cara menentukan ukuran cluster, lihat Merencanakan ukuran cluster Kafka.

6. Opsional. Untuk menentukan kapasitas disk broker, lakukan hal berikut:

   1. Klik tombol Aktifkan ukuran penyimpanan lokal kustom ke posisi aktif.

      Penting: Setelah mengklik Simpan, Anda tidak dapat menonaktifkan setelan ini lagi untuk cluster ini.

   2. Di kolom Custom local storage, masukkan ukuran disk per broker, dalam GiB.

   Jika Anda membiarkan tombol Aktifkan ukuran penyimpanan lokal kustom dalam posisi nonaktif, kapasitas disk default akan disediakan. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi ukuran disk broker.

7. Untuk Network configuration, masukkan detail berikut:

   1. Project: Project tempat subnetwork berada. Subnet harus berada di region yang sama dengan cluster, tetapi projectnya mungkin berbeda.
   2. Jaringan: Jaringan tempat subnet terhubung.
   3. Subnetwork: Nama subnet.
   4. Jalur URI subnet: Kolom ini diisi secara otomatis. Atau, Anda dapat memasukkan jalur subnet di sini. Nama subnet harus dalam format: projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.
   5. Klik Done.

8. Opsional: Untuk menambahkan subnet tambahan, klik Tambahkan subnet yang terhubung.

   Anda dapat menambahkan subnet tambahan, hingga nilai maksimum 10.

9. Opsional: Aktifkan akses publik ke cluster Anda.

   1. Di bagian Public cluster configuration, di bagian Cluster can be reached, pilih In a private network and using public internet.

   2. Di kolom Daftar rentang IPv4 sumber yang diizinkan yang dipisahkan koma dalam notasi CIDR, masukkan rentang IP aplikasi klien Anda dalam notasi CIDR IPv4 (misalnya, 203.0.113.0/24). Alamat IPv6 tidak didukung.

10. Opsional: Tetapkan kunci enkripsi yang dikelola pelanggan (CMEK).

    1. Di bagian Encryption, pilih Cloud KMS key.

    2. Untuk Jenis kunci, pilih Cloud KMS.

    3. Untuk Pilih kunci yang dikelola pelanggan, pilih atau masukkan CMEK.

       Jika Anda tidak melihat opsi ini, pastikan Anda telah mengaktifkan Cloud KMS API untuk project tersebut.

    Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi enkripsi pesan.

11. Klik Create.

gcloud

1. Di konsol Google Cloud , aktifkan Cloud Shell.

   Aktifkan Cloud Shell

   Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan perintah 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.

2. Sebelum menggunakan salah satu data perintah di bawah, lakukan penggantian berikut:

   • PROJECT_ID: project ID.
   • LOCATION: lokasi cluster.
   • CLUSTER_ID: ID cluster.
   • CPU_COUNT: jumlah vCPU untuk cluster.
   • MEMORY: jumlah memori untuk cluster. Contoh: 3GiB.
   • SUBNET_ID: ID subnet yang akan dihubungkan. Contoh: default.
   • ENCRYPTION_KEY: ID kunci enkripsi yang dikelola pelanggan yang akan digunakan untuk cluster. Tindakan tersebut bersifat opsional.
   • LABELS: label yang akan dikaitkan dengan cluster.
   • ALLOWED_SOURCE_IP_RANGES: rentang IP sumber yang diizinkan untuk cluster publik. Tindakan tersebut bersifat opsional.

   Jalankan perintah berikut:

   Linux, macOS, atau Cloud Shell

   Catatan: Pastikan Anda telah menginisialisasi Google Cloud CLI dengan autentikasi dan sebuah project dengan menjalankan gcloud init; atau gcloud auth login dan gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU_COUNT \
       --memory=MEMORY \
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
       --auto-rebalance \
       --encryption-key=ENCRYPTION_KEY \
       --async \
       --labels=LABELS \
       --public-cluster \
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows (PowerShell)

   Catatan: Pastikan Anda telah menginisialisasi Google Cloud CLI dengan autentikasi dan sebuah project dengan menjalankan gcloud init; atau gcloud auth login dan gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID `
       --location=LOCATION `
       --cpu=CPU_COUNT `
       --memory=MEMORY `
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
       --auto-rebalance `
       --encryption-key=ENCRYPTION_KEY `
       --async `
       --labels=LABELS `
       --public-cluster `
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows (cmd.exe)

   Catatan: Pastikan Anda telah menginisialisasi Google Cloud CLI melalui autentikasi dan sebuah project dengan menjalankan gcloud init; atau gcloud auth login dan gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID ^
       --location=LOCATION ^
       --cpu=CPU_COUNT ^
       --memory=MEMORY ^
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
       --auto-rebalance ^
       --encryption-key=ENCRYPTION_KEY ^
       --async ^
       --labels=LABELS ^
       --public-cluster ^
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Anda akan melihat respons seperti berikut:

   done: false
   metadata:
     '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
     apiVersion: v1
     createTime: 'CREATE_TIME'
     requestedCancellation: false
     target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
     verb: create
   name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
   

   Simpan OPERATION_ID untuk melacak progress.

REST

Sebelum menggunakan salah satu data permintaan, lakukan penggantian berikut:

• PROJECT_ID: Project ID Google Cloud Anda
• LOCATION: lokasi cluster.
• CLUSTER_ID: ID cluster.
• CPU_COUNT: jumlah vCPU untuk cluster.
• MEMORY: jumlah memori untuk cluster, dalam byte. Contoh: 3221225472.
• SUBNET_ID: ID subnet yang akan dihubungkan. Contoh: default.

Metode HTTP dan URL:

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID

Meminta isi JSON:

{
  "capacityConfig": {
    "vcpuCount": CPU_COUNT,
    "memoryBytes": MEMORY
  },
  "gcpConfig": {
    "accessConfig": {
      "networkConfigs": [
        {
          "subnet": "projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID"
        }
      ]
    }
  }
}

Untuk mengirim permintaan Anda, perluas salah satu opsi berikut:

curl (Linux, macOS, atau Cloud Shell)

Catatan: Perintah berikut mengasumsikan bahwa Anda telah login ke gcloud CLI menggunakan akun pengguna Anda dengan menjalankan gcloud init atau gcloud auth login, atau dengan menggunakan Cloud Shell, yang secara otomatis membuat Anda login ke gcloud CLI. Anda dapat memeriksa akun yang saat ini aktif dengan menjalankan gcloud auth list.

Simpan isi permintaan dalam file bernama request.json, dan jalankan perintah berikut:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID"

PowerShell (Windows)

Catatan: Perintah berikut mengasumsikan bahwa Anda telah login ke gcloud CLI menggunakan akun pengguna Anda dengan menjalankan gcloud init atau gcloud auth login. Anda dapat memeriksa akun yang saat ini aktif dengan menjalankan gcloud auth list.

Simpan isi permintaan dalam file bernama request.json, dan jalankan perintah berikut:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID" | Select-Object -Expand Content

Anda akan melihat respons JSON yang mirip seperti berikut:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata",
    "createTime": "CREATE_TIME",
    "target": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
    "verb": "create",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Terraform

Anda dapat menggunakan resource Terraform untuk membuat cluster.

resource "google_managed_kafka_cluster" "default" {
  project    = data.google_project.default.project_id # Replace this with your project ID in quotes
  cluster_id = "my-cluster-id"
  location   = "us-central1"
  capacity_config {
    vcpu_count   = 3
    memory_bytes = 3221225472
  }
  gcp_config {
    access_config {
      network_configs {
        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 API Managed Service untuk Apache Kafka Go.

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 createCluster(w io.Writer, projectID, region, clusterID, subnet string, cpu, memoryBytes int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// subnet := "projects/my-project-id/regions/us-central1/subnetworks/default"
	// cpu := 3
	// memoryBytes := 3221225472
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

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

	// Memory must be between 1 GiB and 8 GiB per CPU.
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   cpu,
		MemoryBytes: memoryBytes,
	}
	var networkConfig []*managedkafkapb.NetworkConfig
	networkConfig = append(networkConfig, &managedkafkapb.NetworkConfig{
		Subnet: subnet,
	})
	platformConfig := &managedkafkapb.Cluster_GcpConfig{
		GcpConfig: &managedkafkapb.GcpConfig{
			AccessConfig: &managedkafkapb.AccessConfig{
				NetworkConfigs: networkConfig,
			},
		},
	}
	rebalanceConfig := &managedkafkapb.RebalanceConfig{
		Mode: managedkafkapb.RebalanceConfig_AUTO_REBALANCE_ON_SCALE_UP,
	}
	cluster := &managedkafkapb.Cluster{
		Name:            clusterPath,
		CapacityConfig:  capacityConfig,
		PlatformConfig:  platformConfig,
		RebalanceConfig: rebalanceConfig,
	}

	req := &managedkafkapb.CreateClusterRequest{
		Parent:    locationPath,
		ClusterId: clusterID,
		Cluster:   cluster,
	}
	op, err := client.CreateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 10-40 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created 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 API Java Managed Service untuk Apache Kafka.

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.AccessConfig;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.CreateClusterRequest;
import com.google.cloud.managedkafka.v1.GcpConfig;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.NetworkConfig;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.RebalanceConfig;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateCluster {

  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-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    int cpu = 3;
    long memoryBytes = 3221225472L; // 3 GiB
    createCluster(projectId, region, clusterId, subnet, cpu, memoryBytes);
  }

  public static void createCluster(
      String projectId, String region, String clusterId, String subnet, int cpu, long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig =
        CapacityConfig.newBuilder().setVcpuCount(cpu).setMemoryBytes(memoryBytes).build();
    NetworkConfig networkConfig = NetworkConfig.newBuilder().setSubnet(subnet).build();
    GcpConfig gcpConfig =
        GcpConfig.newBuilder()
            .setAccessConfig(AccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
            .build();
    RebalanceConfig rebalanceConfig =
        RebalanceConfig.newBuilder()
            .setMode(RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP)
            .build();
    Cluster cluster =
        Cluster.newBuilder()
            .setCapacityConfig(capacityConfig)
            .setGcpConfig(gcpConfig)
            .setRebalanceConfig(rebalanceConfig)
            .build();

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

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {

      CreateClusterRequest request =
          CreateClusterRequest.newBuilder()
              .setParent(LocationName.of(projectId, region).toString())
              .setClusterId(clusterId)
              .setCluster(cluster)
              .build();

      // The duration of this operation can vary considerably, typically taking between 10-40
      // minutes.
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.createClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf("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)
      Cluster response = future.get();
      System.out.printf("Created cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.createCluster got err: %s", e.getMessage());
    }
  }
}

Python

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Python 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.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 3
# memory_bytes = 3221225472

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.vcpu_count = cpu
cluster.capacity_config.memory_bytes = memory_bytes
cluster.gcp_config.access_config.network_configs = [
    managedkafka_v1.NetworkConfig(subnet=subnet)
]
cluster.rebalance_config.mode = (
    managedkafka_v1.RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP
)

request = managedkafka_v1.CreateClusterRequest(
    parent=client.common_location_path(project_id, region),
    cluster_id=cluster_id,
    cluster=cluster,
)

try:
    operation = client.create_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # The duration of this operation can vary considerably, typically taking 10-40 minutes.
    # We can set a timeout of 3000s (50 minutes).
    response = operation.result(timeout=3000)
    print("Created cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

Memantau operasi pembuatan cluster

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

• Pembuatan cluster biasanya memerlukan waktu 20-30 menit. Untuk melacak progres pembuatan cluster, perintah gcloud managed-kafka 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.

Memecahkan masalah

Untuk mengetahui informasi tentang cara memecahkan masalah error terkait pembuatan cluster, lihat Error pembuatan cluster Kafka.

Apa langkah selanjutnya?

• Pantau cluster Anda.

• Perbarui cluster Anda.

• Mencantumkan semua cluster Anda.

• Lihat cluster Anda.

• Menghapus cluster.

• Buat topik dalam cluster Anda.

• Konfigurasi jaringan untuk Managed Service untuk Apache Kafka.

Apache Kafka® adalah merek dagang terdaftar milik The Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lain.