建立 Google Cloud Managed Service for Apache Kafka 主題

在 Managed Service for Apache Kafka 中，訊息會依主題分類。主題由多個分割區組成。分區是 Kafka 叢集中單一代理程式擁有的有序且不可變更記錄序列。您必須先建立主題，才能發布或調用訊息。

如要建立主題，可以使用 Google Cloud 控制台、Google Cloud CLI、用戶端程式庫、Managed Kafka API 或開放原始碼 Apache Kafka API。

事前準備

您必須先建立叢集，才能建立主題。請確認已完成下列設定：

Managed Service for Apache Kafka 叢集。

建立主題所需的角色和權限

如要取得建立主題所需的權限，請要求管理員授予您專案的代管 Kafka 主題編輯者 (roles/managedkafka.topicEditor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

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

所需權限

如要建立主題，您必須具備下列權限：

建立主題： managedkafka.topics.create

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

代管 Kafka 主題編輯者角色也包含代管 Kafka 檢視者角色。如要進一步瞭解這個角色，請參閱「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 資源命名指南」。主題名稱無法變更。

分區數量

主題中的分割區數量。您可以編輯主題，增加主題的分區數量，但無法減少。如果主題使用索引鍵，增加分區數量可能會改變訊息的分配方式。

複製係數

每個分割區的副本數量。如未指定值，系統會使用叢集的預設複寫因數。

複製因數越高，資料複製到多個代理程式時，代理程式發生故障時的資料一致性就越好。在正式環境中，建議使用 3 以上的複寫因數。副本數量越多，主題的本機儲存空間和資料傳輸成本就越高。不過，這些作業不會增加永久儲存空間的費用。複寫因數不得超過可用代理程式數量。

其他參數

您也可以設定其他 Apache Kafka 主題層級設定參數。這些項目會指定為 key=value 配對，以覆寫叢集預設值。

主題相關設定包含伺服器預設值，以及可選用的主題專屬覆寫設定。格式為以半形逗號分隔的 KEY=VALUE 配對清單，其中 KEY 是 Kafka 主題設定屬性的名稱，VALUE 是必要設定。這些鍵/值配對可協助您覆寫叢集預設值。例如 flush.ms=10 和 compression.type=producer。

如要查看支援主題層級設定的完整清單，請參閱 Apache Kafka 說明文件中的「主題層級設定」。

建立主題

建立主題前，請先查看主題屬性。

控制台

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

前往「Clusters」(叢集)
按一下要建立主題的叢集。

「叢集詳細資料」頁面隨即開啟。
在叢集詳細資料頁面中，按一下「Create Topic」(建立主題)。

「建立 Kafka 主題」頁面隨即開啟。
在「Topic name」(主題名稱) 中輸入字串。
在「分割區計數」中，輸入所需分割區數量或保留預設值。
針對「複製係數」，請輸入所需複製係數，或保留預設值。
(選用) 如要變更任何主題設定，請在「設定」欄位中，以逗號分隔的方式新增鍵值組。
點選「建立」。

gcloud

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.
執行 gcloud managed-kafka topics create 指令：
```
gcloud managed-kafka topics create TOPIC_ID \
    --cluster=CLUSTER_ID --location=LOCATION_ID \
    --partitions=PARTITIONS \
    --replication-factor=REPLICATION_FACTOR \
    --configs=CONFIGS
```
更改下列內容：
- TOPIC_ID：主題名稱。
- CLUSTER：要在其中建立主題的叢集名稱。
- LOCATION：叢集區域。
- PARTITIONS：主題的分區數量。
- REPLICATION_FACTOR：主題的複寫因數。
- CONFIGS：主題層級的選用參數。以半形逗號分隔的鍵/值組合形式指定。例如：compression.type=producer。

Kafka CLI

執行這項指令前，請先在 Compute Engine VM 上安裝 Kafka 指令列工具。VM 必須能夠連上與 Managed Service for Apache Kafka 叢集連線的子網路。請按照「使用 Kafka 指令列工具產生及取用訊息」中的操作說明進行操作。

執行 kafka-topics.sh 指令，如下所示：

kafka-topics.sh --create --if-not-exists \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --topic TOPIC_ID \
  --partitions PARTITIONS \
  --replication-factor REPLICATION_FACTOR

更改下列內容：

BOOTSTRAP_ADDRESS：Managed Service for Apache Kafka 叢集的啟動位址。
TOPIC_ID：主題名稱。
PARTITIONS：主題的分區數量。
REPLICATION_FACTOR：主題的複製係數。

REST

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

PROJECT_ID：您的 Google Cloud 專案 ID
LOCATION：叢集位置
CLUSTER_ID：叢集 ID
TOPIC_ID：主題的 ID
PARTITION_COUNT：主題的分區數量
REPLICATION_FACTOR：每個分割區的副本數量

HTTP 方法和網址：

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics?topicId=TOPIC_ID

JSON 要求主體：

{
  "name": "TOPIC_ID",
  "partitionCount": PARTITION_COUNT,
  "replicationFactor": REPLICATION_FACTOR
}

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

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/CLUSTER_ID/topics?topicId=TOPIC_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/CLUSTER_ID/topics?topicId=TOPIC_ID" | Select-Object -Expand Content

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

{
  "name": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID",
  "partitionCount": PARTITION_COUNT,
  "replicationFactor": REPLICATION_FACTOR
}

Terraform

您可以使用 Terraform 資源建立主題。

resource "google_managed_kafka_topic" "default" {
  project            = data.google_project.default.project_id # Replace this with your project ID in quotes
  topic_id           = "my-topic-id"
  cluster            = google_managed_kafka_cluster.default.cluster_id
  location           = "us-central1"
  partition_count    = 2
  replication_factor = 3
}

如要瞭解如何套用或移除 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 createTopic(w io.Writer, projectID, region, clusterID, topicID string, partitionCount, replicationFactor int32, configs map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	// partitionCount := 10
	// replicationFactor := 3
	// configs := map[string]string{"min.insync.replicas":"1"}
	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)
	topicConfig := &managedkafkapb.Topic{
		Name:              topicPath,
		PartitionCount:    partitionCount,
		ReplicationFactor: replicationFactor,
		Configs:           configs,
	}

	req := &managedkafkapb.CreateTopicRequest{
		Parent:  clusterPath,
		TopicId: topicID,
		Topic:   topicConfig,
	}
	topic, err := client.CreateTopic(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Created topic: %s\n", topic.Name)
	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.ClusterName;
import com.google.cloud.managedkafka.v1.CreateTopicRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.Topic;
import com.google.cloud.managedkafka.v1.TopicName;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreateTopic {

  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";
    int partitionCount = 100;
    int replicationFactor = 3;
    Map<String, String> configs =
        new HashMap<String, String>() {
          {
            put("min.insync.replicas", "2");
          }
        };
    createTopic(projectId, region, clusterId, topicId, partitionCount, replicationFactor, configs);
  }

  public static void createTopic(
      String projectId,
      String region,
      String clusterId,
      String topicId,
      int partitionCount,
      int replicationFactor,
      Map<String, String> configs)
      throws Exception {
    Topic topic =
        Topic.newBuilder()
            .setName(TopicName.of(projectId, region, clusterId, topicId).toString())
            .setPartitionCount(partitionCount)
            .setReplicationFactor(replicationFactor)
            .putAllConfigs(configs)
            .build();
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      CreateTopicRequest request =
          CreateTopicRequest.newBuilder()
              .setParent(ClusterName.of(projectId, region, clusterId).toString())
              .setTopicId(topicId)
              .setTopic(topic)
              .build();
      // This operation is being handled synchronously.
      Topic response = managedKafkaClient.createTopic(request);
      System.out.printf("Created topic: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.createTopic 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 AlreadyExists
from google.cloud import managedkafka_v1

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# topic_id = "my-topic"
# partition_count = 10
# replication_factor = 3
# configs = {"min.insync.replicas": "1"}

client = managedkafka_v1.ManagedKafkaClient()

topic = managedkafka_v1.Topic()
topic.name = client.topic_path(project_id, region, cluster_id, topic_id)
topic.partition_count = partition_count
topic.replication_factor = replication_factor
# For a list of configs, one can check https://kafka.apache.org/documentation/#topicconfigs
topic.configs = configs

request = managedkafka_v1.CreateTopicRequest(
    parent=client.cluster_path(project_id, region, cluster_id),
    topic_id=topic_id,
    topic=topic,
)

try:
    response = client.create_topic(request=request)
    print("Created topic:", response.name)
except AlreadyExists as e:
    print(f"Failed to create topic {topic.name} with error: {e.message}")

建立 Google Cloud Managed Service for Apache Kafka 主題 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

事前準備

建立主題所需的角色和權限

所需權限

Managed Service for Apache Kafka 主題的屬性

主題名稱

分區數量

複製係數

其他參數

建立主題

控制台

gcloud

Kafka CLI

REST

curl (Linux、macOS 或 Cloud Shell)

PowerShell (Windows)

Terraform

Go

Java

Python

後續步驟

建立 Google Cloud Managed Service for Apache Kafka 主題