Google Cloud Managed Service for Apache Kafka クラスタを作成する

Managed Service for Apache Kafka クラスタは、トピックに整理されたメッセージのストリームを保存して処理するための環境を提供します。

クラスタを作成するには、 Google Cloud コンソール、Google Cloud CLI、クライアント ライブラリ、または Managed Kafka API を使用します。オープンソースの Apache Kafka API を使用してクラスタを作成することはできません。

始める前に

次の内容を理解していることを確認します。

• Managed Service for Apache Kafka の仕組み。

• Managed Service for Apache Kafka に関連するネットワーキングのコンセプト。

クラスタの作成に必要なロールと権限

クラスタの作成に必要な権限を取得するには、プロジェクトに対する Managed Kafka クラスタ編集者 （roles/managedkafka.clusterEditor）IAM ロールを付与するよう管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

この事前定義ロールには、クラスタの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

クラスタを作成するには、次の権限が必要です。

• クラスタを作成します。 managedkafka.clusters.create

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

Managed Service for Apache Kafka クラスタのプロパティ

Managed Service for Apache Kafka クラスタを作成または更新するときは、次のプロパティを指定する必要があります。

クラスタ名

作成する Managed Service for Apache Kafka クラスタの名前または ID。クラスタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。クラスタの名前は不変です。

ロケーション

クラスタを作成するロケーション。ロケーションには、サポートされている Google Cloud リージョンのいずれかを指定する必要があります。クラスタのロケーションは後から変更できません。使用可能なロケーションのリストについては、Managed Service for Apache Kafka のロケーションをご覧ください。

容量の構成

Kafka クラスタのコンピューティング、メモリ、ストレージの容量。クラスタを作成するときは、次の値を指定する必要があります。

• vCPUs: クラスタ内の vCPU の数。クラスタごとに少なくとも 3 つの vCPU が必要です。

• メモリ: クラスタに割り当てられているメモリの量。vCPU あたり 1 GiB ～ 8 GiB の範囲でプロビジョニングする必要があります。

  たとえば、6 個の vCPU を使用してクラスタを作成する場合、クラスタに割り当てることができる最小メモリは 6 GiB（vCPU あたり 1 GiB）で、最大メモリは 48 GiB（vCPU あたり 8 GiB）です。

このサービスは、ブローカーの Persistent Disk ボリュームなど、ブローカーを自動的にプロビジョニングします。必要に応じて、カスタム ローカル ストレージを有効にできます。これにより、各ブローカーのディスク容量を指定できます。詳細については、ブローカーのディスクサイズを構成するをご覧ください。

クラスタに必要な容量を見積もるためのガイドラインについては、Kafka クラスタのサイズを計画するをご覧ください。

ネットワークの構成

ネットワーク構成は、クラスタがアクセスできる VPC サブネットのリストです。メッセージを生成または使用するには、クライアントがこれらのサブネットのいずれかにアクセスできる必要があります。

ネットワーク構成に関するガイドラインを次に示します。

• クラスタには少なくとも 1 つのサブネットが必要です。最大値は 10 です。

• 特定のクラスタでは、ネットワークごとに 1 つのサブネットのみが許可されます。

• 各サブネットはクラスタと同じリージョンに存在する必要があります。プロジェクトとネットワークは異なる場合があります。

• ブローカーとブートストラップ サーバーの IP アドレスは、各サブネットに自動的に割り当てられます。また、これらの IP アドレスの DNS エントリが、対応する VPC ネットワーク内に作成されます。

• 別のプロジェクトのサブネットを追加する場合は、クラスタに関連付けられている Google 管理のサービス アカウントに権限を付与する必要があります。詳細については、プロジェクト間でクラスタを接続するをご覧ください。

クラスタを作成した後、サブネットのリストを更新できます。ネットワーキングの詳細については、Managed Service for Apache Kafka のネットワーキングを構成するをご覧ください。

一般公開クラスタ

必要に応じて、クラスタをパブリック クラスタとして構成し、クライアント アプリケーションがクラスタに接続して、パブリック インターネット経由で Kafka オペレーション（メッセージの送受信など）を実行できるようにすることができます。一般公開アクセスを有効にするには、許可された送信元 IP 範囲を 1 つ以上指定する必要があります。

