צפייה במחבר

כשמציגים את הפרטים של מחבר, אפשר לבדוק את ההגדרה שלו, את מצב הפעולה, את מדיניות ההפעלה מחדש של המשימות ולעקוב אחרי מדדי הביצועים שלו.

כדי לראות את הפרטים של מחבר באשכול Connect, אפשר להשתמש במסוף Google Cloud , ב-CLI של gcloud, בספריית הלקוח של השירות המנוהל ל-Apache Kafka או ב-Managed Kafka API. אי אפשר להשתמש ב-API של Apache Kafka בקוד פתוח כדי להציג את המחברים.

התפקידים וההרשאות שנדרשים כדי לראות מחבר

כדי לקבל את ההרשאות שנדרשות להצגת מחבר, צריך לבקש מהאדמין להקצות לכם את תפקיד ה-IAM‏ Managed Kafka Viewer (roles/managedkafka.viewer) בפרויקט. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

זהו תפקיד מוגדר מראש עם ההרשאות לצפייה במחבר. כדי לראות בדיוק אילו הרשאות נדרשות, אפשר להרחיב את הקטע ההרשאות הנדרשות:

ההרשאות הנדרשות

כדי לראות מחבר, צריך את ההרשאות הבאות:

• נותנים למחברי הרשימות הרשאה באשכול Connect הראשי: managedkafka.connectors.list
• נותנים הרשאה לקבלת פרטי מחבר באשכול הראשי של Connect: managedkafka.connectors.get

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

צפייה בפרטי המחבר

המסוף

1. נכנסים לדף Connect Clusters במסוף Google Cloud .

   מעבר אל Connect Clusters

2. לוחצים על אשכול Connect שמארח את המחבר שרוצים לראות.

   מוצג הדף Connect cluster details.

3. בכרטיסייה משאבים, מוצאים את המחבר ברשימה ולוחצים על השם שלו. תועברו לדף פרטי המחבר.

4. בדף Connector details מוצגות הכרטיסיות הבאות:

   • הגדרה: מציג את ההגדרה של המחבר, כולל:
     • שם: השם של המחבר.
     • מצב: מצב הפעולה של המחבר. לדוגמה, ריצה.
     • מדיניות הפעלה מחדש של משימות: המדיניות להפעלה מחדש של משימות שנכשלו. לדוגמה, האם להפעיל מחדש משימות שנכשלו או לא, ועם אילו הגדרות של השהיה.
     • מאפייני ההגדרה: רשימה של צמדי מפתח/ערך שמגדירים את ההגדרה של המחבר.
   • מעקב: כולל תרשימים למעקב אחרי המחבר, למשל:
     • מספר השגיאות במשימה: מספר המשימות שנתקלו בשגיאות.
     • ‫Active task count: מספר המשימות שפעילות כרגע.

   בדף הזה יש גם לחצנים לעריכה, מחיקה, השהיה, עצירה והפעלה מחדש של המחבר.

gcloud

1. במסוף Google Cloud , מפעילים את Cloud Shell.

   הפעלת Cloud Shell

   בחלק התחתון של Google Cloud המסוף יתחיל סשן של Cloud Shell ותופיע הודעה של שורת הפקודה. Cloud Shell היא סביבת מעטפת שבה ה-CLI של Google Cloud מותקן ומוגדרים ערכים לפרויקט הקיים. הסשן יופעל תוך כמה שניות.

2. משתמשים בפקודה gcloud managed-kafka connectors describe כדי לתאר מחבר:

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

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫CONNECTOR_ID: שדה חובה. המזהה של מחבר שרוצים לתאר.
   • ‫LOCATION: שדה חובה. המיקום של אשכול Connect שמכיל את המחבר.
   • ‫CONNECT_CLUSTER_ID: שדה חובה. המזהה של אשכול Connect שמכיל את המחבר.

פקודה לדוגמה:

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

פלט לדוגמה:

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 בקטע התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של ה-API של Go עבור שירות מנוהל ל-Apache Kafka.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, צריך להגדיר את Application Default Credentials‏(ADC). מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

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

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במאמר התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של Managed Service for Apache Kafka Java API.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, מגדירים את ה-Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

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

לפני שמנסים את הדוגמה הזו, פועלים לפי הוראות ההגדרה של Python במאמר התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של ה-API בשפת Python של שירות מנוהל ל-Apache Kafka.

כדי לבצע אימות לשירות המנוהל ל-Apache Kafka, מגדירים את ה-Application Default Credentials. מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

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®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.