删除 Google Cloud Managed Service for Apache Kafka 主题

如需删除单个主题，您可以使用 Google Cloud 控制台、Google Cloud CLI、客户端库、Managed Kafka API 或开源 Apache Kafka API。

注意： 请勿删除名为 __remote_log_metadata 的主题。 Managed Service for Apache Kafka 会创建此主题以支持分层存储。如果您删除该文件，集群将无法读取或写入数据。

删除主题所需的角色和权限

如需获得删除主题所需的权限，请让您的管理员向您授予项目的 Managed Kafka Topic Editor(roles/managedkafka.topicEditor) IAM 角色。 如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

此预定义角色包含删除主题所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

您需要具备以下权限才能删除主题：

• 删除主题： managedkafka.topics.delete

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

如需详细了解此角色，请参阅 Managed Service for Apache Kafka 预定义角色。

删除主题

删除主题是不可逆的操作，会导致存储在该主题中的所有数据永久丢失。请确保您已进行适当的备份或导出任何必要的数据，然后再继续操作。

在删除主题之前，请确保停止或重新配置订阅了该主题的所有使用方，以使其从其他主题进行消费。

如需删除主题，请按照以下步骤操作：

控制台

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

   转到“集群”

2. 在集群列表中，点击要删除的主题所属的集群。

   系统会打开集群详情页面。在集群详情页面中，资源标签页会列出主题。

3. 点击要删除的主题。

   主题详细信息 页面打开。

4. 点击删除并确认操作。

gcloud

1. 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.

2. 运行 gcloud managed-kafka topics delete 命令：

   gcloud managed-kafka topics delete TOPIC_ID \
       --cluster=CLUSTER_ID \
       --location=LOCATION_ID
   

   此命令会从指定的 Google Cloud Managed Service for Apache Kafka 集群中移除指定的主题。与该主题关联的所有数据都会被删除，并且生产者或消费者无法再访问该主题。

   替换以下内容：

   • TOPIC_ID：要删除的主题的 ID。
   • CLUSTER_ID：主题所在集群的 ID。
   • LOCATION_ID：集群的位置。

REST

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

• PROJECT_ID：您的 Google Cloud 项目 ID
• LOCATION：集群的位置
• CLUSTER_ID：集群的 ID
• TOPIC_ID：主题的 ID

HTTP 方法和网址：

DELETE https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID

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

curl（Linux、macOS 或 Cloud Shell）

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

执行以下命令：

curl -X DELETE \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID"

PowerShell (Windows)

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

执行以下命令：

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

Invoke-WebRequest `
    -Method DELETE `
    -Headers $headers `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID" | Select-Object -Expand Content

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

{}

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 deleteTopic(w io.Writer, projectID, region, clusterID, topicID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	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)
	topicPath := fmt.Sprintf("%s/topics/%s", clusterPath, topicID)
	req := &managedkafkapb.DeleteTopicRequest{
		Name: topicPath,
	}
	if err := client.DeleteTopic(ctx, req); err != nil {
		return fmt.Errorf("client.DeleteTopic got err: %w", err)
	}
	fmt.Fprint(w, "Deleted topic\n")
	return nil
}

Java

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

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

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.TopicName;
import java.io.IOException;

public class DeleteTopic {

  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 topicId = "my-topic";
    deleteTopic(projectId, region, clusterId, topicId);
  }

  public static void deleteTopic(String projectId, String region, String clusterId, String topicId)
      throws Exception {
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      // This operation is being handled synchronously.
      managedKafkaClient.deleteTopic(TopicName.of(projectId, region, clusterId, topicId));
      System.out.println("Deleted topic");
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.deleteTopic 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 NotFound
from google.cloud import managedkafka_v1

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

client = managedkafka_v1.ManagedKafkaClient()

topic_path = client.topic_path(project_id, region, cluster_id, topic_id)
request = managedkafka_v1.DeleteTopicRequest(name=topic_path)

try:
    client.delete_topic(request=request)
    print("Deleted topic")
except NotFound as e:
    print(f"Failed to delete topic {topic_id} with error: {e.message}")

后续步骤

• 监控主题

• 查看使用方群组。

• 列出集群的所有消费群组。

• 删除使用方群组。