許可された送信元 IP 範囲を構成する場合は、次のルールが適用されます。

• 許可される送信元 IP 範囲はすべて、IPv4 CIDR 表記で指定する必要があります（例: 203.0.113.0/24 または 198.51.100.5/32）。

• 各 CIDR サブネットのサイズは /16～/32 の範囲にする必要があります。

• CIDR 範囲は重複しないようにしてください。

• 許可するすべてのソース IP 範囲は、一般公開されたルーティング可能な範囲である必要があります。10.0.0.0/8（RFC 1918）などのプライベート IP 範囲は拒否されます。

• 許可される送信元 IP 範囲は最大 500 個まで指定できます。

• IPv6 アドレスはサポートされていません。

注意: クラスタを保護するには、信頼できる外部 IP 範囲からのアクセスのみを許可します。クラスタを信頼できない範囲に公開しないでください。

クラスタの一般公開アクセスを有効にする前に、クライアントをパブリック クラスタに接続するを読んで、一般公開アクセス機能の実装方法を確認することをおすすめします。

ラベル

ラベルは、整理と識別に役立つ Key-Value ペアです。ラベルを使用すると、環境に基づいてリソースを分類できます。例: "env:production"、"owner:data-engineering"。

ラベルに基づいてリソースをフィルタして検索できます。たとえば、部門ごとに複数の Managed Service for Apache Kafka クラスタがあるとします。ラベル "department:marketing" を使用してクラスタを構成し、検索して関連するクラスタを見つけることができます。

再分散の構成

この設定は、サービスがブローカー間でパーティション レプリカを自動的に再調整するかどうかを決定します。

使用可能なモードは次のとおりです。

• スケールアップ時の自動再調整: このオプションを有効にすると、クラスタをスケールアップするときに、サービスがレプリカの再調整を自動的にトリガーします。このモードは、均等な負荷分散を維持するのに役立ちますが、再調整オペレーション中にパフォーマンスに一時的に影響する可能性があります。

• 再調整なし: このオプションを有効にすると、サービスはレプリカを自動的に再調整しません。

詳細については、ブローカー パーティションの再調整をご覧ください。

暗号化

Managed Service for Apache Kafka は、Google-owned and Google-managed encryption keys （デフォルト）または顧客管理の暗号鍵（CMEK）でメッセージを暗号化できます。すべてのメッセージは、保存時も転送時も暗号化されます。クラスタの暗号化タイプは変更できません。

デフォルトでは Google-owned and Google-managed encryption keys が使用されます。これらの鍵は、 Google Cloud のインフラストラクチャ内で作成、管理、保存されます。

CMEK は、Cloud Key Management Service を使用して管理する暗号鍵です。この機能を使用すると、サポートされている Google Cloud サービス内で保存データの暗号化に使用する鍵をより細かく制御できます。CMEK を使用すると、Cloud Key Management Service に関連する追加費用が発生します。CMEK を使用する場合、キーリングは、使用するリソースと同じロケーションに存在する必要があります。詳細については、メッセージ暗号化を構成するをご覧ください。

mTLS 構成

必要に応じて、クライアント証明書を使用する代替認証方法として mTLS を構成できます。構成の内容は次のとおりです。

• CA プール: クラスタがクライアント認証で信頼する 1 ～ 10 個の Certificate Authority Service（CAS）プールのリスト。

• SSL プリンシパル マッピング ルール: Kafka ACL で使用する長い証明書プリンシパル名を簡略化するための ssl.principal.mapping.rules ブローカー プロパティ（省略可ですが、推奨）。

mTLS の詳細については、mTLS 認証を構成するをご覧ください。

クラスタの作成

クラスタを作成する前に、クラスタ プロパティのドキュメントをお読みください。

通常、クラスタの作成には 20～30 分かかります。

クラスタを作成する手順は次のとおりです。

コンソール

1. Google Cloud コンソールで、[クラスタ] ページに移動します。

   [クラスタ] に移動

2. [作成] を選択します。

   [Kafka クラスタの作成] ページが開きます。

3. [クラスタ名] に文字列を入力します。

   クラスタの命名方法の詳細については、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。

