Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

建立連結叢集

Connect 叢集可為連接器提供環境，協助將資料從現有 Kafka 部署項目移至 Google Cloud Managed Service for Apache Kafka 叢集，或將資料從 Managed Service for Apache Kafka 叢集移至其他 Google Cloud 服務或其他 Kafka 叢集。次要 Kafka 叢集可以是另一個 Google Cloud Managed Service for Apache Kafka 叢集、自行管理的叢集，或地端部署叢集。

事前準備

請確認您已建立 Managed Service for Apache Kafka 叢集。您需要 Managed Service for Apache Kafka 叢集的名稱，才能連結 Connect 叢集。

每個 Connect 叢集都會與一個 Managed Service for Apache Kafka 叢集相關聯。這個叢集會儲存 Connect 叢集上執行的連接器狀態。

建立 Connect 叢集所需的角色和權限

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

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

所需權限

必須具備下列權限，才能建立 Connect 叢集：

授予在指定位置建立 Connect 叢集的權限：managedkafka.connectClusters.create

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

必要的 ACL 主體

根據預設，如果未設定 ACL，Managed Service for Apache Kafka 叢集會允許 Connect 叢集存取資源。方法是將 allow.everyone.if.no.acl.found 設為 true (預設設定)。

不過，如果 Managed Service for Apache Kafka 叢集已設定 ACL，Connect 叢集不會自動具備資源的讀寫權限。您必須手動授予。

在 ACL 中做為主體的 Connect 叢集服務帳戶，格式如下：User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com。

如果您已在 Kafka 叢集設定 ACL，請使用下列指令，授予 Connect 叢集主題的讀寫權限，以及消費者群組的讀取權限：

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --operation WRITE --topic *

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --group *

如要進一步瞭解這些指令，請參閱「設定 Apache Kafka ACL，精細控管存取權」。

在其他專案中建立 Connect 叢集

Managed Service for Apache Kafka 會使用服務代理存取Google Cloud 資源。服務代理與您建立叢集的專案相關聯。

如果建立 Connect 叢集的專案與 Managed Service for Apache Kafka 叢集不同，則 Connect 叢集和 Kafka 叢集會使用與各自專案相關聯的服務代理。在這種情況下，Connect 叢集的服務代理必須具備 Kafka 叢集專案中 Google Cloud 資源的存取權限。

如要授予必要權限，請將代管 Kafka 服務代理角色，授予 Kafka 叢集專案中的 Connect 叢集服務代理。舉例來說，如果您在專案 kafka-project 中建立 Kafka 叢集，並在專案 connect-project 中建立 Connect 叢集，請將 kafka-project 的 Managed Kafka 服務代理角色授予與 connect-project 相關聯的服務代理。

服務代理的電子郵件地址格式如下：service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com，其中 PROJECT_NUMBER 是專案編號。如要進一步瞭解如何授予角色，請參閱「建立角色並授予服務代理」。

Connect 叢集的屬性

本節說明 Connect 叢集的屬性。

Connect 叢集名稱

要建立的 Connect 叢集名稱，如要查看 Connect 叢集的命名準則，請參閱「Managed Service for Apache Kafka 資源命名指南」。叢集名稱無法變更。

主要 Kafka 叢集

與 Connect 叢集相關聯的 Managed Service for Apache Kafka 叢集。這個相關聯的叢集 (主要叢集) 會儲存 Connect 叢集上執行的連接器狀態。一般來說，主要 Managed Service for Apache Kafka 叢集也是所有來源連接器的目的地，以及在 Connect 叢集上執行的所有接收器連接器的輸入來源。

單一 Managed Service for Apache Kafka 叢集可以有多個 Connect 叢集。如果選擇其他專案中的 Managed Service for Apache Kafka 叢集，請務必設定適當權限。

建立 Connect 叢集後，就無法更新為其他 Kafka 叢集。

區域共置對延遲時間和網路費用的好處

