עדכון מחבר

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

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

לפני שמתחילים

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

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

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

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

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

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

• נותנים את הרשאת העדכון למחבר באשכול ההורה של Connect: managedkafka.connectors.update
• צריך להעניק את ההרשאה list connectors באשכול Connect הראשי: ההרשאה הזו נדרשת רק כדי לעדכן מחבר באמצעות Google Cloud המסוף

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

מאפיינים שאפשר לערוך במחבר

המאפיינים שניתן לערוך במחבר תלויים בסוג שלו. הנה סיכום של המאפיינים שאפשר לערוך בסוגי המחברים הנתמכים:

מחבר מקור MirrorMaker 2.0

• שמות של נושאים שמופרדים בפסיקים או ביטוי רגולרי של נושאים: הנושאים שרוצים לשכפל.

  מידע נוסף על הנכס זמין במאמר שמות של נושאים.

• הגדרה: הגדרות נוספות של המחבר.

  מידע נוסף על הנכס זמין במאמר הגדרה.

• מדיניות הפעלה מחדש של משימות: המדיניות להפעלה מחדש של משימות מחבר שנכשלו.

  מידע נוסף על המאפיין זמין במאמר בנושא מדיניות הפעלה מחדש של משימות.

מחבר BigQuery Sink

• נושאים: נושאי Kafka שמהם יועברו הנתונים בסטרימינג.

  מידע נוסף על המאפיין זמין במאמר בנושא נושאים.

• קבוצת נתונים: קבוצת הנתונים ב-BigQuery שבה יאוחסנו הנתונים.

  מידע נוסף על המאפיין זמין במאמר בנושא מערך נתונים.

• הגדרה: הגדרות נוספות של המחבר.

  מידע נוסף על המאפיין זמין במאמר הגדרה.

• מדיניות הפעלה מחדש של משימות: המדיניות להפעלה מחדש של משימות מחבר שנכשלו.

  מידע נוסף על המאפיין זמין במאמר בנושא מדיניות הפעלה מחדש של משימות.

מחבר Cloud Storage Sink

• נושאים: נושאי Kafka שמהם יועברו הנתונים בסטרימינג.

  מידע נוסף על המאפיין זמין במאמר בנושא נושאים.

• קטגוריה של Cloud Storage: הקטגוריה של Cloud Storage שבה יאוחסנו הנתונים.

  מידע נוסף על המאפיין זמין במאמר קטגוריה.

• הגדרה: הגדרות נוספות של המחבר.

  מידע נוסף על הנכס זמין במאמר הגדרה.

• מדיניות הפעלה מחדש של משימות: המדיניות להפעלה מחדש של משימות מחבר שנכשלו.

  מידע נוסף על המאפיין זמין במאמר בנושא מדיניות הפעלה מחדש של משימות.

מחבר Pub/Sub Source

• מינוי ל-Pub/Sub: המינוי ל-Pub/Sub שממנו יתקבלו ההודעות.
• נושא Kafka: נושא Kafka שאליו יועברו ההודעות בסטרימינג.
• הגדרה: הגדרות נוספות של המחבר. מידע נוסף זמין במאמר הגדרת המחבר.
• מדיניות הפעלה מחדש של משימות: המדיניות להפעלה מחדש של משימות מחבר שנכשלו. מידע נוסף זמין במאמר בנושא מדיניות הפעלה מחדש של משימות.

מחבר Pub/Sub Sink

• נושאים: נושאי Kafka שמתוכם יועברו הודעות בסטרימינג.

  מידע נוסף על הנכס זמין במאמר נושאים.

• נושא Pub/Sub: נושא Pub/Sub שאליו יש לשלוח הודעות.

  מידע נוסף על המאפיין זמין במאמר בנושא נושא Pub/Sub.

• הגדרה: הגדרות נוספות של המחבר.

  מידע נוסף על הנכס זמין במאמר הגדרה.

• מדיניות הפעלה מחדש של משימות: המדיניות להפעלה מחדש של משימות מחבר שנכשלו.

  מידע נוסף על המאפיין זמין במאמר בנושא מדיניות הפעלה מחדש של משימות.

עדכון מחבר

עדכון של מחבר עלול לגרום להפסקה זמנית בזרימת הנתונים בזמן שהשינויים מוחלים.

המסוף

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

   מעבר אל Connect Clusters

2. לוחצים על Connect cluster (חיבור לאשכול) שמארח את המחבר שרוצים לעדכן.

   מוצג הדף Connect cluster details.

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

   תועברו לדף פרטי המחבר.

4. לוחצים על Edit.

5. מעדכנים את המאפיינים הנדרשים של המחבר. הנכסים הזמינים משתנים בהתאם לסוג המחבר.

6. לוחצים על Save.

gcloud

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

   הפעלת Cloud Shell

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

