Apache Kafka용 Google Cloud 관리형 서비스 클러스터 만들기

Managed Service for Apache Kafka 클러스터는 주제로 구성된 메시지 스트림을 저장하고 처리하는 환경을 제공합니다.

클러스터를 만들려면 Google Cloud 콘솔, Google Cloud CLI, 클라이언트 라이브러리 또는 Managed Kafka API를 사용하면 됩니다. 오픈소스 Apache Kafka API를 사용하여 클러스터를 만들 수는 없습니다.

시작하기 전에

다음 사항을 숙지하고 있는지 확인하세요.

• Apache Kafka용 관리형 서비스의 작동 방식

• Apache Kafka용 관리형 서비스와 관련된 네트워킹 개념

클러스터 생성에 필요한 역할 및 권한

클러스터를 만드는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 관리형 Kafka 클러스터 편집자 (roles/managedkafka.clusterEditor) IAM 역할을 부여해 달라고 요청하세요. 역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

이 사전 정의된 역할에는 클러스터를 만드는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.

필수 권한

클러스터를 만들려면 다음 권한이 필요합니다.

• 클러스터 만들기: managedkafka.clusters.create

커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한을 부여받을 수도 있습니다.

관리형 Kafka 클러스터 편집자 역할로는 Apache Kafka용 관리형 서비스 클러스터에서 주제와 소비자 그룹을 만들거나 삭제하거나 수정할 수 없습니다. 또한 클러스터 내에서 메시지를 게시하거나 소비하기 위한 데이터 플레인 액세스도 허용하지 않습니다. 이 역할에 대한 자세한 내용은 Apache Kafka용 관리형 서비스 사전 정의된 역할을 참고하세요.

Apache Kafka용 관리형 서비스 클러스터의 속성

Apache Kafka용 관리형 서비스 클러스터를 만들거나 업데이트할 때는 다음 속성을 지정해야 합니다.

클러스터 이름

생성 중인 Apache Kafka용 관리형 서비스 클러스터의 이름 또는 ID입니다. 클러스터 이름을 지정하는 방법에 대한 가이드라인은 Apache Kafka용 관리형 서비스 리소스 이름 지정 가이드라인을 참고하세요. 클러스터 이름은 변경할 수 없습니다.

위치

클러스터를 생성 중인 위치입니다. 위치는 지원되는 Google Cloud 리전 중 하나여야 합니다. 클러스터의 위치는 나중에 변경할 수 없습니다. 사용 가능한 위치 목록은 Apache Kafka용 관리형 서비스 위치를 참고하세요.

용량 구성

용량을 구성하려면 Kafka 설정의 vCPU 수 및 메모리 양을 구성해야 합니다. 클러스터의 용량을 구성하는 방법을 자세히 알아보려면 Kafka 클러스터 크기 계획을 참고하세요.

다음은 용량 구성의 속성입니다.

• vCPUs: 클러스터의 vCPU 수입니다. 클러스터당 vCPU가 3개 이상 필요합니다.

• 메모리: 클러스터에 할당된 메모리 양입니다. vCPU당 1GiB에서 8GiB 사이로 프로비저닝해야 합니다.

  예를 들어 vCPU가 6개인 클러스터를 만드는 경우 클러스터에 할당할 수 있는 최소 메모리는 6GiB (vCPU당 1GiB)이고 최대 메모리는 48GiB (vCPU당 8GiB)입니다.

클러스터를 만든 후 메모리와 vCPU 수를 변경하는 방법에 대한 자세한 내용은 클러스터 크기 업데이트를 참고하세요.

네트워크 구성

네트워크 구성은 클러스터에 액세스할 수 있는 VPC 서브넷 목록입니다. 메시지를 생성하거나 소비하려면 클라이언트가 이러한 서브넷 중 하나에 도달할 수 있어야 합니다.

다음은 네트워크 구성에 대한 몇 가지 가이드라인입니다.

• 클러스터에는 하나 이상의 서브넷이 필요합니다. 최댓값은 10입니다.

• 주어진 클러스터에는 네트워크당 정확히 하나의 서브넷만 허용됩니다.

• 각 서브넷은 클러스터와 동일한 리전에 있어야 합니다. 프로젝트와 네트워크는 다를 수 있습니다.

• 브로커 및 부트스트랩 서버의 IP 주소가 각 서브넷에 자동으로 할당됩니다. 또한 이러한 IP 주소의 DNS 항목이 해당하는 VPC 네트워크에 생성됩니다.