將 Managed Service for Apache Kafka 和 Connect 叢集放在同一區域，可減少延遲時間和網路費用。舉例來說，假設 Managed Service for Apache Kafka 叢集位於 region-a，而您使用接收器連接器，將資料從這個 Managed Service for Apache Kafka 叢集 (來源) 寫入同樣位於 region-a 的 BigQuery 資料表 (接收器)。如果您在 region-a 部署 Connect 叢集，這項部署選項可盡量縮短 BigQuery 寫入作業的延遲時間，並消除 Managed Service for Apache Kafka 叢集與 Connect 叢集之間的跨區域網路傳輸費用。

多系統延遲時間和成本考量

Kafka Connect 會使用連接器在系統之間移動資料。連接器的一端一律會與 Managed Service for Apache Kafka 叢集互動。單一 Kafka Connect 叢集可以執行多個連接器，每個連接器可做為「來源」 (從系統提取資料) 或「接收器」 (將資料推送至系統)。

如果 Connect 叢集與 Managed Service for Apache Kafka 叢集位於相同區域，兩者之間的通訊延遲時間會較短，但每個連接器也會與其他系統互動，例如 BigQuery 資料表或其他 Kafka 叢集。即使 Connect 叢集和 Managed Service for Apache Kafka 叢集位於同一位置，其他系統也可能位於不同區域。這會導致延遲時間和成本增加。整體管道延遲時間取決於這三個系統的位置：Managed Service for Apache Kafka 叢集、Connect 叢集，以及來源或接收器系統。

舉例來說，如果 Managed Service for Apache Kafka 叢集位於 region-a，Connect 叢集位於 region-b，且您使用 Cloud Storage 連接器連結 region-c 中的 bucket，則須支付兩次網路躍點的費用 (region-a 到 region-b，然後 region-b 到 region-c，或反向，視連接器方向而定)。

規劃 Connect 叢集位置時，請仔細考量所有相關區域，盡可能降低延遲時間和成本。

容量設定

請在「運算資源設定」專區，提供 Connect 叢集的 vCPU 數量和每個 vCPU 的記憶體容量。建立 Connect 叢集後，您可以更新其容量。以下是容量設定的屬性：

vCPU：指派給 Connect 叢集的 vCPU 數量。最小值為 3 個 vCPU。
記憶體：指派給每個 vCPU 的記憶體容量。每個 vCPU 必須佈建 1 GiB 至 8 GiB。叢集建立後，您可以在這些限制內增加或減少記憶體容量。

舉例來說，如果您建立的叢集有 6 個 vCPU，則可分配給叢集的記憶體下限為 6 GiB (每個 vCPU 1 GiB)，上限為 48 GiB (每個 vCPU 8 GiB)。

分配給 Connect 叢集中每個 worker 的 vCPU 和記憶體，會大幅影響叢集的效能、容量和費用。以下詳細說明 vCPU 和記憶體對 Connect 叢集的影響。

vCPU 數量

Kafka Connect 會將連接器的工作分成多項任務。每項工作都能平行處理資料。vCPU 越多，可同時執行的工作就越多，處理量也越高。
vCPU 數量越多，Connect 叢集的費用越高。

記憶體

Kafka Connect 會使用記憶體緩衝資料，以便在連接器和 Managed Service for Apache Kafka 之間傳輸資料。記憶體越大，緩衝區就越大。記憶體容量較大可提高處理量，尤其是處理大量資料串流時。處理大型訊息或記錄的連接器需要足夠的記憶體，才能處理這些訊息或記錄，而不會發生 OutOfMemoryError 例外狀況。
記憶體越多，Connect 叢集的費用越高。
如果您使用繁重的轉換邏輯，則需要更多記憶體配置。

您的目標是為 Connect 叢集選擇合適的容量設定。為此，您必須瞭解 Connect 叢集可處理的處理量。

Worker (主要) 子網路

