创建 Google Cloud Managed Service for Apache Kafka 集群

Managed Service for Apache Kafka 集群提供了一个环境，用于存储和处理组织成主题的消息流。

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

准备工作

验证您是否熟悉以下内容：

创建集群所需的角色和权限

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

此预定义角色可提供创建集群所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

您需要具备以下权限才能创建集群：

  • 创建集群： managedkafka.clusters.create

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

Managed Service for Apache Kafka 集群的属性

创建或更新 Managed Service for Apache Kafka 集群时，您必须指定以下属性。

集群名称

您正在创建的 Managed Service for Apache Kafka 集群的名称或 ID。有关如何命名集群的指南，请参阅 Managed Service for Apache Kafka 资源命名指南。集群的名称不可变。

位置

您要创建集群的位置。该位置必须是受支持的 Google Cloud 区域之一。集群的位置以后无法更改。如需获取可用位置列表，请参阅 Managed Service for Apache Kafka 位置。

容量配置

Kafka 集群的计算、内存和存储空间容量。创建集群时，您必须指定以下值：

  • vCPUs：集群中的 vCPU 数量。每个集群至少需要 3 个 vCPU。

  • 内存：分配给集群的内存量。您必须为每个 vCPU 预配 1 GiB 到 8 GiB 之间的内存。

    例如，如果您创建了一个具有 6 个 vCPU 的集群，则可为该集群分配的最小内存为 6 GiB（每个 vCPU 1 GiB），最大内存为 48 GiB（每个 vCPU 8 GiB）。

该服务会自动预配 broker，包括 broker 的 Persistent Disk 卷。您可以选择启用自定义本地存储空间，以便为每个代理指定磁盘容量。如需了解详情，请参阅配置代理磁盘大小。

如需了解有关估算集群所需容量的指南，请参阅规划 Kafka 集群的大小。

网络配置

网络配置是集群可访问的 VPC 子网的列表。如需生成或使用消息，客户端必须能够访问这些子网之一。

以下是一些网络配置指南：

  • 一个集群至少需要 1 个子网。最大值为 10。

  • 对于任何给定的集群，每个网络只允许有 1 个子网。

  • 每个子网都必须与集群位于同一区域。项目和网络可以不同。

  • 代理和引导服务器的 IP 地址在每个子网中自动分配。此外，系统会在相应的 VPC 网络中创建这些 IP 地址的 DNS 条目。

  • 如果您添加了来自其他项目的子网，则必须向与集群关联的 Google 代管式服务账号授予权限。如需了解详情，请参阅跨项目连接集群。

创建集群后，您可以更新子网列表。如需详细了解网络，请参阅为 Managed Service for Apache Kafka 配置网络。

公共集群

您可以选择将集群配置为公共集群，以允许客户端应用通过公共互联网连接到集群并运行 Kafka 操作（包括发送和接收消息）。如需启用公开访问权限，您必须指定一个或多个允许的来源 IP 范围。

配置允许的来源 IP 范围时，请遵循以下规则：

  • 所有允许的源 IP 范围都必须采用 IPv4 CIDR 表示法（例如 203.0.113.0/24 或 198.51.100.5/32）指定。

  • 每个 CIDR 子网的大小必须介于 /16 和 /32 之间。

  • CIDR 范围不得重叠。

  • 所有允许的来源 IP 地址范围都必须可公开路由。系统会拒绝专用 IP 范围，例如 10.0.0.0/8 (RFC 1918)。

  • 您最多可以指定 500 个允许的来源 IP 范围。

  • 不支持 IPv6 地址。

在为集群启用公开访问之前，我们建议您阅读将客户端连接到公开集群，了解公开访问功能的实现方式。

标签

标签是键值对，可帮助您进行组织和标识。 借助标签，您可以根据环境对资源进行分类。例如，"env:production" 和 "owner:data-engineering"。

您可以根据资源的标签过滤和搜索资源。例如，假设您有多个适用于不同部门的 Managed Service for Apache Kafka 集群。您可以配置并搜索带有 "department:marketing" 标签的集群，以找到相关集群。

