Melihat topik Google Cloud Managed Service for Apache Kafka

Untuk melihat informasi mendetail tentang satu topik, Anda dapat menggunakan Google Cloud konsol, 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 untuk 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 yang telah ditentukan 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 untuk Apache Kafka.

Properti topik di konsol

Di konsol, Anda dapat melihat properti topik berikut:

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

    • Nama: ID unik topik dalam cluster.

    • Partisi: Jumlah partisi dalam topik. Partisi membagi data topik menjadi beberapa 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 untuk Apache Kafka yang menjadi tempat topik berada.

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

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

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

    • Jumlah byte: Diagram deret waktu yang menunjukkan kecepatan byte diproduksi atau dikirim ke topik. Hal 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 menunjukkan kecepatan permintaan yang dibuat ke topik. Hal ini mencerminkan aktivitas dan penggunaan topik secara keseluruhan. Metrik terkait adalah managedkafka.googleapis.com/topic_request_count.

    • Segmen log per 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 Google Cloud konsol, buka halaman Clusters.

    Buka Cluster

    Cluster yang Anda buat di project akan tercantum.

  2. Klik cluster yang topiknya ingin Anda lihat.

    Halaman detail cluster akan ditampilkan. Di halaman detail cluster, untuk tab Resources, topik akan tercantum.

  3. Untuk melihat topik tertentu, klik nama topik.

    Halaman detail topik akan ditampilkan.

gcloud

  1. Di konsol, aktifkan Cloud Shell. Google Cloud

    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.

  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 lengkap 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 kumpulan 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 untuk 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:

Perintah ini menampilkan subset properti topik, termasuk hal berikut:

  • 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

Outputnya adalah daftar setelan sebagai key-value pair. Flag --all menampilkan semua setelan konfigurasi. Untuk mendapatkan daftar setelan konfigurasi dinamis saja, hapus flag --all.

REST

Sebelum menggunakan salah satu data permintaan, lakukan penggantian berikut:

  • PROJECT_ID: ID proyek 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:

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 Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat Dokumentasi referensi Managed Service untuk 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 untuk 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.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 untuk 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 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?