刪除 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 主題編輯者(roles/managedkafka.topicEditor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和機構的存取權」。

這個預先定義的角色具備刪除主題所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

如要刪除主題，您必須具備下列權限：

• 刪除主題： managedkafka.topics.delete

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

如要進一步瞭解這個角色，請參閱「Managed Service for Apache Kafka 預先定義的角色」。

刪除主題

刪除主題是不可逆的操作，會導致該主題中儲存的所有資料永久遺失。請務必先備份或匯出所有必要資料，再繼續操作。

請務必先停止或重新設定訂閱主題的所有消費者，從其他主題取用資料，再刪除主題。

如要刪除主題，請按照下列步驟操作：

控制台

1. 前往 Google Cloud 控制台的「Clusters」(叢集) 頁面。

   前往「Clusters」(叢集)

2. 在叢集清單中，按一下要刪除的主題所屬的叢集。

   「叢集詳細資料」頁面隨即開啟。在叢集詳細資料頁面中，主題會列在「資源」分頁中。

3. 按一下要刪除的主題。

   「主題詳細資料」頁面隨即開啟。

4. 按一下 [Delete] (刪除) 並確認執行。

gcloud

1. 在 Google Cloud 控制台中啟用 Cloud Shell。

   啟用 Cloud Shell

   Google Cloud 主控台底部會開啟一個 Cloud Shell 工作階段，並顯示指令列提示。Cloud Shell 是已安裝 Google Cloud CLI 的殼層環境，並已針對您目前的專案設定好相關值。工作階段可能要幾秒鐘的時間才能初始化。

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}")

後續步驟

• 監控主題

• 查看消費者群組。

• 列出叢集的所有用戶群組。

• 刪除消費者群組。