更新 Connect 集群

您可以修改 Connect 集群，以更新 vCPU 数量、内存、网络和标签等属性。

如需修改 Connect 集群，您可以使用 Google Cloud 控制台、gcloud CLI、客户端库或 Managed Kafka API。您无法使用开源 Apache Kafka API 更新 Connect 集群。

准备工作

并非所有 Connect 集群属性都可以修改。在更新之前，请先查看 Connect 集群的属性。

修改 Connect 集群所需的角色和权限

如需获得修改 Connect 集群所需的权限，请让您的管理员为您授予项目的 Managed Kafka Connect Cluster Editor (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 集群	否
位置	否
标签	是（添加/修改/删除）
Secret	是（添加/删除）

控制台

在 Google Cloud 控制台中，前往连接集群页面。

前往“关联集群”
点击要更新的 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：子网的网络中的 DNS 域名，将对连接集群可见。
- LABELS：（可选）要与集群关联的标签。如需详细了解标签的格式，请参阅标签。要添加的标签键值对列表。键必须以小写字符开头，并且只能包含连字符 (-)、下划线 (_)、小写字符和数字。值只能包含连字符 (-)、下划线 (_)、小写字符和数字。
- SECRET：（可选）要加载到工作器中的 Secret。必须提供 Secret Manager 中的确切 Secret 版本，不支持别名。一个集群最多可加载 32 个 Secret。格式： 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 参考文档。