worker 子網路 (也稱為主要子網路) 會將虛擬私有雲網路連線至 Connect 叢集。這個子網路可讓叢集 worker 連到用戶網路中的來源和接收器端點，例如 Managed Service for Apache Kafka 叢集或自行託管的 Kafka 叢集。

設定 worker 子網路時，請留意下列幾項規定：

必須提供 worker 子網路。
子網路必須與 Connect 叢集位於相同區域。
子網路必須與主要 Kafka 叢集已連結子網路清單中的其中一個子網路位於同一個上層虛擬私有雲。
子網路 CIDR 範圍的大小下限為 /22 (1024 個位址)。

叢集 worker 會透過 Private Service Connect 介面，獲派 worker 子網路中的 IP 位址。Worker 可連至子網路虛擬私有雲網路可存取的任何網路目的地，但須符合下列規定：

端點不得位於 172.16.0.0/14 CIDR 範圍內。這個範圍已保留給 Managed Service for Apache Kafka Connect 內部使用。
防火牆規則必須允許流量。請參閱「設定網路連結的安全性」。
如要處理網際網路流量，您必須設定 Cloud NAT。舉例來說，MirrorMaker 連接器需要 Cloud NAT，才能從可透過網際網路存取的 Kafka 叢集複製資料。
如要存取位於不同虛擬私有雲 (非 worker 子網路虛擬私有雲) 的 Private Service Connect 端點，請務必使用支援的用戶端設定 (例如 NCC)。詳情請參閱「關於透過端點存取已發布的服務」。

可解析的 DNS 網域

可解析的 DNS 網域 (也稱為 DNS 網域名稱) 可讓用戶虛擬私有雲網路中的 DNS 位址提供給租戶虛擬私有雲。這樣一來，Connect 叢集就能將 DNS 名稱解析為 IP 位址，方便與其他服務通訊，包括 MirrorMaker 連接器的其他 Kafka 叢集。

如果是可解析的 DNS 網域，您可以選取 Managed Service for Apache Kafka 叢集。您不必為主要 Managed Service for Apache Kafka 叢集設定 DNS 網域名稱，因為其啟動位址會自動納入可解析的 DNS 網域清單。

不過，您也可以手動指定 DNS 網域，如果選取外部 Kafka 叢集，就必須這麼做。系統會自動納入主要 Managed Service for Apache Kafka 叢集的 DNS 網域。其他 Kafka 叢集仍須設定 DNS 網域。

Secret Manager 資源

部分連接器在設定時需要密碼等機密資料。如要安全地管理這類資料，您可以將資料儲存在 Secret Manager 中，並授予 Connect 叢集密鑰存取權。

如要搭配使用 Secret Manager 密鑰和 Kafka Connect，請按照下列步驟操作：

將下列 IAM 角色授予代管 Kafka 服務帳戶。這些角色可讓 Connect 叢集查看密鑰及存取其酬載。
- Secret Manager 檢視者 (roles/secretmanager.viewer)
- Secret Manager 密鑰存取者 (roles/secretmanager.secretAccessor)
建立含有機密資料的 Secret。詳情請參閱「建立密鑰」。
建立或更新 Connect 叢集時，請指定叢集可存取的 Secret。每個 Connect 叢集最多可指定 32 個 Secret。

Secret 會以檔案形式掛接在叢集 worker 中。連接器對這些檔案具有唯讀存取權。建立連接器時，連接器的設定屬性可以參照密鑰。

如要參照 Secret 檔案的路徑，請使用下列格式：
```
/var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION
```
範例：ssl.truststore.location=/var/secrets/project1-truststore-1
如要將 Secret 的值做為設定值 (例如密碼)，請使用下列格式：
```
${directory:/var/secrets:PROJECT_NAME-SECRET_NAME-SECRET_VERSION}
```
範例：password=${directory:/var/secrets:project1-database_password-3}

更改下列內容：

