Melihat konektor

Dengan melihat detail konektor, Anda dapat memeriksa konfigurasi, status operasional, kebijakan memulai ulang tugas, dan memantau metrik performanya.

Untuk melihat detail konektor di cluster Connect, Anda dapat menggunakan Google Cloud konsol, gcloud CLI, library klien Managed Service untuk Apache Kafka, atau Managed Kafka API. Anda tidak dapat menggunakan Apache Kafka API open source untuk melihat konektor.

Peran dan izin yang diperlukan untuk melihat konektor

Untuk mendapatkan izin yang diperlukan guna melihat konektor, 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 konektor. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk melihat konektor:

  • Berikan izin list connectors di cluster Connect induk: managedkafka.connectors.list
  • Berikan izin untuk mendapatkan detail konektor di cluster Connect induk: managedkafka.connectors.get

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

Melihat detail konektor

Konsol

  1. Di konsol Google Cloud , buka halaman Connect Clusters.

    Buka Connect Clusters

  2. Klik cluster Connect yang menghosting konektor yang ingin Anda lihat.

    Halaman Connect cluster details akan ditampilkan.

  3. Di tab Resources, temukan konektor dalam daftar, lalu klik namanya. Anda akan dialihkan ke halaman Detail konektor.

  4. Halaman Connector details menampilkan tab berikut:

    • Konfigurasi: Menampilkan konfigurasi konektor, termasuk:
      • Nama: Nama konektor.
      • Status: Status operasional konektor. Misalnya, Berlari.
      • Kebijakan mulai ulang tugas: Kebijakan untuk memulai ulang tugas yang gagal. Misalnya, apakah akan memulai ulang tugas yang gagal atau tidak, dan dengan setelan mundur yang mana.
      • Properti konfigurasi: Daftar pasangan nilai kunci yang menentukan konfigurasi konektor.
    • Pemantauan: Menyediakan grafik untuk memantau konektor, seperti:
      • Jumlah error tugas: Jumlah tugas yang mengalami error.
      • Jumlah tugas aktif: Jumlah tugas yang saat ini aktif.

    Halaman ini juga menyertakan tombol untuk mengedit, menghapus, menjeda, menghentikan, dan memulai ulang konektor.

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. Gunakan perintah gcloud managed-kafka connectors describe untuk mendeskripsikan konektor:

    gcloud managed-kafka connectors describe CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID

    Ganti kode berikut:

    • CONNECTOR_ID: Wajib diisi. ID konektor yang ingin Anda jelaskan.
    • LOCATION: Wajib diisi. Lokasi cluster Connect yang berisi konektor.
    • CONNECT_CLUSTER_ID: Wajib diisi. ID cluster Connect yang berisi konektor.

Contoh perintah:

gcloud managed-kafka connectors describe test-connector \
  --location=us-central1 \
  --connect-cluster=test-connect-cluster

Contoh output:

config:
  connector.class: com.google.cloud.kafka.connect.pubsub.PubsubSinkConnector
  kafka.topic.regex: .*
  key.converter: org.apache.kafka.connect.storage.StringConverter
  project: test-project
  tasks.max: '1'
  topic: test-pubsub-topic
  value.converter: org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable: 'false'
createTime: '2024-03-13T05:17:34.123456Z'
labels:
  test-label-key: test-label-value
name: projects/test-project/locations/us-central1/connectClusters/test-connect-cluster/connectors/test-connector
state: RUNNING
taskRestartPolicy: RESTART_WITH_EXPONENTIAL_BACKOFF
updateTime: '2024-03-13T05:18:15.987654Z'

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 getConnector(w io.Writer, projectID, region, connectClusterID, connectorID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "my-connector"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	connectorPath := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s/connectors/%s", projectID, region, connectClusterID, connectorID)
	req := &managedkafkapb.GetConnectorRequest{
		Name: connectorPath,
	}
	connector, err := client.GetConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.GetConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Got connector: %#v\n", connector)
	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.rpc.ApiException;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;

public class GetConnector {

  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 connectorId = "my-connector";
    getConnector(projectId, region, clusterId, connectorId);
  }

  public static void getConnector(
      String projectId, String region, String clusterId, String connectorId) throws IOException {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      ConnectorName name = ConnectorName.of(projectId, region, clusterId, connectorId);
      // This operation is handled synchronously.
      Connector connector = managedKafkaConnectClient.getConnector(name);
      System.out.println(connector.getAllFields());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.getConnector got err: %s\n", 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 NotFound
from google.cloud.managedkafka_v1.services.managed_kafka_connect import ManagedKafkaConnectClient
from google.cloud import managedkafka_v1

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"
# connector_id = "my-connector"

connect_client = ManagedKafkaConnectClient()

connector_path = connect_client.connector_path(
    project_id, region, connect_cluster_id, connector_id
)
request = managedkafka_v1.GetConnectorRequest(
    name=connector_path,
)

try:
    connector = connect_client.get_connector(request=request)
    print("Got connector:", connector)
except NotFound as e:
    print(f"Failed to get connector {connector_id} with error: {e}")

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