4. [ロケーション] に、サポートされているロケーションを入力します。

   サポートされているロケーションの詳細については、サポートされている Managed Service for Apache Kafka のロケーションをご覧ください。

5. [容量構成] で、[メモリ] と [vCPU] の値を入力します。

   クラスタのサイズ設定方法の詳細については、Kafka クラスタのサイズを計画するをご覧ください。

6. 省略可。ブローカーのディスク容量を指定する手順は次のとおりです。

   1. [カスタム ローカル ストレージ サイズを有効にする] をクリックしてオンにします。

      重要: [保存] をクリックした後、このクラスタでこの設定を再度無効にすることはできません。

   2. [カスタム ローカル ストレージ] フィールドに、ブローカーあたりのディスクサイズを GiB 単位で入力します。

   [カスタム ローカル ストレージ サイズを有効にする] 切り替えをオフのままにすると、デフォルトのディスク容量がプロビジョニングされます。詳細については、ブローカーのディスクサイズを構成するをご覧ください。

7. [ネットワーク構成] に、次の詳細を入力します。

   1. プロジェクト: サブネットワークが配置されているプロジェクト。サブネットはクラスタと同じリージョンに存在する必要がありますが、プロジェクトは異なっていてもかまいません。
   2. ネットワーク: サブネットが接続されているネットワーク。
   3. サブネットワーク: サブネットの名前。
   4. サブネットの URI パス: このフィールドは自動的に入力されます。または、ここにサブネット パスを入力することもできます。サブネットの名前は、projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID の形式にする必要があります。
   5. [完了] をクリックします。

8. 省略可: サブネットを追加するには、[接続されたサブネットを追加] をクリックします。

   追加のサブネットは最大 10 個まで追加できます。

9. 省略可: クラスタへの公開アクセスを有効にします。

   1. [一般公開クラスタの構成] セクションの [クラスタにアクセス可能] で、[プライベート ネットワークとパブリック インターネットを使用] を選択します。

   2. [許可する送信元 IPv4 範囲の CIDR 表記のカンマ区切りリスト] フィールドに、クライアント アプリケーションの IP 範囲を IPv4 CIDR 表記（203.0.113.0/24 など）で入力します。IPv6 アドレスはサポートされていません。

10. 省略可: 顧客管理の暗号鍵（CMEK）を設定します。

    1. [暗号化] で、[Cloud KMS 鍵] を選択します。

    2. [鍵のタイプ] で [Cloud KMS] を選択します。

    3. [顧客管理の暗号鍵を選択] で、CMEK を選択または入力します。

       このオプションが表示されない場合は、プロジェクトで Cloud KMS API が有効になっていることを確認してください。

    詳細については、メッセージ暗号化を構成するをご覧ください。

11. [作成] をクリックします。

gcloud

1. Google Cloud コンソールで Cloud Shell をアクティブにします。

   Cloud Shell をアクティブにする

   Google Cloud コンソールの下部にある Cloud Shell セッションが開始し、コマンドライン プロンプトが表示されます。Cloud Shell はシェル環境です。Google Cloud CLI がすでにインストールされており、現在のプロジェクトの値もすでに設定されています。セッションが初期化されるまで数秒かかることがあります。