• 다른 프로젝트의 서브넷을 추가하는 경우 클러스터와 연결된 Google 관리 서비스 계정에 권한을 부여해야 합니다. 자세한 내용은 프로젝트 간 클러스터 연결을 참고하세요.

클러스터를 만든 후 서브넷 목록을 업데이트할 수 있습니다. 네트워킹에 대한 자세한 내용은 Apache Kafka용 관리형 서비스의 네트워킹 구성을 참고하세요.

라벨

라벨은 구성 및 식별에 도움이 되는 키-값 쌍입니다. 라벨을 사용하면 환경에 따라 리소스를 분류할 수 있습니다. 예로는 "env:production" 및 "owner:data-engineering"이 있습니다.

라벨을 기준으로 리소스를 필터링하고 검색할 수 있습니다. 예를 들어 부서별로 여러 개의 Apache Kafka용 관리형 서비스 클러스터가 있다고 가정해 보겠습니다. 라벨 "department:marketing"로 클러스터를 구성하고 검색하여 관련 클러스터를 빠르게 찾을 수 있습니다.

재분산 구성

이 설정은 서비스가 브로커 간에 파티션 복제본을 자동으로 리밸런싱하는지 여부를 결정합니다.

사용 가능한 모드는 다음과 같습니다.

• 확장 시 자동 리밸런싱: 이 옵션을 사용 설정하면 클러스터를 확장할 때 서비스에서 자동으로 복제본의 리밸런싱을 트리거합니다. 이 모드는 균등한 부하 분산을 유지하는 데 도움이 되지만 리밸런싱 작업 중에 일시적으로 성능에 영향을 줄 수 있습니다.

• 리밸런스 없음: 이 옵션을 사용 설정하면 서비스에서 복제본을 자동으로 리밸런스하지 않습니다.

암호화

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 풀: 클라이언트 인증을 위해 클러스터에서 신뢰하는 인증 기관 서비스(CAS) 풀 목록입니다(1~10개).

• SSL 주 구성원 매핑 규칙: Kafka ACL에서 사용할 긴 인증서 주 구성원 이름을 간소화하는 선택사항이지만 권장되는 ssl.principal.mapping.rules 브로커 속성입니다.

mTLS에 대한 자세한 내용은 mTLS 인증 구성을 참고하세요.

클러스터 만들기

클러스터를 만들기 전에 클러스터 속성 문서를 검토하세요.

클러스터를 만드는 데는 일반적으로 20~30분 정도 걸립니다.

클러스터를 만들려면 다음 단계를 따르세요.

콘솔

1. Google Cloud 콘솔에서 클러스터 페이지로 이동합니다.

   클러스터로 이동

2. 만들기를 선택합니다.

   Kafka 클러스터 만들기 페이지가 열립니다.

3. 클러스터 이름에 문자열을 입력합니다.

   클러스터 이름을 지정하는 방법에 대한 자세한 내용은 Apache Kafka용 관리형 서비스 리소스 이름 지정 가이드라인을 참고하세요.

4. 위치에 지원되는 위치를 입력합니다.

   지원되는 위치에 대한 자세한 내용은 지원되는 Apache Kafka용 관리형 서비스 위치를 참고하세요.

5. 용량 구성의 경우 메모리 및 vCPU 값을 입력합니다.

   Apache Kafka용 관리형 서비스 클러스터의 크기를 조정하는 방법에 대한 자세한 내용은 Kafka 클러스터 크기 계획을 참고하세요.

6. 네트워크 구성에 다음 세부정보를 입력합니다.

   1. 프로젝트: 서브네트워크가 있는 프로젝트입니다. 서브넷은 클러스터와 동일한 리전에 있어야 하지만 프로젝트는 다를 수 있습니다.
   2. 네트워크: 서브넷이 연결된 네트워크입니다.
   3. 서브네트워크: 서브넷의 이름입니다.
   4. 서브넷 URI 경로: 이 필드는 자동으로 채워집니다. 또는 여기에 서브넷 경로를 입력할 수 있습니다. 서브넷 이름은 projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID 형식이어야 합니다.
   5. 완료를 클릭합니다.

7. (선택사항) 연결된 서브넷 추가를 클릭하여 서브넷을 추가합니다.

   최대 10개까지 서브넷을 추가할 수 있습니다.

8. 다른 기본값은 유지합니다.

9. 만들기를 클릭합니다.

gcloud

1. 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.