重新平衡配置

此设置决定了服务是否会自动在代理之间重新平衡分区副本。

可用的模式包括：

  • 在扩缩时自动重新平衡：启用此选项后，当您扩缩集群时，服务会自动触发副本重新平衡。此模式有助于保持均匀的负载分配，但在重新平衡操作期间可能会暂时影响性能。

  • 不重新平衡：启用此选项后，服务不会自动重新平衡副本。

如需了解详情，请参阅重新平衡代理分区。

加密

Managed Service for Apache Kafka 可以使用Google-owned and Google-managed encryption keys （默认）或客户管理的加密密钥 (CMEK) 对消息进行加密。每条消息在静态加密时和传输过程中都会加密。集群的加密类型不可变。

默认情况下使用 Google-owned and Google-managed encryption keys。这些密钥完全由 Google Cloud 在其基础架构中创建、管理和存储。

CMEK 是指您使用 Cloud Key Management Service 管理的加密密钥。借助此功能，您可以更好地控制用于在受支持的 Google Cloud 服务中加密静态数据的密钥。使用 CMEK 会产生与 Cloud Key Management Service 相关的额外费用。对于 CMEK 用途，您的密钥环必须与您使用它的资源位于同一位置。如需了解详情，请参阅配置消息加密。

mTLS 配置

您可以选择将 mTLS 配置为使用客户端证书的替代身份验证方法。配置包括以下内容：

  • CA 池：集群信任的 1 到 10 个 Certificate Authority Service (CAS) 池，用于客户端身份验证。

  • SSL 主账号映射规则：一种可选但建议使用的 ssl.principal.mapping.rules 代理属性，用于简化 Kafka ACL 中使用的长证书主体名称。

如需详细了解 mTLS，请参阅配置 mTLS 身份验证。

创建集群

在创建集群之前，请先阅读集群属性的相关文档。

创建集群通常需要 20-30 分钟。

如需创建集群，请按照以下步骤操作：

控制台

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

    转到“集群”

  2. 选择创建。

    系统会打开 Create Kafka cluster 页面。

  3. 对于集群名称，请输入一个字符串。

    如需详细了解如何命名集群，请参阅 Managed Service for Apache Kafka 资源命名指南。

  4. 在位置部分中，输入受支持的位置。

    如需详细了解支持的位置，请参阅支持的 Managed Service for Apache Kafka 位置。

  5. 对于容量配置，请输入内存和 vCPU 的值。

    如需详细了解如何确定集群大小，请参阅规划 Kafka 集群大小。

  6. 可选。如需指定代理磁盘容量，请执行以下操作：

    1. 点击启用自定义本地存储空间大小切换开关，以切换到开启位置。

    2. 在自定义本地存储空间字段中，输入每个代理的磁盘大小（以 GiB 为单位）。

    如果您将启用自定义本地存储空间大小切换开关保持在关闭位置，系统会预配默认磁盘容量。如需了解详情，请参阅配置代理磁盘大小。

  7. 对于网络配置，请输入以下详细信息：

    1. 项目：子网所在的项目。子网必须与集群位于同一区域，但项目可能不同。
    2. 网络：子网所连接到的网络。
    3. 子网：子网的名称。
    4. 子网 URI 路径：此字段会自动填充。或者，您也可以在此处输入子网路径。子网的名称必须采用以下格式：projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID。
    5. 点击完成。

  8. 可选：如需添加其他子网，请点击添加已连接的子网。

    您可以添加其他子网，最多可添加 10 个。

  9. 可选：启用对集群的公开访问权限。

    1. 在公共集群配置部分中，在集群可访问下，选择在专用网络中并使用公共互联网。

    2. 在允许的来源 IPv4 范围的逗号分隔列表（采用 CIDR 表示法）字段中，以 IPv4 CIDR 表示法输入客户端应用的 IP 范围（例如 203.0.113.0/24）。不支持 IPv6 地址。

  10. 可选：设置客户管理的加密密钥 (CMEK)。

    1. 在加密下，选择 Cloud KMS 密钥。

    2. 在密钥类型中，选择 Cloud KMS。

    3. 对于选择客户管理的密钥，请选择或输入 CMEK。

      如果您没有看到此选项，请确保您已为项目启用 Cloud KMS API。

    如需了解详情，请参阅配置消息加密。

  11. 点击创建。

