Melihat topik Google Cloud Managed Service for Apache Kafka

Untuk melihat informasi mendetail tentang satu topik, Anda dapat menggunakan konsolGoogle Cloud , Google Cloud CLI, library klien, Managed Kafka API, atau Apache Kafka API open source.

Peran dan izin yang diperlukan untuk melihat topik

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

Izin yang diperlukan

Izin berikut diperlukan untuk melihat topik:

• Daftar topik: managedkafka.topics.list
• Dapatkan topik: managedkafka.topics.get

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

Untuk mengetahui informasi selengkapnya tentang peran IAM Managed Kafka Viewer (roles/managedkafka.viewer), lihat Peran bawaan Managed Service for Apache Kafka.

Properti topik di konsol

Di konsol, Anda dapat melihat properti topik berikut:

• Konfigurasi: Tab ini memberikan detail konfigurasi umum tentang topik, termasuk berikut ini:

  • Name: ID unik topik dalam cluster.

  • Partisi: Jumlah partisi dalam topik. Partisi membagi data topik menjadi segmen untuk skalabilitas dan paralelisme.

  • Replika: Jumlah salinan (replika) yang dipertahankan untuk setiap partisi guna memastikan redundansi dan ketersediaan data.

  • Cluster: Nama cluster Managed Service for Apache Kafka tempat topik berada.

  • Region: Google Cloud Region tempat cluster dan topik berada.

  • Parameter topik non-default: Setiap penggantian konfigurasi tingkat topik yang telah ditetapkan untuk topik, yang berbeda dari default seluruh cluster.

• Pemantauan: Tab ini menyediakan diagram visual yang menampilkan metrik utama terkait aktivitas dan performa topik. Diagram ini mencakup:

  • Jumlah byte: Diagram deret waktu yang menunjukkan kecepatan byte diproduksi atau dikirim ke topik. Ini menunjukkan volume data yang dipublikasikan ke topik dari waktu ke waktu. Metrik yang sesuai adalah managedkafka.googleapis.com/byte_in_count.

  • Jumlah permintaan: Diagram deret waktu yang menampilkan rasio permintaan yang dibuat ke topik. Nilai ini mencerminkan keseluruhan aktivitas dan penggunaan topik. Metrik terkait adalah managedkafka.googleapis.com/topic_request_count.

  • Segmen log menurut partisi: Diagram ini menampilkan jumlah segmen log aktif untuk setiap partisi dalam topik. Segmen log adalah file fisik di disk tempat Kafka menyimpan data topik. Metrik yang relevan adalah managedkafka.googleapis.com/log_segments.

• Grup konsumen: Bagian ini mencantumkan grup konsumen yang berlangganan topik. Grup konsumen adalah sekumpulan konsumen yang bekerja sama untuk membaca pesan dari topik.

Melihat topik

Konsol

1. Di konsol Google Cloud , buka halaman Clusters.

   Buka Cluster

   Cluster yang Anda buat dalam project akan dicantumkan.

2. Klik cluster yang topiknya ingin Anda lihat.

   Halaman detail cluster akan ditampilkan. Di halaman detail cluster, topik dicantumkan di tab Resources.

3. Untuk melihat topik tertentu, klik nama topik.

   Halaman detail topik akan ditampilkan.

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. Jalankan perintah gcloud managed-kafka topics describe:

   gcloud managed-kafka topics describe TOPIC_ID \
     --cluster=CLUSTER_ID --location=LOCATION_ID
   

   Perintah ini mengambil dan menampilkan detail komprehensif tentang topik yang ditentukan. Informasi ini mencakup setelan konfigurasinya, seperti jumlah partisi, faktor replikasi, dan penggantian konfigurasi tingkat topik.

   Ganti kode berikut:

   • TOPIC_ID: ID topik.
   • CLUSTER_ID: ID cluster yang berisi topik.
   • LOCATION_ID: Lokasi cluster.

Perintah gcloud managed-kafka topics describe menampilkan informasi minimal tentang topik, seperti jumlah partisi dan faktor replikasi. Untuk mendapatkan informasi yang lebih mendetail, termasuk penetapan partisi dan set lengkap setelan konfigurasi, gunakan alat command line kafka-topics.sh.

