创建 Google Cloud Managed Service for Apache Kafka 集群

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

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

准备工作

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

• Managed Service for Apache Kafka 的运作方式。

• 与 Managed Service for Apache Kafka 相关的网络概念。

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

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

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

所需权限

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

• 创建集群： managedkafka.clusters.create

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

借助 Managed Kafka Cluster Editor 角色，您无法在 Managed Service for Apache Kafka 集群上创建、删除或修改主题和使用方群组。它也不允许数据平面访问在集群内发布或使用消息。如需详细了解此角色，请参阅 Managed Service for Apache Kafka 预定义角色。

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 设置配置 vCPU 数量和内存量。如需详细了解如何配置集群容量，请参阅规划 Kafka 集群规模。

以下是容量配置的属性：

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

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

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

如需详细了解如何在创建集群后更改内存和 vCPU 数量，请参阅更新集群规模。

网络配置

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

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

• 一个集群至少需要一个子网。最多 10 个。

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

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

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

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

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

标签

标签是有助于您进行组织和标识的键值对。标签可用于根据环境对资源进行分类。例如 "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 池：集群信任的用于客户端身份验证的 Certificate Authority Service (CAS) 池的列表，其中包含 1 到 10 个 CAS 池。

• SSL 正文映射规则：一种可选但建议使用的 ssl.principal.mapping.rules 代理属性，用于简化较长的证书正文名称，以便在 Kafka ACL 中使用。

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

创建集群

在创建集群之前，请查看集群属性的相关文档。

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

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

控制台

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

   转到“集群”

2. 选择创建。

   系统会打开创建 Kafka 集群页面。

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

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

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

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

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

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

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

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

7. 可选：点击添加连接的子网，添加其他子网。

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

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

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

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

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

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

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

9. 点击创建。

gcloud

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

   激活 Cloud Shell

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

2. 运行 gcloud managed-kafka clusters create 命令：

   gcloud managed-kafka clusters create CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU \
       --memory=MEMORY \
       --subnets=SUBNETS \
       --auto-rebalance \
       --encryption-key=ENCRYPTION_KEY \
       --async \
       --labels=LABELS
   

   替换以下内容：

   • CLUSTER_ID：集群的 ID 或名称。

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

   • LOCATION：集群的位置。

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

   • CPU：集群的 vCPU 数量。

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

   • MEMORY：集群的内存量。 使用“MB”“MiB”“GB”“GiB”“TB”或“TiB”单位。例如，“10GiB”。

   • SUBNETS：要连接到的子网列表。 使用英文逗号分隔多个子网值。

     子网的格式为 projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID。

   • auto-rebalance：当集群中的 CPU 数量发生变化时，启用在代理之间自动重新平衡主题分区的功能。此功能默认处于启用状态。

   • ENCRYPTION_KEY：要用于集群的客户管理的加密密钥的 ID。

     格式为 projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/CRYPTO_KEY。

   • --async：允许系统发送创建请求并立即返回响应，而无需等待操作完成。 借助 --async 标志，您可以在后台创建集群的同时继续执行其他任务。如果不使用该标志，系统会等待操作完成，然后返回响应。您必须等到集群完全更新后，才能继续执行其他任务。

   • LABELS：要与集群关联的标签。

     如需详细了解标签的格式，请参阅标签。

   您会收到类似如下所示的响应：

   Create request issued for: [CLUSTER_ID]
   Check operation [projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID] for status.
   

   存储 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"
        }
      ]
    }
  }
}

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

curl（Linux、macOS 或 Cloud Shell）

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

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

curl -X POST \
     -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?clusterId=CLUSTER_ID"

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 POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID" | 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": "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 集群创建错误。

接下来怎么做？

• 监控集群。

• 更新集群。

• 列出所有集群。

• 查看集群。

• 删除集群。

• 在集群中创建主题。

• 为 Managed Service for Apache Kafka 配置网络

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