Apache Kafka용 Google Cloud 관리형 서비스 클러스터 업데이트

Apache Kafka용 Google Cloud 관리형 서비스 클러스터를 수정하여 vCPU 수 및 메모리를 포함한 클러스터 크기, 연결된 서브넷 목록, 공개 클러스터에 허용되는 소스 IP 범위, 자동 재분산 구성, mTLS 구성과 같은 속성을 업데이트할 수 있습니다.

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

vCPU 수 및 메모리와 같은 특정 속성을 업데이트하려면 서비스에서 클러스터를 다시 시작해야 할 수 있습니다. 서비스는 한 번에 하나의 브로커에서 클러스터를 다시 시작합니다. 이 프로세스 중에 개별 브로커에 대한 요청이 실패할 수 있지만 이러한 실패는 일시적입니다. 흔히 사용되는 클라이언트 라이브러리는 이러한 오류를 자동으로 처리합니다.

필수 역할 및 권한

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

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

필수 권한

클러스터를 업데이트하려면 다음 권한이 필요합니다.

  • 클러스터 수정: managedkafka.clusters.update

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

클러스터 크기 조절

클러스터의 vCPU 수 또는 메모리를 업데이트하는 경우 다음 규칙이 적용됩니다.

  • 클러스터의 전체 vCPU-메모리 비율은 항상 1:1에서 1:8 사이를 유지해야 합니다.

  • 기존 브로커마다 vCPU 1개와 메모리 1GiB 이상이 있어야 합니다. 브로커 수는 줄어들지 않습니다.

  • 클러스터에 커스텀 디스크 구성이 있는 경우 업데이트는 로컬 스토리지의 디스크 구성 요구사항 을 충족해야 합니다.

  • 업스케일링하는 경우 브로커당 평균 vCPU 및 메모리는 업데이트 전 평균에 비해 10% 이상 감소할 수 없습니다. 예를 들어 클러스터를 vCPU 45개 (브로커 3개)에서 vCPU 48개 (브로커 4개)로 업스케일링하려고 하면 브로커당 평균 vCPU가 15개에서 12개로 감소하여 20% 감소하므로 10% 한도를 초과합니다.

    vCPU 수를 10% 이상 줄여야 하는 경우 여러 단계로 줄이는 것이 좋습니다. 각 업데이트 후 리소스 사용률을 모니터링하고 필요한 경우 파티션을 재분산합니다.

    하지만 업데이트 후 브로커에 충분한 용량이 있다고 확신하는 경우 gcloud managed-kafka clusters update 명령어를 allow_broker_downscale_on_cluster_upscale=true 플래그와 함께 실행하여 이 체크를 사용 중지할 수 있습니다. 이 플래그는 잠재적인 성능 위험을 수락한다는 신호를 보냅니다.

자세한 내용은 클러스터 크기 업데이트를 참고하세요.

공개 클러스터 구성

기존 클러스터의 공개 액세스를 사용 설정 또는 사용 중지하고 허용된 소스 IP 범위를 추가하거나 삭제할 수 있습니다. 허용된 소스 IP 범위의 요구사항 및 규칙에 대한 자세한 내용은 공개 클러스터를 참고하세요.

Managed Service for Apache Kafka는 Cloud Next Generation Firewall을 사용하여 공개 클러스터에 대한 액세스를 제한합니다. 허용된 소스 IP 범위를 삭제하거나 공개 액세스를 사용 중지하는 것은 새 연결에만 적용됩니다. 자세한 내용은 기존 트래픽에 미치는 영향을 참고하세요.

클러스터 수정

클러스터를 수정하려면 다음 단계를 따르세요.

Console

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

클러스터로 이동

  1. 클러스터 목록에서 속성을 수정할 클러스터를 클릭합니다.

콘솔에 클러스터 세부정보 페이지가 표시됩니다.

  1. 클러스터 세부정보 페이지에서 수정 을 클릭합니다.

  2. 필요에 따라 속성을 수정합니다. 콘솔에서 클러스터의 다음 속성을 수정할 수 있습니다.

    • 메모리
    • vCPUs
    • 서브넷
    • 재분산 구성
    • mTLS 구성
    • 라벨

  3. 저장 을 클릭합니다.

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: 클러스터의 메모리 양입니다. 예: 10GiB
    • SUBNET_ID: 연결할 서브넷의 서브넷 ID입니다. 예: default
    • LABELS: 클러스터와 연결할 라벨입니다.
    • ALLOWED_SOURCE_IP_RANGES: 공개 클러스터 인터넷 액세스를 위해 허용된 소스 IPv4 CIDR 범위입니다.

    다음 명령어를 실행합니다.

    Linux, macOS 또는 Cloud Shell

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

    Windows(PowerShell)

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

    Windows(cmd.exe)

    gcloud managed-kafka clusters update CLUSTER_ID ^
    --location=LOCATION ^
    --cpu=CPU_COUNT ^
    --memory=MEMORY ^
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
    --auto-rebalance ^
    --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: update
