创建 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）。

该服务会自动预配 代理，包括 代理的 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 （创建 Kafka 集群）页面。

  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 或其关联公司在美国和/或其他国家/地区的注册商标。