PROJECT_NAME： Google Cloud 專案的名稱。
SECRET_NAME：Secret 的名稱。
SECRET_VERSION：Secret 版本。

建立 Connect 叢集

建立叢集前，請先參閱連結叢集屬性的說明文件。

建立 Connect 叢集需要 20 到 30 分鐘。

控制台

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

前往「Connect Clusters」(連結叢集)
按一下「Create」(建立)。
在「Connect cluster name」(Connect 叢集名稱) 欄位中，輸入 Connect 叢集的名稱。詳情請參閱「Managed Service for Apache Kafka 資源命名指南」。
在「Primary Kafka cluster」(主要 Kafka 叢集) 清單中，選取 Managed Service for Apache Kafka 叢集。詳情請參閱「主要 Kafka 叢集」。
在「Region」(區域) 清單中，選取 Connect 叢集的位置。如要進一步瞭解如何選取位置，請參閱「主要 Kafka 叢集」。
在「Capacity configuration」(容量設定) 區段中，輸入下列欄位的值，或保留預設值。
- 在「vCPUs」vCPUs欄位中，輸入叢集的虛擬 CPU 數量。
- 在「Memory」(記憶體) 欄位中，輸入每個 CPU 的記憶體容量 (以 GiB 為單位)。值不得超過每個 CPU 8 GiB。
詳情請參閱「容量設定」。
在「Network configuration」(網路設定) 區段中，從「Network」(網路) 清單選取虛擬私有雲網路，或保留預設值。選取主要 Kafka 叢集後，系統就會填入這份清單。
在「Worker subnet」(工作站子網路) 區段中，從「Subnet」(子網路) 清單選取子網路，或保留預設值。詳情請參閱「worker 子網路」。選取子網路後，「Subnet URI path」(子網路 URI 路徑) 欄位會自動填入。
選用：新增可解析的 DNS 網域。系統會自動將主要 Kafka 叢集的 DNS 網域新增為可解析的 DNS 網域。如要指定其他 DNS 網域，請按照下列步驟操作：
1. 展開「Resolvable DNS domains」(可解析的 DNS 網域) 專區。
2. 按一下「Add DNS domain」(新增 DNS 網域)。
3. 如要新增現有 Managed Service for Apache Kafka 叢集的 DNS 網域，請從「Kafka cluster」(Kafka 叢集) 清單中選取叢集。否則，請在「DNS domain」(DNS 網域) 欄位中輸入 DNS 網域。
4. 按一下「完成」。
選用：如要新增 Secret Manager 資源，請按照下列步驟操作：
1. 展開「Secret Manager 資源」部分。
2. 按一下「Add secret resource」(新增密鑰資源)。
3. 在「Secret」清單中選取密鑰。
4. 在「Secret version」(Secret 版本) 清單中，選取 Secret 版本。
5. 按一下「完成」。
選用：新增標籤來整理專案。如要新增標籤，請按照下列步驟操作：
1. 展開「標籤」部分。
2. 按一下「新增標籤」。
3. 在「Key」(鍵) 欄位中，輸入標籤的鍵。
4. 在「Value」(值) 欄位中，輸入標籤的值。
點選「建立」。

gcloud

在 Google Cloud 控制台啟用 Cloud Shell。

啟用 Cloud Shell