name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
    • 공개 액세스를 사용 중지하려면 --no-public-cluster 플래그를 사용합니다.
    • 명령어와 함께 --async 플래그를 사용하면 시스템에서 업데이트 요청을 전송하고 작업이 완료될 때까지 기다리지 않고 즉시 응답을 반환합니다. --async 플래그를 사용하면 클러스터 업데이트가 백그라운드에서 진행되는 동안 다른 작업을 계속할 수 있습니다. --async 플래그를 사용하지 않으면 시스템에서 작업이 완료될 때까지 기다린 후 응답을 반환합니다. 다른 작업을 계속하려면 클러스터가 완전히 업데이트될 때까지 기다려야 합니다.

REST

요청 데이터를 사용하기 전에 다음을 바꿉니다.

  • PROJECT_ID: Google Cloud 프로젝트 ID
  • LOCATION: 클러스터의 위치입니다.
  • CLUSTER_ID: 클러스터의 ID입니다.
  • UPDATE_MASK: 업데이트할 필드(정규화된 이름의 쉼표로 구분된 목록)입니다. 예: capacityConfig.vcpuCount,capacityConfig.memoryBytes
  • CPU_COUNT: 클러스터의 vCPU 수입니다.
  • MEMORY: 클러스터의 메모리 양(바이트)입니다. 예: 3221225472
  • SUBNET_ID: 연결할 서브넷의 서브넷 ID입니다. 예: default

HTTP 메서드 및 URL: 

PATCH https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK

요청 JSON 본문: 

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

요청을 보내려면 다음 옵션 중 하나를 펼칩니다.

다음과 비슷한 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": "update",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

요청 본문에는 UPDATE_MASK 쿼리 매개변수에 지정된 대로 업데이트하는 필드만 포함합니다.

  • 서브넷을 추가하려면 다음 형식으로 networkConfigs에 새 항목을 추가합니다. projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID. 예: projects/sample-project/regions/us-central1/subnetworks/default

  • 공개 액세스를 사용 설정하거나 허용된 소스 IP 범위를 업데이트하려면 gcpConfig.accessConfig.publicClusterConfig 쿼리 매개변수에 UPDATE_MASK를 포함하고 요청 본문에 allowedSourceIpRanges 배열을 지정합니다. 요청 본문 예시:

    {
  "gcpConfig": {
    "accessConfig": {
      "publicClusterConfig": {
        "allowedSourceIpRanges": [
          "203.0.113.0/24"
        ]
      }
    }
  }
}

  • 공개 액세스를 사용 중지하려면 gcpConfig.accessConfig.publicClusterConfig를 UPDATE_MASK 쿼리 매개변수에 포함하고 요청 본문에 빈 JSON 객체 {}를 전달하거나 publicClusterConfig를 생략합니다. 요청 본문 예시:

    {}

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"
	"google.golang.org/protobuf/types/known/fieldmaskpb"

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

func updateCluster(w io.Writer, projectID, region, clusterID string, memory int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// memoryBytes := 4221225472
	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)
	capacityConfig := &managedkafkapb.CapacityConfig{
		MemoryBytes: memory,
	}
	cluster := &managedkafkapb.Cluster{
		Name:           clusterPath,
		CapacityConfig: capacityConfig,
	}
	paths := []string{"capacity_config.memory_bytes"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateClusterRequest{
		UpdateMask: updateMask,
		Cluster:    cluster,
	}
	op, err := client.UpdateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateCluster got err: %w", err)
	}
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Updated cluster: %#v\n", resp)
	return nil
}

자바

이 샘플을 사용해 보기 전에 클라이언트 라이브러리 설치의 자바 설정 안내를 따르세요. 자세한 내용은 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.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.ClusterName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.UpdateClusterRequest;
import com.google.protobuf.FieldMask;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class UpdateCluster {

  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";
    long memoryBytes = 25769803776L; // 24 GiB
    updateCluster(projectId, region, clusterId, memoryBytes);
  }

  public static void updateCluster(
      String projectId, String region, String clusterId, long memoryBytes) throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setMemoryBytes(memoryBytes).build();
    Cluster cluster =
        Cluster.newBuilder()
            .setName(ClusterName.of(projectId, region, clusterId).toString())
            .setCapacityConfig(capacityConfig)
            .build();
    FieldMask updateMask = FieldMask.newBuilder().addPaths("capacity_config.memory_bytes").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.updateClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {
      UpdateClusterRequest request =
          UpdateClusterRequest.newBuilder().setUpdateMask(updateMask).setCluster(cluster).build();
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.updateClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details. CreateCluster contains sample code for polling logs.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf("Cluster update started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(),
          operation.isDone(),
          future.getMetadata().get().toString());

      Cluster response = future.get();
      System.out.printf("Updated cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.updateCluster 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
from google.protobuf import field_mask_pb2

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# memory_bytes = 4295000000

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.memory_bytes = memory_bytes
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.append("capacity_config.memory_bytes")

# For a list of editable fields, one can check https://cloud.google.com/managed-kafka/docs/create-cluster#properties.
request = managedkafka_v1.UpdateClusterRequest(
    update_mask=update_mask,
    cluster=cluster,
)

try:
    operation = client.update_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Updated cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

제한사항

Managed Service for Apache Kafka 클러스터를 만든 후에는 다음 속성을 업데이트할 수 없습니다.

  • 클러스터 이름
  • 클러스터 위치
  • 암호화 유형

암호화 유형은 변경할 수 없지만 암호화 키를 순환 할 수 있습니다.

다음 단계

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