2. gcloud managed-kafka clusters create 명령어를 실행합니다.

   gcloud managed-kafka clusters create CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU \
       --memory=MEMORY \
       --subnets=SUBNETS \
       --auto-rebalance \
       --encryption-key=ENCRYPTION_KEY \
       --async \
       --labels=LABELS
   

   다음을 바꿉니다.

   • CLUSTER_ID: 클러스터의 ID 또는 이름입니다.

     클러스터 이름을 지정하는 방법에 대한 자세한 내용은 Apache Kafka용 관리형 서비스 리소스 이름 지정 가이드라인을 참고하세요.

   • LOCATION: 클러스터의 위치입니다.

     지원되는 위치에 대한 자세한 내용은 Apache Kafka용 관리형 서비스 위치를 참고하세요.

   • CPU: 클러스터의 vCPU 수입니다.

     Apache Kafka용 관리형 서비스 클러스터의 크기를 조정하는 방법에 대한 자세한 내용은 Kafka 클러스터 크기 계획을 참고하세요.

   • MEMORY: 클러스터의 메모리 양입니다. 'MB', 'MiB', 'GB', 'GiB', 'TB', 'TiB' 단위를 사용하세요. 예: '10GiB'

   • SUBNETS: 연결할 서브넷 목록입니다. 쉼표를 사용하여 여러 서브넷 값을 구분합니다.

     서브넷의 형식은 projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID입니다.

   • auto-rebalance: 클러스터의 CPU 수가 변경될 때 브로커 간에 주제 파티션의 자동 재분산을 사용 설정합니다. 이 기능은 기본적으로 사용 설정되어 있습니다.

   • ENCRYPTION_KEY: 클러스터에 사용할 고객 관리 암호화 키의 ID입니다.

     형식은 projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/CRYPTO_KEY입니다.

   • --async: 시스템이 생성 요청을 전송하고 작업이 완료될 때까지 기다리지 않고 즉시 응답을 반환하도록 합니다. --async 플래그를 사용하면 클러스터가 백그라운드에서 생성되는 동안 다른 작업을 계속할 수 있습니다. 플래그를 사용하지 않으면 시스템은 작업이 완료될 때까지 기다린 후 응답을 반환합니다. 클러스터가 완전히 업데이트될 때까지 기다려야 다른 작업을 계속할 수 있습니다.

   • LABELS: 클러스터와 연결할 라벨입니다.

     라벨 형식에 대한 자세한 내용은 라벨을 참고하세요.

   다음과 비슷한 응답이 표시됩니다.

   Create request issued for: [CLUSTER_ID]
   Check operation [projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID] for status.
   

   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 설정 안내를 따르세요. 자세한 내용은 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 설정 안내를 따르세요. 자세한 내용은 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 설정 안내를 따르세요. 자세한 내용은 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)을 사용하며, 다음 명령어를 사용하여 모니터링할 수 있습니다.

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

  다음을 바꿉니다.

  • OPERATION_ID을 이전 섹션의 작업 ID 값으로 바꿉니다.
  • LOCATION을 이전 섹션의 위치 값으로 바꿉니다.

문제 해결

다음은 클러스터를 만들 때 발생할 수 있는 오류입니다.

Service agent service-${PROJECT_NUMBER}@gcp-sa-managedkafka.iam.gserviceaccount.com has not been granted the required role cloudkms.cryptoKeyEncrypterDecrypter to encrypt data using the KMS key.

Apache Kafka용 관리형 서비스 서비스 에이전트에 Cloud KMS 키에 액세스하는 데 필요한 권한이 없습니다. CMEK 구성에 필요한 역할에 관한 문서를 참고하세요.

Service does not have permission to retrieve subnet. Please grant service-${PROJECT_NUMBER}@gcp-sa-managedkafka.iam.gserviceaccount.com the managedkafka.serviceAgent role in the IAM policy of the project ${SUBNET_PROJECT} and ensure the Compute Engine API is enabled in project ${SUBNET_PROJECT}

Apache Kafka용 관리형 서비스 서비스 에이전트에 Kafka 클라이언트가 실행되는 VPC 네트워크에서 네트워킹을 구성하는 데 필요한 역할이 없습니다. 자세한 내용은 프로젝트 간 클러스터 연결을 참고하세요.

다음 단계

• 클러스터 모니터링

• 클러스터 업데이트

• 모든 클러스터를 나열합니다.

• 클러스터 보기

• 클러스터 삭제

• 클러스터 내에 주제를 만듭니다.

• Apache Kafka용 관리형 서비스의 네트워킹 구성

Apache Kafka®는 미국 및/또는 다른 국가에서 사용되는 Apache Software Foundation 또는 해당 계열사의 등록 상표입니다.