列出 Google Cloud Managed Service for Apache Kafka 叢集

如要列出叢集，可以使用 Google Cloud 控制台、Google Cloud CLI、用戶端程式庫或 Managed Kafka API。您無法使用開放原始碼 Apache Kafka API 列出叢集。

列出叢集所需的角色和權限

如要取得列出叢集所需的權限，請要求管理員授予您專案的代管 Kafka 檢視者 (roles/managedkafka.viewer) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

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

所需權限

您必須具備下列權限，才能列出叢集：

  • 列出叢集：managedkafka.clusters.list
  • 取得叢集詳細資料：managedkafka.clusters.get

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

列出叢集

如要列出專案中的所有叢集，請按照下列步驟操作：

控制台

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

    前往「Clusters」(叢集)

  2. 系統會列出您在專案中建立的叢集。這個頁面會顯示下列叢集屬性：

    • 「Cluster name」(叢集名稱)：Managed Service for Apache Kafka 叢集的專屬 ID。您可以在各種作業中使用這個名稱來參照叢集。
    • 叢集狀態：指出叢集目前的運作狀態，例如「Active」(有效)。
    • 區域：託管叢集的地理位置。
    • 啟動網址：Kafka 用戶端用來連線至叢集的啟動位址。
    • 記憶體：分配給叢集的記憶體總量。這會決定處理訊息流量和儲存空間的容量。
    • vCPU：指派給叢集的 vCPU 數量。這會影響叢集的處理能力和處理並行作業的能力。
    • 標籤：可附加至叢集的鍵/值組合，用於整理、篩選和自動化。

  3. 選用：如要尋找與特定啟動位址相關聯的叢集，請在「Filter」(篩選器) 欄位中輸入啟動網址。

gcloud

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

    啟用 Cloud Shell

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

  2. 執行 gcloud managed-kafka clusters list 指令：

    gcloud managed-kafka clusters list --location=LOCATION \
    --limit=LIMIT

    如要找出與特定啟動位址相關聯的叢集，請依 bootstrapAddress 欄位篩選：

    gcloud managed-kafka clusters list --location=LOCATION \
    --filter="bootstrapAddress:BOOTSTRAP_ADDRESS"

    更改下列內容：

    • LOCATION：叢集的位置 (區域)，例如 us-central1。區域也會包含在啟動網址中，以點分隔的第三個區段。
    • LIMIT：要列出的叢集數量上限。
    • BOOTSTRAP_ADDRESS：叢集的啟動網址或主機名稱。

REST

使用任何要求資料之前，請先修改下列項目的值：

  • PROJECT_ID：您的 Google Cloud 專案 ID
  • LOCATION：叢集的位置

HTTP 方法和網址：

GET https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters

請展開以下其中一個選項，以傳送要求：

您應該會收到如下的 JSON 回覆：

{
  "clusters": [
    {
      "name": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
      "createTime": "CREATE_TIME",
      "updateTime": "UPDATE_TIME",
      "capacityConfig": {
        "vcpuCount": "CPU_COUNT",
        "memoryBytes": "MEMORY"
      },
      "rebalanceConfig": {},
      "gcpConfig": {
        "accessConfig": {
          "networkConfigs": [
            {
              "subnet": "projects/PROJECT_ID/locations/LOCATION/subnetworks/SUBNET_ID"
            }
          ]
        }
      },
      "state": "ACTIVE",
      "satisfiesPzi": false,
      "satisfiesPzs": false,
      "tlsConfig": {
        "trustConfig": {}
      },
      "updateOptions": {}
    }
  ]
}

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/iterator"
	"google.golang.org/api/option"

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

func listClusters(w io.Writer, projectID, region string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	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)
	req := &managedkafkapb.ListClustersRequest{
		Parent: locationPath,
	}
	clusterIter := client.ListClusters(ctx, req)
	for {
		res, err := clusterIter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("clusterIter.Next() got err: %w", err)
		}
		fmt.Fprintf(w, "Got cluster: %v", res)
	}
	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.Cluster;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import java.io.IOException;

public class ListClusters {

  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
    listClusters(projectId, region);
  }

  public static void listClusters(String projectId, String region) throws Exception {
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      LocationName locationName = LocationName.of(projectId, region);
      // This operation is being handled synchronously.
      for (Cluster cluster : managedKafkaClient.listClusters(locationName).iterateAll()) {
        System.out.println(cluster.getAllFields());
      }
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.listClusters got err: %s", e.getMessage());
    }
  }
}

Python

試用這個範例前，請先按照「 安裝用戶端程式庫」中的 Python 設定操作說明進行。詳情請參閱 Managed Service for Apache Kafka Python API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定 ADC」。

from google.cloud import managedkafka_v1

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"

client = managedkafka_v1.ManagedKafkaClient()

request = managedkafka_v1.ListClustersRequest(
    parent=client.common_location_path(project_id, region),
)

response = client.list_clusters(request=request)
for cluster in response:
    print("Got cluster:", cluster)

後續步驟

Apache Kafka® 是 Apache Software Foundation 或其關係企業在美國和/或其他國家/地區的註冊商標。