2. כדי לעדכן מחבר, משתמשים בפקודה gcloud managed-kafka connectors update:

   כדי לעדכן את ההגדרות של מחבר, אפשר להשתמש בדגל --configs עם צמדי מפתח-ערך מופרדים בפסיקים, או בדגל --config-file עם נתיב לקובץ JSON או YAML.

   הנה התחביר שמשתמש בדגל --configs עם צמדי מפתח/ערך שמופרדים באמצעות פסיקים.

   gcloud managed-kafka connectors update CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID \
       --configs=KEY1=VALUE1,KEY2=VALUE2...
   

   הנה התחביר שמשתמש בדגל --config-file עם נתיב לקובץ JSON או YAML.

   gcloud managed-kafka connectors update CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID \
       --config-file=PATH_TO_CONFIG_FILE
   

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

   • ‫CONNECTOR_ID: שדה חובה. המזהה של מחבר שרוצים לעדכן.
   • ‫LOCATION: שדה חובה. המיקום של אשכול Connect שמכיל את המחבר.
   • ‫CONNECT_CLUSTER_ID: שדה חובה. המזהה של אשכול Connect שמכיל את המחבר.
   • ‫KEY1=VALUE1,KEY2=VALUE2...: מאפייני ההגדרות שרוצים לעדכן, מופרדים בפסיקים. לדוגמה, tasks.max=2,value.converter.schemas.enable=true.
   • ‫PATH_TO_CONFIG_FILE: הנתיב לקובץ JSON או YAML שמכיל את מאפייני ההגדרה שרוצים לעדכן. לדוגמה, config.json.

   דוגמה לפקודה באמצעות --configs:

   gcloud managed-kafka connectors update test-connector \
       --location=us-central1 \
       --connect-cluster=test-connect-cluster \
       --configs=tasks.max=2,value.converter.schemas.enable=true
   

   דוגמה לפקודה באמצעות --config-file. קובץ לדוגמה בשם update_config.yaml:

   tasks.max: 3
   topic: updated-test-topic
   

   הנה פקודה לדוגמה שמשתמשת בקובץ:

   gcloud managed-kafka connectors update test-connector \
       --location=us-central1 \
       --connect-cluster=test-connect-cluster \
       --config-file=update_config.yaml
   

המשך

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של 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"
	"google.golang.org/protobuf/types/known/fieldmaskpb"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func updateConnector(w io.Writer, projectID, region, connectClusterID, connectorID string, config map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "my-connector"
	// config := map[string]string{"tasks.max": "6"}
	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)
	connector := &managedkafkapb.Connector{
		Name:    connectorPath,
		Configs: config,
	}
	paths := []string{"configs"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateConnectorRequest{
		UpdateMask: updateMask,
		Connector:  connector,
	}
	resp, err := client.UpdateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Updated connector: %#v\n", resp)
	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 com.google.protobuf.FieldMask;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class UpdateConnector {

  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";
    // The new value for the 'tasks.max' configuration.
    String maxTasks = "5";
    updateConnector(projectId, region, clusterId, connectorId, maxTasks);
  }

  public static void updateConnector(
      String projectId, String region, String clusterId, String connectorId, String maxTasks)
      throws IOException {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      Map<String, String> configMap = new HashMap<>();
      configMap.put("tasks.max", maxTasks);

      Connector connector =
          Connector.newBuilder()
              .setName(ConnectorName.of(projectId, region, clusterId, connectorId).toString())
              .putAllConfigs(configMap)
              .build();

      // The field mask specifies which fields to update. Here, we update the 'config' field.
      FieldMask updateMask = FieldMask.newBuilder().addPaths("config").build();

      // This operation is handled synchronously.
      Connector updatedConnector = managedKafkaConnectClient.updateConnector(connector, updateMask);
      System.out.printf("Updated connector: %s\n", updatedConnector.getName());
      System.out.println(updatedConnector.getAllFields());

    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.updateConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

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

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

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector
from google.protobuf import field_mask_pb2

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"
# connector_id = "my-connector"
# configs = {
#     "tasks.max": "6",
#     "value.converter.schemas.enable": "true"
# }

connect_client = ManagedKafkaConnectClient()

connector = Connector()
connector.name = connect_client.connector_path(
    project_id, region, connect_cluster_id, connector_id
)
connector.configs = configs
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.append("config")

# For a list of editable fields, one can check https://cloud.google.com/managed-service-for-apache-kafka/docs/connect-cluster/update-connector#editable-properties.
request = managedkafka_v1.UpdateConnectorRequest(
    update_mask=update_mask,
    connector=connector,
)

try:
    operation = connect_client.update_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Updated connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

You can also update the connector's task restart policy without
including the configuration, by using the `--task-restart-min-backoff`
and `--task-restart-max-backoff` flags. For example:

```sh
gcloud managed-kafka connectors update test-connector \
  --location=us-central1 \
  --connect-cluster=test-connect-cluster \
  --task-restart-min-backoff="60s" \
  --task-restart-max-backoff="90s"

‫Apache Kafka®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.