Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

更新 Google Cloud Managed Service for Apache Kafka 集群

您可以修改 Google Cloud Managed Service for Apache Kafka 集群，以更新集群大小（包括 vCPU 数量和内存）、连接的子网列表、公共集群的允许来源 IP 范围、自动重新平衡配置和 mTLS 配置等属性。

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

更新某些属性（例如 vCPU 数量和内存）可能需要服务重启集群。该服务一次重启一个代理。在此过程中，对各个代理的请求可能会失败，但这些失败是暂时性的。常用的客户端库会自动处理这些错误。

所需角色和权限

如需获得更新集群所需的权限，请让您的管理员为您授予项目的Managed Kafka Cluster Editor (roles/managedkafka.clusterEditor) IAM 角色。如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

此预定义角色包含更新集群所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

更新集群需要以下权限：

修改集群： managedkafka.clusters.update

您也可以使用自定义角色或其他预定义角色来获取这些权限。

调整集群大小

如果您更新集群的 vCPU 数量或内存，则适用以下规则：

集群的整体 vCPU 与内存之比必须始终介于 1:1 和 1:8 之间。
每个现有代理必须至少有 1 个 vCPU 和 1 GiB 内存。代理数量永远不会减少。
如果集群具有自定义磁盘配置，则更新必须满足本地存储的磁盘配置要求。
如果扩容，每个代理的平均 vCPU 和内存与更新前的平均值相比，不能减少超过 10%。例如，如果您尝试将集群从 45 个 vCPU（3 个代理）扩容到 48 个 vCPU（4 个代理），则每个代理的平均 vCPU 从 15 个减少到 12 个，减少了 20%，超过了 10% 的限制。

如果您需要将 vCPU 数量减少超过 10%，我们建议分几个阶段减少。每次更新后，请监控资源利用率，并根据需要重新平衡分区。

但是，如果您确信代理在更新后将有足够的容量，则可以通过运行带有 allow_broker_downscale_on_cluster_upscale=true 标志的 gcloud managed-kafka clusters update 命令来停用此检查。此标志表示您接受潜在的性能风险。

如需了解详情，请参阅更新集群大小。

公共集群配置

您可以为现有集群启用或停用公共访问权限，以及添加或移除允许的来源 IP 范围。如需详细了解允许的来源 IP 范围的要求和规则，请参阅公共集群。

Managed Service for Apache Kafka 使用 Cloud Next Generation Firewall 来限制对公共集群的访问。移除允许的来源 IP 范围或停用公共访问权限仅适用于新连接。如需了解详情，请参阅对现有流量的影响。

修改集群

如需修改集群，请按以下步骤操作：

控制台

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

转到“集群”

在集群列表中，点击要修改其属性的集群。

控制台会显示集群详情页面。

在集群详情页面上，点击修改。
根据需要修改属性。您可以在控制台中修改集群的以下属性：
- 内存
- vCPUs
- 子网
- 重新平衡配置
- mTLS 配置
- 标签
点击保存。

gcloud

在 Google Cloud 控制台中，激活 Cloud Shell。

激活 Cloud Shell

Cloud Shell 会话随即会在控制台 Google Cloud 的底部启动，并显示命令行提示符。Cloud Shell 是一个已安装 Google Cloud CLI 且已为当前项目设置值的 Shell 环境。该会话可能需要几秒钟来完成初始化。
在使用以下任何命令数据之前，请进行以下替换：
- PROJECT_ID：项目 ID。
- LOCATION：集群的位置。
- CLUSTER_ID：集群的 ID。
- CPU_COUNT：集群的 vCPU 数量。
- MEMORY：集群的内存量。示例：10GiB。
- SUBNET_ID：要连接的子网的子网 ID。示例：default。
- LABELS：要与集群关联的标签。
- ALLOWED_SOURCE_IP_RANGES：公共集群互联网访问的允许来源 IPv4 CIDR 范围。
执行以下命令：
Linux、macOS 或 Cloud Shell

注意：通过运行 gcloud init 或 gcloud auth login 和 gcloud config set project，确保您已使用身份验证和项目初始化了 Google Cloud CLI。
```
gcloud managed-kafka clusters update CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU_COUNT \
    --memory=MEMORY \
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
    --auto-rebalance \
    --labels=LABELS \
    --public-cluster \
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES
```
Windows (PowerShell)

注意：通过运行 gcloud init 或 gcloud auth login 和 gcloud config set project，确保您已使用身份验证和项目初始化了 Google Cloud CLI。
```
gcloud managed-kafka clusters update CLUSTER_ID `
    --location=LOCATION `
    --cpu=CPU_COUNT `
    --memory=MEMORY `
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
    --auto-rebalance `
    --labels=LABELS `
    --public-cluster `
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES
```
Windows (cmd.exe)

注意：通过运行 gcloud init 或 gcloud auth login 和 gcloud config set project，确保您已使用身份验证和项目初始化了 Google Cloud CLI。
```
gcloud managed-kafka clusters update CLUSTER_ID ^
    --location=LOCATION ^
    --cpu=CPU_COUNT ^
    --memory=MEMORY ^
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
    --auto-rebalance ^
    --labels=LABELS ^
    --public-cluster ^
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES
```
您应该会收到类似如下所示的响应：
```
done: false
metadata:
  '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
  apiVersion: v1
  createTime: 'CREATE_TIME'
  requestedCancellation: false
  target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
  verb: update