gcloud

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

    激活 Cloud Shell

    Cloud Shell 会话随即会在 Google Cloud 控制台的底部启动，并显示命令行提示符。Cloud Shell 是一个已安装 Google Cloud CLI 且已为当前项目设置值的 Shell 环境。该会话可能需要几秒钟时间来完成初始化。

  2. 在使用下面的命令数据之前，请先进行以下替换：

    • PROJECT_ID：项目 ID。
    • LOCATION：集群的位置。
    • CLUSTER_ID：集群的 ID。
    • CPU_COUNT：集群的 vCPU 数量。
    • MEMORY：集群的内存量。示例：3GiB。
    • SUBNET_ID：要连接到的子网的子网 ID。示例：default。
    • ENCRYPTION_KEY：要用于集群的客户管理的加密密钥 ID。此参数为可选参数。
    • LABELS：要与集群关联的标签。
    • ALLOWED_SOURCE_IP_RANGES：公共集群的允许源 IP 范围。此参数为可选参数。

    执行以下命令：

    Linux、macOS 或 Cloud Shell

    gcloud managed-kafka clusters create CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU_COUNT \
    --memory=MEMORY \
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
    --auto-rebalance \
    --encryption-key=ENCRYPTION_KEY \
    --async \
    --labels=LABELS \
    --public-cluster \
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

    Windows (PowerShell)

    gcloud managed-kafka clusters create CLUSTER_ID `
    --location=LOCATION `
    --cpu=CPU_COUNT `
    --memory=MEMORY `
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
    --auto-rebalance `
    --encryption-key=ENCRYPTION_KEY `
    --async `
    --labels=LABELS `
    --public-cluster `
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

    Windows (cmd.exe)

    gcloud managed-kafka clusters create CLUSTER_ID ^
    --location=LOCATION ^
    --cpu=CPU_COUNT ^
    --memory=MEMORY ^
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
    --auto-rebalance ^
    --encryption-key=ENCRYPTION_KEY ^
    --async ^
    --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: create
name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID

    存储 OPERATION_ID 以跟踪 progress。

REST

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

  • PROJECT_ID：您的 Google Cloud 项目 ID
  • LOCATION：集群的位置。
  • CLUSTER_ID：集群的 ID。
  • CPU_COUNT：集群的 vCPU 数量。
  • MEMORY：集群的内存量（以字节为单位）。示例：3221225472。
  • SUBNET_ID：要连接到的子网的子网 ID。示例：default。

HTTP 方法和网址：

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID

请求 JSON 正文：

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

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

您应该收到类似以下内容的 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": "create",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Terraform

您可以使用 Terraform 资源创建集群。

resource "google_managed_kafka_cluster" "default" {
  project    = data.google_project.default.project_id # Replace this with your project ID in quotes
  cluster_id = "my-cluster-id"
  location   = "us-central1"
  capacity_config {
    vcpu_count   = 3
    memory_bytes = 3221225472
  }
  gcp_config {
    access_config {
      network_configs {
        subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

如需了解如何应用或移除 Terraform 配置，请参阅基本 Terraform 命令。

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"

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

func createCluster(w io.Writer, projectID, region, clusterID, subnet string, cpu, memoryBytes int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// subnet := "projects/my-project-id/regions/us-central1/subnetworks/default"
	// cpu := 3
	// memoryBytes := 3221225472
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

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

	// Memory must be between 1 GiB and 8 GiB per CPU.
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   cpu,
		MemoryBytes: memoryBytes,
	}
	var networkConfig []*managedkafkapb.NetworkConfig
	networkConfig = append(networkConfig, &managedkafkapb.NetworkConfig{
		Subnet: subnet,
	})
	platformConfig := &managedkafkapb.Cluster_GcpConfig{
		GcpConfig: &managedkafkapb.GcpConfig{
			AccessConfig: &managedkafkapb.AccessConfig{
				NetworkConfigs: networkConfig,
			},
		},
	}
	rebalanceConfig := &managedkafkapb.RebalanceConfig{
		Mode: managedkafkapb.RebalanceConfig_AUTO_REBALANCE_ON_SCALE_UP,
	}
	cluster := &managedkafkapb.Cluster{
		Name:            clusterPath,
		CapacityConfig:  capacityConfig,
		PlatformConfig:  platformConfig,
		RebalanceConfig: rebalanceConfig,
	}

	req := &managedkafkapb.CreateClusterRequest{
		Parent:    locationPath,
		ClusterId: clusterID,
		Cluster:   cluster,
	}
	op, err := client.CreateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 10-40 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created cluster: %s\n", resp.Name)
	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.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.AccessConfig;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.CreateClusterRequest;
import com.google.cloud.managedkafka.v1.GcpConfig;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.NetworkConfig;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.RebalanceConfig;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateCluster {

  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";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    int cpu = 3;
    long memoryBytes = 3221225472L; // 3 GiB
    createCluster(projectId, region, clusterId, subnet, cpu, memoryBytes);
  }

  public static void createCluster(
      String projectId, String region, String clusterId, String subnet, int cpu, long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig =
        CapacityConfig.newBuilder().setVcpuCount(cpu).setMemoryBytes(memoryBytes).build();
    NetworkConfig networkConfig = NetworkConfig.newBuilder().setSubnet(subnet).build();
    GcpConfig gcpConfig =
        GcpConfig.newBuilder()
            .setAccessConfig(AccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
            .build();
    RebalanceConfig rebalanceConfig =
        RebalanceConfig.newBuilder()
            .setMode(RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP)
            .build();
    Cluster cluster =
        Cluster.newBuilder()
            .setCapacityConfig(capacityConfig)
            .setGcpConfig(gcpConfig)
            .setRebalanceConfig(rebalanceConfig)
            .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.createClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {

      CreateClusterRequest request =
          CreateClusterRequest.newBuilder()
              .setParent(LocationName.of(projectId, region).toString())
              .setClusterId(clusterId)
              .setCluster(cluster)
              .build();

      // The duration of this operation can vary considerably, typically taking between 10-40
      // minutes.
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.createClusterOperationCallable().futureCall(request);

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

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone =  True)
      Cluster response = future.get();
      System.out.printf("Created cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.createCluster 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

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 3
# memory_bytes = 3221225472

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.vcpu_count = cpu
cluster.capacity_config.memory_bytes = memory_bytes
cluster.gcp_config.access_config.network_configs = [
    managedkafka_v1.NetworkConfig(subnet=subnet)
]
cluster.rebalance_config.mode = (
    managedkafka_v1.RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP
)

request = managedkafka_v1.CreateClusterRequest(
    parent=client.common_location_path(project_id, region),
    cluster_id=cluster_id,
    cluster=cluster,
)

try:
    operation = client.create_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # The duration of this operation can vary considerably, typically taking 10-40 minutes.
    # We can set a timeout of 3000s (50 minutes).
    response = operation.result(timeout=3000)
    print("Created cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

监控集群创建操作

只有在您使用 gcloud CLI 创建集群时，才能运行以下命令。

  • 创建集群通常需要 20-30 分钟。为了跟踪集群创建的进度，gcloud managed-kafka clusters create 命令使用长时间运行的操作 (LRO)，您可以使用以下命令监控该操作：

    gcloud managed-kafka operations describe OPERATION_ID \
    --location=LOCATION

    替换以下内容：

    • OPERATION_ID 替换为上一部分中的操作 ID 值。
    • 将 LOCATION 替换为上一部分中的位置值。

问题排查

如需了解如何排查与创建集群相关的错误，请参阅 Kafka 集群创建错误。

接下来怎么做？

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