Kafka CLI

Sebelum menjalankan perintah ini, instal alat command line Kafka di VM Compute Engine. VM harus dapat menjangkau subnet yang terhubung ke cluster Managed Service for Apache Kafka Anda. Ikuti petunjuk di Membuat dan menggunakan pesan dengan alat command line Kafka.

Untuk melihat detail tentang topik, jalankan perintah kafka-topics.sh --describe:

kafka-topics.sh --describe \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --topic TOPIC_ID

Ganti kode berikut:

• BOOTSTRAP_ADDRESS: Alamat bootstrap cluster Managed Service untuk Apache Kafka.
• TOPIC_ID: ID topik.

Perintah ini menampilkan subset properti topik, termasuk berikut ini:

• Jumlah partisi
• Faktor replikasi
• Penetapan partisi
• Konfigurasi dinamis (setelan yang telah Anda tetapkan secara eksplisit)
• Konfigurasi statis (setelan yang diterapkan saat cluster dimulai)

Untuk melihat kumpulan lengkap setelan konfigurasi untuk topik, termasuk setelan dengan nilai default, jalankan perintah kafka-configs.sh --describe:

kafka-configs.sh --describe \
--bootstrap-server=BOOTSTRAP_ADDRESS \
--command-config client.properties \
--entity-type topics \
--entity-name TOPIC_ID \
--all

Output-nya adalah daftar setelan sebagai key-value pair. Flag --all menampilkan semua setelan konfigurasi. Untuk mendapatkan daftar hanya setelan konfigurasi dinamis, hilangkan flag --all.

REST

Sebelum menggunakan salah satu data permintaan, lakukan penggantian berikut:

• PROJECT_ID: Project ID Google Cloud Anda
• LOCATION: lokasi cluster
• CLUSTER_ID: ID cluster
• TOPIC_ID: ID topik

Metode HTTP dan URL:

GET https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_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.

Jalankan perintah berikut:

curl -X GET \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_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.

Jalankan perintah berikut:

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

Invoke-WebRequest `
    -Method GET `
    -Headers $headers `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID" | Select-Object -Expand Content

Anda akan melihat respons JSON seperti berikut:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID",
  "partitionCount": PARTITION_COUNT,
  "replicationFactor": REPLICATION_FACTOR
}

Go

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di bagian 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 getTopic(w io.Writer, projectID, region, clusterID, topicID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

	clusterPath := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", projectID, region, clusterID)
	topicPath := fmt.Sprintf("%s/topics/%s", clusterPath, topicID)
	req := &managedkafkapb.GetTopicRequest{
		Name: topicPath,
	}
	topic, err := client.GetTopic(ctx, req)
	if err != nil {
		return fmt.Errorf("client.GetTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Got topic: %#v\n", topic)
	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 for Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.Topic;
import com.google.cloud.managedkafka.v1.TopicName;
import java.io.IOException;

public class GetTopic {

  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 topicId = "my-topic";
    getTopic(projectId, region, clusterId, topicId);
  }

  public static void getTopic(String projectId, String region, String clusterId, String topicId)
      throws Exception {
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      // This operation is being handled synchronously.
      Topic topic =
          managedKafkaClient.getTopic(TopicName.of(projectId, region, clusterId, topicId));
      System.out.println(topic.getAllFields());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.getTopic 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 for Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

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

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# topic_id = "my-topic"

client = managedkafka_v1.ManagedKafkaClient()

topic_path = client.topic_path(project_id, region, cluster_id, topic_id)
request = managedkafka_v1.GetTopicRequest(
    name=topic_path,
)

try:
    topic = client.get_topic(request=request)
    print("Got topic:", topic)
except NotFound as e:
    print(f"Failed to get topic {topic_id} with error: {e.message}")

Apa langkah selanjutnya?

• Menghapus topik

• Mengedit topik

• Mencantumkan semua grup konsumen

• Melihat grup konsumen