name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
```
- 如需停用公共访问权限，请使用 --no-public-cluster 标志。
- 如果您在命令中使用 --async 标志，系统会发送更新请求并立即返回响应，而无需等待操作完成。使用 --async 标志后，您可以在集群更新在后台进行时继续执行其他任务。如果您不使用 --async 标志，系统会等待操作完成，然后返回响应。您必须等到集群完全更新后，才能继续执行其他任务。

REST

在使用任何请求数据之前，请进行以下替换：

PROJECT_ID：您的 Google Cloud 项目 ID
LOCATION：集群的位置。
CLUSTER_ID：集群的 ID。
UPDATE_MASK：要更新的字段，以完全限定名称的英文逗号分隔列表的形式表示。示例： capacityConfig.vcpuCount,capacityConfig.memoryBytes
CPU_COUNT：集群的 vCPU 数量。
MEMORY：集群的内存量（以字节为单位）。示例：3221225472。
SUBNET_ID：要连接的子网的子网 ID。示例：default。

HTTP 方法和网址：

PATCH https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK

请求 JSON 正文：

{
  "capacityConfig": {
    "vcpuCount": CPU_COUNT,
    "memoryBytes": MEMORY
  },
  "gcpConfig": {
    "accessConfig": {
      "networkConfigs": [
        {
          "subnet": "projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID"
        }
      ]
    }
  }
}

如需发送您的请求，请展开以下选项之一：

curl（Linux、macOS 或 Cloud Shell）

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI，或者使用了 Cloud Shell，这会使您自动登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK"

PowerShell (Windows)

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK" | Select-Object -Expand Content

您应该收到类似以下内容的 JSON 响应：

{
  "name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata",
    "createTime": "CREATE_TIME",
    "target": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
    "verb": "update",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

在请求正文中，仅包含您要更新的字段，如在 UPDATE_MASK 查询参数中所指定。

如需添加子网，请以以下格式将新条目附加到 networkConfigs： projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID。示例：projects/sample-project/regions/us-central1/subnetworks/default。

如需启用公共访问权限或更新允许的来源 IP 范围，请在 gcpConfig.accessConfig.publicClusterConfig查询参数中添加 UPDATE_MASK，并在请求正文中指定 allowedSourceIpRanges数组。请求正文示例：

{
  "gcpConfig": {
    "accessConfig": {
      "publicClusterConfig": {
        "allowedSourceIpRanges": [
          "203.0.113.0/24"
        ]
      }
    }
  }
}

如需停用公共访问权限，请在 UPDATE_MASK查询参数中添加 gcpConfig.accessConfig.publicClusterConfig，并在请求正文中传递一个空 JSON 对象 {}（或省略 publicClusterConfig）。请求正文示例：
```
{}
```

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 updateCluster(w io.Writer, projectID, region, clusterID string, memory int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// memoryBytes := 4221225472
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

	clusterPath := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", projectID, region, clusterID)
	capacityConfig := &managedkafkapb.CapacityConfig{
		MemoryBytes: memory,
	}
	cluster := &managedkafkapb.Cluster{
		Name:           clusterPath,
		CapacityConfig: capacityConfig,
	}
	paths := []string{"capacity_config.memory_bytes"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateClusterRequest{
		UpdateMask: updateMask,
		Cluster:    cluster,
	}
	op, err := client.UpdateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateCluster 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 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.Cluster;
import com.google.cloud.managedkafka.v1.ClusterName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.UpdateClusterRequest;
import com.google.protobuf.FieldMask;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class UpdateCluster {

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

  public static void updateCluster(
      String projectId, String region, String clusterId, long memoryBytes) throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setMemoryBytes(memoryBytes).build();
    Cluster cluster =
        Cluster.newBuilder()
            .setName(ClusterName.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
    ManagedKafkaSettings.Builder settingsBuilder = ManagedKafkaSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.updateClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {
      UpdateClusterRequest request =
          UpdateClusterRequest.newBuilder().setUpdateMask(updateMask).setCluster(cluster).build();
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.updateClusterOperationCallable().futureCall(request);

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

      Cluster response = future.get();
      System.out.printf("Updated cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.updateCluster got err: %s", e.getMessage());
    }
  }
}

Python

在试用此示例之前，请按照安装客户端库中的 Python 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Python API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置 ADC。

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.protobuf import field_mask_pb2

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

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
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-kafka/docs/create-cluster#properties.
request = managedkafka_v1.UpdateClusterRequest(
    update_mask=update_mask,
    cluster=cluster,
)

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

限制

创建 Managed Service for Apache Kafka 集群后，您无法更新以下属性：

集群名称
集群位置
加密类型

虽然您无法更改加密类型，但可以轮替加密密钥。

接下来怎么做？

Apache Kafka® 是 Apache Software Foundation 或其关联公司在美国和/或其他国家/地区的注册商标。