Google Cloud 控制台底部會開啟 Cloud Shell 工作階段，並顯示指令列提示。Cloud Shell 是已安裝 Google Cloud CLI 的殼層環境，並已設定適用於您目前專案的值。工作階段可能要幾秒鐘的時間才能初始化。
請執行 gcloud managed-kafka connect-clusters create 指令：
```
gcloud managed-kafka connect-clusters create CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU \
    --memory=MEMORY \
    --primary-subnet=WORKER_SUBNET \
    --kafka-cluster=KAFKA_CLUSTER \
    [--project=PROJECT_ID] \
    [--secret=SECRET] \
    [--dns-name=DNS_DOMAIN_NAME] \
    [--config-file=CONFIG_FILE] \
    [--labels=LABELS]
    [--async]
```
更改下列內容：
- CONNECT_CLUSTER_ID：Connect 叢集的 ID 或名稱。如要查看 Connect 叢集的命名準則，請參閱「Managed Service for Apache Kafka 資源命名指南」。Connect 叢集的名稱無法變更。
- LOCATION：建立 Connect 叢集的位置。必須是支援的 Google Cloud區域。Connect 叢集建立後即無法變更位置。如需可用位置清單，請參閱「Managed Service for Apache Kafka 位置」一文。如要進一步瞭解位置建議，請參閱「主要 Kafka 叢集」。
- CPU：Connect 叢集的 vCPU 數量。最小值為 3 個 vCPU。請參閱「vCPU 數量」。
- MEMORY：Connect 叢集的記憶體容量。請使用「MB」、「MiB」、「GB」、「GiB」、「TB」或「TiB」單位。例如「3GiB」。每個 vCPU 必須佈建 1 GiB 至 8 GiB。請參閱「記憶體」。
- WORKER_SUBNET：Connect 叢集的 worker 子網路。
  
  子網路的格式為 projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID。
  
  worker 子網路必須與 Connect 叢集位於相同區域。
- PROJECT_ID：(選用)Google Cloud 專案的 ID。如未提供，則會使用目前的專案。
- KAFKA_CLUSTER：與 Connect 叢集相關聯的主要 Managed Service for Apache Kafka 叢集 ID 或完整名稱。請參閱「Kafka 叢集」。Kafka 叢集的格式為 projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID。
  
  建立 Connect 叢集後，就無法更新為其他 Kafka 叢集。
- SECRET：(選用) 要載入 worker 的 Secret。必須提供 Secret Manager 的確切 Secret 版本，不支援別名。一個叢集最多可載入 32 個 Secret。格式：projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID
- DNS_DOMAIN_NAME：(選用) 子網路的 DNS 網域名稱，可供 Connect 叢集查看。Connect 叢集可以使用網域名稱存取資源，而不必依賴 IP 位址。請參閱「DNS 對接」。
- LABELS：(選用) 要與叢集建立關聯的標籤。如要進一步瞭解標籤格式，請參閱「標籤」。要新增的標籤「鍵=值」組合清單。鍵的開頭須為小寫字元，且只能使用連字號 (-)、底線 (_)、小寫字元和數字。值只能使用連字號 (-)、底線 (_)、小寫半形字元和數字。
- CONFIG_FILE：(選用) JSON 或 YAML 檔案的路徑，其中包含從叢集或連接器預設值覆寫的設定。這個檔案也支援內嵌 JSON 或 YAML。
- --async：(選用) 立即返回，不等待執行中的作業完成。有了 --async 旗標，您可以在背景建立叢集時繼續執行其他工作。如果不使用旗標，系統會等待作業完成，再傳回回應。您必須等到叢集完全更新後，才能繼續執行其他工作。
您會收到類似以下的回覆：
```
Create request issued for: [sample-connectcluster]
Check operation [projects/test-project/locations/us-east1/operations/operation-1753590328249-63ae19098cc06-64300a0a-06512d02] for status.
```
儲存 OPERATION_ID 以追蹤進度。舉例來說，這裡的值是 operation-1753590328249-63ae19098cc06-64300a0a-06512d02。

Terraform

您可以使用 Terraform 資源建立Connect 叢集。

