עדכון אשכול Connect

אפשר לערוך אשכול Connect כדי לעדכן מאפיינים כמו מספר ליבות ה-vCPU, הזיכרון, הרשת והתוויות.

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

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

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

תפקידים והרשאות שנדרשים כדי לערוך אשכול Connect

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

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

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

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

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

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

עריכה של אשכול Connect

כדי לעדכן מאפיינים מסוימים, כמו מעבד (CPU) וזיכרון, צריך להפעיל מחדש את האשכול.

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

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

מאפיין (property)	ניתן לעריכה
מעבדים וירטואלים	כן
זיכרון	כן
רשת	כן
רשת משנה של עובדים	כן
דומייני DNS שניתן לפתור	כן (הוספה או מחיקה)
שם האשכול לחיבור	לא
אשכול Kafka	לא
מיקום	לא
תוויות	כן (הוספה/עריכה/מחיקה)
סודות	כן (הוספה או מחיקה)

המסוף

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

   מעבר אל Connect Clusters

2. לוחצים על אשכול Connect שרוצים לעדכן.

   מוצג הדף Connect cluster details.

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

   מופיע הדף Edit Kafka Connect cluster (עריכת אשכול Kafka Connect).

4. מבצעים את השינויים הנדרשים במאפיינים שאפשר לערוך.

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

gcloud

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

   הפעלת Cloud Shell

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

2. מריצים את הפקודה gcloud managed-kafka connect-clusters update:

   gcloud managed-kafka connect-clusters update CONNECT_CLUSTER_ID \
       --location=LOCATION \
       [--cpu=CPU --memory=MEMORY
        | --clear-dns-names \
        | --dns-name=DNS_NAME --clear-labels \
        | --labels=LABELS --clear-secrets \
        | --secret=SECRET [--primary-subnet=WORKER_SUBNET \
       [--async]
   

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

   • ‫CONNECT_CLUSTER_ID: המזהה או השם של אשכול Connect. אי אפשר לשנות את השם של אשכול Connect.
   • ‫LOCATION: המיקום של אשכול Connect. אי אפשר לשנות את המיקום של אשכול Connect.
   • ‫CPU: מספר ליבות ה-vCPU של אשכול Connect. הערך המינימלי הוא 3 vCPUs.
   • ‫MEMORY: כמות הזיכרון של אשכול Connect. צריך להשתמש ביחידות 'MB',‏ 'MiB',‏ 'GB',‏ 'GiB',‏ 'TB' או 'TiB'. לדוגמה, '10GiB'. צריך להקצות בין 1 ל-8 גיגה-בייט לכל CPU וירטואלי.
   • ‫DNS_NAME: שם הדומיין של ה-DNS מהרשת של רשת המשנה שיוצג ב-Connect Cluster.
   • ‫LABELS: (אופציונלי) תוויות לשיוך לאשכול. מידע נוסף על הפורמט של תוויות זמין במאמר בנושא תוויות. רשימה של צמדי מפתח=ערך של תוויות להוספה. המפתחות צריכים להתחיל באות קטנה באנגלית, ויכולים לכלול רק מקפים (-), קווים תחתונים (_), אותיות קטנות באנגלית וספרות. הערכים יכולים להכיל רק מקפים (-), קווים תחתונים (_), אותיות קטנות וספרות.
   • ‫SECRET: (אופציונלי) סודות לטעינה לעובדים. צריך לספק גרסאות מדויקות של סודות מ-Secret Manager, אין תמיכה בכינויים. אפשר לטעון עד 32 ערכי Secret לאותו אשכול. פורמט: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID

   • ‫WORKER_SUBNET: רשת המשנה של העובדים באשכול Connect. רשת המשנה של העובד צריכה להיות באותו אזור כמו אשכול Connect.

     הפורמט של רשת המשנה הוא projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.

המשך

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של 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 updateConnectCluster(w io.Writer, projectID, region, clusterID string, memoryBytes int64, labels map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-connect-cluster"
	// memoryBytes := 25769803776 // 24 GiB in bytes
	// labels := map[string]string{"environment": "production"}
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	clusterPath := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, clusterID)

	// Capacity configuration update
	capacityConfig := &managedkafkapb.CapacityConfig{
		MemoryBytes: memoryBytes,
	}

	connectCluster := &managedkafkapb.ConnectCluster{
		Name:           clusterPath,
		CapacityConfig: capacityConfig,
		Labels:         labels,
	}
	paths := []string{"capacity_config.memory_bytes", "labels"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateConnectClusterRequest{
		UpdateMask:     updateMask,
		ConnectCluster: connectCluster,
	}
	op, err := client.UpdateConnectCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateConnectCluster got err: %w", err)
	}
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Updated connect cluster: %#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.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.ConnectCluster;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.UpdateConnectClusterRequest;
import com.google.protobuf.FieldMask;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class UpdateConnectCluster {

  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";
    long memoryBytes = 25769803776L; // 24 GiB
    updateConnectCluster(projectId, region, clusterId, memoryBytes);
  }

  public static void updateConnectCluster(
      String projectId, String region, String clusterId, long memoryBytes) throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setMemoryBytes(memoryBytes).build();
    ConnectCluster connectCluster = ConnectCluster.newBuilder()
        .setName(ConnectClusterName.of(projectId, region, clusterId).toString())
        .setCapacityConfig(capacityConfig)
        .build();
    FieldMask updateMask = FieldMask.newBuilder().addPaths("capacity_config.memory_bytes").build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.updateConnectClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create(
        settingsBuilder.build())) {
      UpdateConnectClusterRequest request = UpdateConnectClusterRequest.newBuilder()
          .setUpdateMask(updateMask)
          .setConnectCluster(connectCluster).build();
      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
          .updateConnectClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details. CreateConnectCluster contains sample
      // code for polling logs.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf(
          "Connect cluster update started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(),
          operation.isDone(),
          future.getMetadata().get().toString());

      ConnectCluster response = future.get();
      System.out.printf("Updated connect cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaConnectClient.updateConnectCluster 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 ConnectCluster
from google.protobuf import field_mask_pb2

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

connect_client = ManagedKafkaConnectClient()

connect_cluster = ConnectCluster()
connect_cluster.name = connect_client.connect_cluster_path(
    project_id, region, connect_cluster_id
)
connect_cluster.capacity_config.memory_bytes = memory_bytes
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.append("capacity_config.memory_bytes")

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

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

מה השלב הבא?

• רשימת כל אשכולות Connect

• הצגת פרטים על אשכול Connect

• מחיקת אשכול Connect

• סקירה כללית על מחברים

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