Connect クラスタを更新する

Connect クラスタを編集して、vCPU の数、メモリ、ネットワーク、ラベルなどのプロパティを更新できます。

Connect クラスタを編集するには、 Google Cloud コンソール、gcloud CLI、クライアントライブラリ、または Managed Kafka API を使用します。オープンソースの Apache Kafka API を使用して Connect クラスタを更新することはできません。

始める前に

Connect クラスタのすべてのプロパティを編集できるわけではありません。更新する前に、Connect クラスタのプロパティを確認します。

Connect クラスタの編集に必要なロールと権限

Connect クラスタの編集に必要な権限を取得するには、プロジェクトに対する Managed Kafka Connect クラスタ編集者（roles/managedkafka.connectClusterEditor）IAM ロールを付与するよう管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

この事前定義ロールには、Connect クラスタの編集に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

Connect クラスタを編集するには、次の権限が必要です。

指定したロケーションに対する Connect クラスタ権限を更新に付与します。 managedkafka.connectClusters.update
指定されたロケーションで、ビューに Connect クラスタ権限を付与します。この権限は、 Google Cloud コンソールを使用して Connect クラスタを更新する場合にのみ必要です。 managedkafka.connectors.list

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

このロールの詳細については、Managed Service for Apache Kafka の事前定義ロールをご覧ください。

Connect クラスタを編集する

CPU やメモリなどの特定のプロパティを更新するには、クラスタの再起動が必要です。

クラスタを再起動するとデータは保持されますが、レイテンシが増加する可能性があります。クラスタ内のワーカーの初期数によって、再起動の期間が決まります。

次の Connect クラスタプロパティを更新できます。

プロパティ	編集可能
vCPU	はい
メモリ	はい
ネットワーク	はい
ワーカーサブネット	はい
解決可能な DNS ドメイン	はい（追加/削除）
Connect クラスタ名	いいえ
Kafka クラスタ	いいえ
場所	いいえ
ラベル	はい（追加/編集/削除）
シークレット	はい（追加/削除）

コンソール

Google Cloud コンソールで、[クラスタを接続] ページに移動します。

[Connect クラスタ] に移動
更新する Connect クラスタをクリックします。

[クラスタの詳細を接続] ページが表示されます。
[編集] をクリックします。

[Kafka Connect クラスタの編集] ページが表示されます。
編集可能なプロパティに必要な変更を加えます。
[保存] をクリックします。

gcloud

In the Google Cloud console, activate Cloud Shell.

Activate Cloud Shell

At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.
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 クラスタの ID または名前。Connect クラスタの名前は不変です。
- LOCATION: Connect クラスタのロケーション。Connect クラスタのロケーションは変更できません。
- CPU: Connect クラスタの vCPU の数。最小値は 3 vCPU です。
- MEMORY: Connect クラスタのメモリ容量。「MB」、「MiB」、「GB」、「GiB」、「TB」、「TiB」の単位を使用します。たとえば、「10GiB」です。vCPU あたり 1 GiB ～ 8 GiB の範囲でプロビジョニングする必要があります。
- DNS_NAME: Connect クラスタに公開するサブネットのネットワークの DNS ドメイン名。
- LABELS:（省略可）クラスタに関連付けるラベル。ラベルの形式について詳しくは、ラベルをご覧ください。追加するラベルの KEY=VALUE ペアのリスト。キーは小文字で始まり、使用できるのはハイフン（-）、アンダースコア（_）、小文字、数字のみです。値に使用できるのは、ハイフン（-）、アンダースコア（_）、小文字、数字のみです。
- SECRET:（省略可）ワーカーに読み込む Secret。Secret Manager の正確なシークレットバージョンを指定する必要があります。エイリアスはサポートされていません。1 つのクラスタに最大 32 個のシークレットを読み込むことができます。形式: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID
- WORKER_SUBNET: Connect クラスタのワーカーサブネット。ワーカーサブネットは Connect クラスタと同じリージョンに存在する必要があります。
  
  サブネットの形式は projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID です。

Go

このサンプルを試す前に、クライアントライブラリをインストールするにある Go の設定手順を行ってください。詳細については、 Managed Service for Apache Kafka Go API のリファレンスドキュメントをご覧ください。

Managed Service for Apache Kafka に対する認証を行うには、アプリケーションのデフォルト認証情報（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 リファレンスドキュメントをご覧ください。

Managed Service for Apache Kafka に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の 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 の設定手順を行ってください。詳細については、 Managed Service for Apache Kafka Python API リファレンスドキュメントをご覧ください。

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}")

次のステップ

Apache Kafka® は、Apache Software Foundation または米国その他の諸国における関連会社の商標です。

Connect クラスタを更新する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。