resource "google_managed_kafka_connect_cluster" "default" {
  provider           = google-beta
  project            = data.google_project.default.project_id
  connect_cluster_id = "my-connect-cluster-id"
  location           = "us-central1"
  kafka_cluster      = google_managed_kafka_cluster.default.id
  capacity_config {
    vcpu_count   = 12
    memory_bytes = 12884901888 # 12 GiB
  }
  gcp_config {
    access_config {
      network_configs {
        primary_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 createConnectCluster(w io.Writer, projectID, region, clusterID, kafkaCluster string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-connect-cluster"
	// kafkaCluster := "projects/my-project-id/locations/us-central1/clusters/my-kafka-cluster"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/connectClusters/%s", locationPath, clusterID)

	// Capacity configuration with 12 vCPU and 12 GiB RAM
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   12,
		MemoryBytes: 12884901888, // 12 GiB in bytes
	}

	// Optionally, you can also specify accessible subnets and resolvable DNS
	// domains as part of your network configuration. For example:
	// networkConfigs := []*managedkafkapb.ConnectNetworkConfig{
	// 	{
	// 		PrimarySubnet:      primarySubnet,
	// 		AdditionalSubnets:  []string{"subnet-1", "subnet-2"},
	// 		DnsDomainNames:     []string{"domain-1", "domain-2"},
	// 	},
	// }

	connectCluster := &managedkafkapb.ConnectCluster{
		Name:           clusterPath,
		KafkaCluster:   kafkaCluster,
		CapacityConfig: capacityConfig,
	}

	req := &managedkafkapb.CreateConnectClusterRequest{
		Parent:           locationPath,
		ConnectClusterId: clusterID,
		ConnectCluster:   connectCluster,
	}
	op, err := client.CreateConnectCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnectCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 5-15 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created connect 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.CapacityConfig;
import com.google.cloud.managedkafka.v1.ConnectAccessConfig;
import com.google.cloud.managedkafka.v1.ConnectCluster;
import com.google.cloud.managedkafka.v1.ConnectGcpConfig;
import com.google.cloud.managedkafka.v1.ConnectNetworkConfig;
import com.google.cloud.managedkafka.v1.CreateConnectClusterRequest;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateConnectCluster {

  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-connect-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    String kafkaCluster = "my-kafka-cluster"; // The Kafka cluster to connect to
    int cpu = 12;
    long memoryBytes = 12884901888L; // 12 GiB
    createConnectCluster(projectId, region, clusterId, subnet, kafkaCluster, cpu, memoryBytes);
  }

  public static void createConnectCluster(
      String projectId,
      String region,
      String clusterId,
      String subnet,
      String kafkaCluster,
      int cpu,
      long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setVcpuCount(cpu)
        .setMemoryBytes(memoryBytes).build();
    ConnectNetworkConfig networkConfig = ConnectNetworkConfig.newBuilder()
        .setPrimarySubnet(subnet)
        .build();
    // Optionally, you can also specify additional accessible subnets and resolvable
    // DNS domains as part of your network configuration. For example:
    // .addAllAdditionalSubnets(List.of("subnet-1", "subnet-2"))
    // .addAllDnsDomainNames(List.of("dns-1", "dns-2"))
    ConnectGcpConfig gcpConfig = ConnectGcpConfig.newBuilder()
        .setAccessConfig(ConnectAccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
        .build();
    ConnectCluster connectCluster = ConnectCluster.newBuilder()
        .setCapacityConfig(capacityConfig)
        .setGcpConfig(gcpConfig)
        .setKafkaCluster(kafkaCluster)
        .build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.createConnectClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient
        .create(settingsBuilder.build())) {
      CreateConnectClusterRequest request = CreateConnectClusterRequest.newBuilder()
          .setParent(LocationName.of(projectId, region).toString())
          .setConnectClusterId(clusterId)
          .setConnectCluster(connectCluster)
          .build();

      // The duration of this operation can vary considerably, typically taking
      // between 10-30 minutes.
      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
          .createConnectClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf(
          "Connect 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)
      ConnectCluster response = future.get();
      System.out.printf("Created connect cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaConnectClient.createConnectCluster got err: %s\n", 
          e.getMessage());
      throw e;
    }
  }
}

Python

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