2. 後述のコマンドデータを使用する前に、次のように置き換えます。

   • PROJECT_ID: プロジェクト ID。
   • LOCATION: クラスタのロケーション。
   • CLUSTER_ID: クラスタの ID。
   • CPU_COUNT: クラスタの vCPU の数。
   • MEMORY: クラスタのメモリ容量。例: 3GiB
   • SUBNET_ID: 接続先のサブネットのサブネット ID。例: default
   • ENCRYPTION_KEY: クラスタに使用する顧客管理の暗号鍵 ID。これは省略可能です。
   • LABELS: クラスタに関連付けるラベル。
   • ALLOWED_SOURCE_IP_RANGES: パブリック クラスタで許可されている送信元 IP 範囲。これは省略可能です。

   次のコマンドを実行します。

   Linux、macOS、Cloud Shell

   注: 認証とプロジェクトを使用して Google Cloud CLI を初期化するには、gcloud init、gcloud auth login、または gcloud config set project のいずれかを実行します。

   gcloud managed-kafka clusters create CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU_COUNT \
       --memory=MEMORY \
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
       --auto-rebalance \
       --encryption-key=ENCRYPTION_KEY \
       --async \
       --labels=LABELS \
       --public-cluster \
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows（PowerShell）

   注: 認証とプロジェクトを使用して Google Cloud CLI を初期化するには、gcloud init、gcloud auth login、または gcloud config set project のいずれかを実行します。

   gcloud managed-kafka clusters create CLUSTER_ID `
       --location=LOCATION `
       --cpu=CPU_COUNT `
       --memory=MEMORY `
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
       --auto-rebalance `
       --encryption-key=ENCRYPTION_KEY `
       --async `
       --labels=LABELS `
       --public-cluster `
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows（cmd.exe）

   注: 認証とプロジェクトを使用して Google Cloud CLI を初期化するには、gcloud init、gcloud auth login、または gcloud config set project のいずれかを実行します。

   gcloud managed-kafka clusters create CLUSTER_ID ^
       --location=LOCATION ^
       --cpu=CPU_COUNT ^
       --memory=MEMORY ^
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
       --auto-rebalance ^
       --encryption-key=ENCRYPTION_KEY ^
       --async ^
       --labels=LABELS ^
       --public-cluster ^
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   次のようなレスポンスが返されます。

   done: false
   metadata:
     '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
     apiVersion: v1
     createTime: 'CREATE_TIME'
     requestedCancellation: false
     target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
     verb: create
   name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
   

   progress をトラッキングするために OPERATION_ID を保存します。

REST

リクエストのデータを使用する前に、次のように置き換えます。

• PROJECT_ID: 実際の Google Cloud プロジェクト ID
• LOCATION: クラスタのロケーション。
• CLUSTER_ID: クラスタの ID。
• CPU_COUNT: クラスタの vCPU の数。
• MEMORY: クラスタのメモリ量（バイト単位）。例: 3221225472
• SUBNET_ID: 接続先のサブネットのサブネット ID。例: default

HTTP メソッドと URL:

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

リクエストの本文（JSON）:

{
  "capacityConfig": {
    "vcpuCount": CPU_COUNT,
    "memoryBytes": MEMORY
  },
  "gcpConfig": {
    "accessConfig": {
      "networkConfigs": [
        {
          "subnet": "projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID"
        }
      ]
    }
  }
}

リクエストを送信するには、次のいずれかのオプションを展開します。

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?clusterId=CLUSTER_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?clusterId=CLUSTER_ID" | Select-Object -Expand Content

次のような JSON レスポンスが返されます。

{
  "name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata",
    "createTime": "CREATE_TIME",
    "target": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
    "verb": "create",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Terraform

Terraform リソースを使用してクラスタを作成できます。

resource "google_managed_kafka_cluster" "default" {
  project    = data.google_project.default.project_id # Replace this with your project ID in quotes
  cluster_id = "my-cluster-id"
  location   = "us-central1"
  capacity_config {
    vcpu_count   = 3
    memory_bytes = 3221225472
  }
  gcp_config {
    access_config {
      network_configs {
        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 createCluster(w io.Writer, projectID, region, clusterID, subnet string, cpu, memoryBytes int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// subnet := "projects/my-project-id/regions/us-central1/subnetworks/default"
	// cpu := 3
	// memoryBytes := 3221225472
	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)
	clusterPath := fmt.Sprintf("%s/clusters/%s", locationPath, clusterID)

	// Memory must be between 1 GiB and 8 GiB per CPU.
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   cpu,
		MemoryBytes: memoryBytes,
	}
	var networkConfig []*managedkafkapb.NetworkConfig
	networkConfig = append(networkConfig, &managedkafkapb.NetworkConfig{
		Subnet: subnet,
	})
	platformConfig := &managedkafkapb.Cluster_GcpConfig{
		GcpConfig: &managedkafkapb.GcpConfig{
			AccessConfig: &managedkafkapb.AccessConfig{
				NetworkConfigs: networkConfig,
			},
		},
	}
	rebalanceConfig := &managedkafkapb.RebalanceConfig{
		Mode: managedkafkapb.RebalanceConfig_AUTO_REBALANCE_ON_SCALE_UP,
	}
	cluster := &managedkafkapb.Cluster{
		Name:            clusterPath,
		CapacityConfig:  capacityConfig,
		PlatformConfig:  platformConfig,
		RebalanceConfig: rebalanceConfig,
	}

	req := &managedkafkapb.CreateClusterRequest{
		Parent:    locationPath,
		ClusterId: clusterID,
		Cluster:   cluster,
	}
	op, err := client.CreateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 10-40 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created 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.AccessConfig;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.CreateClusterRequest;
import com.google.cloud.managedkafka.v1.GcpConfig;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.NetworkConfig;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.RebalanceConfig;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateCluster {

  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 subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    int cpu = 3;
    long memoryBytes = 3221225472L; // 3 GiB
    createCluster(projectId, region, clusterId, subnet, cpu, memoryBytes);
  }

  public static void createCluster(
      String projectId, String region, String clusterId, String subnet, int cpu, long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig =
        CapacityConfig.newBuilder().setVcpuCount(cpu).setMemoryBytes(memoryBytes).build();
    NetworkConfig networkConfig = NetworkConfig.newBuilder().setSubnet(subnet).build();
    GcpConfig gcpConfig =
        GcpConfig.newBuilder()
            .setAccessConfig(AccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
            .build();
    RebalanceConfig rebalanceConfig =
        RebalanceConfig.newBuilder()
            .setMode(RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP)
            .build();
    Cluster cluster =
        Cluster.newBuilder()
            .setCapacityConfig(capacityConfig)
            .setGcpConfig(gcpConfig)
            .setRebalanceConfig(rebalanceConfig)
            .build();

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

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {

      CreateClusterRequest request =
          CreateClusterRequest.newBuilder()
              .setParent(LocationName.of(projectId, region).toString())
              .setClusterId(clusterId)
              .setCluster(cluster)
              .build();

      // The duration of this operation can vary considerably, typically taking between 10-40
      // minutes.
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.createClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf("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)
      Cluster response = future.get();
      System.out.printf("Created cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.createCluster 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 GoogleAPICallError
from google.cloud import managedkafka_v1

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 3
# memory_bytes = 3221225472

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.vcpu_count = cpu
cluster.capacity_config.memory_bytes = memory_bytes
cluster.gcp_config.access_config.network_configs = [
    managedkafka_v1.NetworkConfig(subnet=subnet)
]
cluster.rebalance_config.mode = (
    managedkafka_v1.RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP
)

request = managedkafka_v1.CreateClusterRequest(
    parent=client.common_location_path(project_id, region),
    cluster_id=cluster_id,
    cluster=cluster,
)

try:
    operation = client.create_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # The duration of this operation can vary considerably, typically taking 10-40 minutes.
    # We can set a timeout of 3000s (50 minutes).
    response = operation.result(timeout=3000)
    print("Created cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

クラスタ作成オペレーションをモニタリングする

次のコマンドは、gcloud CLI を使用してクラスタを作成した場合にのみ実行できます。

• 通常、クラスタの作成には 20 ～ 30 分かかります。クラスタ作成の進行状況を追跡するために、gcloud managed-kafka clusters create コマンドは長時間実行オペレーション（LRO）を使用します。次のコマンドを使用して、LRO をモニタリングできます。

  gcloud managed-kafka operations describe OPERATION_ID \
      --location=LOCATION
  

  次のように置き換えます。

  • OPERATION_ID: 前のセクションのオペレーション ID の値。
  • LOCATION は、前のセクションのロケーションの値に置き換えます。

トラブルシューティング

クラスタの作成に関連するエラーのトラブルシューティング方法については、Kafka クラスタの作成エラーをご覧ください。

次のステップ

• クラスタをモニタリングする。

• クラスタを更新する。

• すべてのクラスタを一覧表示します。

• クラスタを表示する。

• クラスタを削除する。

• クラスタ内にトピックを作成します。

• Managed Service for Apache Kafka のネットワーキングを構成する。

Apache Kafka® は、Apache Software Foundation または米国その他の諸国における関連会社の商標です。