커넥터 업데이트

커넥터를 수정하여 읽거나 쓰는 주제를 변경하거나, 데이터 변환을 수정하거나, 오류 처리 설정을 조정하는 등 구성을 업데이트할 수 있습니다.

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

시작하기 전에

커넥터를 업데이트하기 전에 기존 구성을 검토하고 변경사항이 미칠 수 있는 영향을 파악하세요.

커넥터를 업데이트하는 데 필요한 역할 및 권한

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

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

필수 권한

커넥터를 수정하려면 다음 권한이 필요합니다.

• 상위 Connect 클러스터에 커넥터 업데이트 권한을 부여합니다. managedkafka.connectors.update
• 상위 연결 클러스터에 목록 커넥터 권한을 부여합니다. This permission is only required for updating a connector using the Google Cloud console

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

관리형 Kafka 커넥터 편집자 역할에 대한 자세한 내용은 Apache Kafka용 Google Cloud 관리형 서비스 사전 정의된 역할을 참고하세요.

커넥터의 수정 가능한 속성

커넥터의 수정 가능한 속성은 유형에 따라 다릅니다. 지원되는 커넥터 유형의 수정 가능한 속성은 다음과 같습니다.

MirrorMaker 2.0 소스 커넥터

• 쉼표로 구분된 주제 이름 또는 주제 정규식: 복제할 주제입니다.

  속성에 대한 자세한 내용은 주제 이름을 참고하세요.

• 구성: 커넥터의 추가 구성 설정입니다.

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

• 작업 재시작 정책: 실패한 커넥터 작업을 다시 시작하는 정책입니다.

  속성에 관한 자세한 내용은 작업 다시 시작 정책을 참고하세요.

BigQuery 싱크 커넥터

• 주제: 데이터를 스트리밍할 Kafka 주제입니다.

  속성에 관한 자세한 내용은 주제를 참고하세요.

• 데이터 세트: 데이터를 저장할 BigQuery 데이터 세트입니다.

  속성에 대한 자세한 내용은 데이터 세트를 참고하세요.

• 구성: 커넥터의 추가 구성 설정입니다.

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

• 작업 재시작 정책: 실패한 커넥터 작업을 다시 시작하는 정책입니다.

  속성에 관한 자세한 내용은 작업 다시 시작 정책을 참고하세요.

Cloud Storage 싱크 커넥터

• 주제: 데이터를 스트리밍할 Kafka 주제입니다.

  속성에 관한 자세한 내용은 주제를 참고하세요.

• Cloud Storage 버킷: 데이터를 저장할 Cloud Storage 버킷입니다.

  속성에 관한 자세한 내용은 버킷을 참고하세요.

• 구성: 커넥터의 추가 구성 설정입니다.

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

• 작업 재시작 정책: 실패한 커넥터 작업을 다시 시작하는 정책입니다.

  속성에 관한 자세한 내용은 작업 다시 시작 정책을 참고하세요.

Pub/Sub 소스 커넥터

• Pub/Sub 구독: 메시지를 수신할 Pub/Sub 구독입니다.
• Kafka 주제: 메시지를 스트리밍할 Kafka 주제입니다.
• 구성: 커넥터의 추가 구성 설정입니다. 자세한 내용은 커넥터 구성을 참고하세요.
• 작업 재시작 정책: 실패한 커넥터 작업을 다시 시작하는 정책입니다. 자세한 내용은 작업 다시 시작 정책을 참고하세요.

Pub/Sub 싱크 커넥터

• 주제: 메시지를 스트리밍할 Kafka 주제입니다.

  속성에 관한 자세한 내용은 주제를 참고하세요.

• Pub/Sub 주제: 메시지를 보낼 Pub/Sub 주제입니다.

  속성에 대한 자세한 내용은 Pub/Sub 주제를 참고하세요.

• 구성: 커넥터의 추가 구성 설정입니다.

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

• 작업 재시작 정책: 실패한 커넥터 작업을 다시 시작하는 정책입니다.

  속성에 관한 자세한 내용은 작업 다시 시작 정책을 참고하세요.

커넥터 업데이트

커넥터를 업데이트하면 변경사항이 적용되는 동안 데이터 흐름이 일시적으로 중단될 수 있습니다.

콘솔

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

   클러스터 연결로 이동

2. 업데이트하려는 커넥터를 호스팅하는 Connect 클러스터를 클릭합니다.

   클러스터 연결 세부정보 페이지가 표시됩니다.

3. 리소스 탭에서 목록에 있는 커넥터를 찾아 이름을 클릭합니다.

   커넥터 세부정보 페이지로 리디렉션됩니다.

4. 수정을 클릭합니다.

5. 커넥터의 필수 속성을 업데이트합니다. 사용 가능한 속성은 커넥터 유형에 따라 다릅니다.

6. 저장을 클릭합니다.

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 connectors update 명령어를 사용하여 커넥터를 업데이트합니다.

   쉼표로 구분된 키-값 쌍이 있는 --configs 플래그 또는 JSON 또는 YAML 파일의 경로가 있는 --config-file 플래그를 사용하여 커넥터의 구성을 업데이트할 수 있습니다.

   다음은 쉼표로 구분된 키-값 쌍과 함께 --configs 플래그를 사용하는 구문입니다.

   gcloud managed-kafka connectors update CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID \
       --configs=KEY1=VALUE1,KEY2=VALUE2...
   

   다음은 JSON 또는 YAML 파일의 경로와 함께 --config-file 플래그를 사용하는 구문입니다.

   gcloud managed-kafka connectors update CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID \
       --config-file=PATH_TO_CONFIG_FILE
   

   다음을 바꿉니다.

   • CONNECTOR_ID: (필수사항) 업데이트할 커넥터의 ID입니다.
   • LOCATION: (필수사항) 커넥터가 포함된 Connect 클러스터의 위치입니다.
   • CONNECT_CLUSTER_ID: (필수사항) 커넥터가 포함된 Connect 클러스터의 ID입니다.
   • KEY1=VALUE1,KEY2=VALUE2...: 업데이트할 쉼표로 구분된 구성 속성입니다. 예를 들면 tasks.max=2,value.converter.schemas.enable=true입니다.
   • PATH_TO_CONFIG_FILE: 업데이트할 구성 속성이 포함된 JSON 또는 YAML 파일의 경로입니다. 예를 들면 config.json입니다.

   --configs을 사용하는 명령어의 예:

   gcloud managed-kafka connectors update test-connector \
       --location=us-central1 \
       --connect-cluster=test-connect-cluster \
       --configs=tasks.max=2,value.converter.schemas.enable=true
   

   --config-file을 사용하는 명령어의 예 다음은 update_config.yaml이라는 샘플 파일입니다.

   tasks.max: 3
   topic: updated-test-topic
   

   다음은 파일을 사용하는 샘플 명령어입니다.

   gcloud managed-kafka connectors update test-connector \
       --location=us-central1 \
       --connect-cluster=test-connect-cluster \
       --config-file=update_config.yaml
   

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

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

func updateConnector(w io.Writer, projectID, region, connectClusterID, connectorID string, config map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "my-connector"
	// config := map[string]string{"tasks.max": "6"}
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	connectorPath := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s/connectors/%s", projectID, region, connectClusterID, connectorID)
	connector := &managedkafkapb.Connector{
		Name:    connectorPath,
		Configs: config,
	}
	paths := []string{"configs"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateConnectorRequest{
		UpdateMask: updateMask,
		Connector:  connector,
	}
	resp, err := client.UpdateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Updated connector: %#v\n", resp)
	return nil
}

자바

이 샘플을 시도하기 전에 클라이언트 라이브러리 설치의 Java 설정 안내를 따르세요. 자세한 내용은 Apache Kafka용 관리형 서비스 Java API 참조 문서를 참고하세요.

Managed Service for Apache Kafka에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 ADC 설정을 참고하세요.

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.protobuf.FieldMask;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class UpdateConnector {

  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 connectorId = "my-connector";
    // The new value for the 'tasks.max' configuration.
    String maxTasks = "5";
    updateConnector(projectId, region, clusterId, connectorId, maxTasks);
  }

  public static void updateConnector(
      String projectId, String region, String clusterId, String connectorId, String maxTasks)
      throws IOException {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      Map<String, String> configMap = new HashMap<>();
      configMap.put("tasks.max", maxTasks);

      Connector connector =
          Connector.newBuilder()
              .setName(ConnectorName.of(projectId, region, clusterId, connectorId).toString())
              .putAllConfigs(configMap)
              .build();

      // The field mask specifies which fields to update. Here, we update the 'config' field.
      FieldMask updateMask = FieldMask.newBuilder().addPaths("config").build();

      // This operation is handled synchronously.
      Connector updatedConnector = managedKafkaConnectClient.updateConnector(connector, updateMask);
      System.out.printf("Updated connector: %s\n", updatedConnector.getName());
      System.out.println(updatedConnector.getAllFields());

    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.updateConnector got err: %s\n", 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
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector
from google.protobuf import field_mask_pb2

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"
# connector_id = "my-connector"
# configs = {
#     "tasks.max": "6",
#     "value.converter.schemas.enable": "true"
# }

connect_client = ManagedKafkaConnectClient()

connector = Connector()
connector.name = connect_client.connector_path(
    project_id, region, connect_cluster_id, connector_id
)
connector.configs = configs
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.append("config")

# For a list of editable fields, one can check https://cloud.google.com/managed-service-for-apache-kafka/docs/connect-cluster/update-connector#editable-properties.
request = managedkafka_v1.UpdateConnectorRequest(
    update_mask=update_mask,
    connector=connector,
)

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

You can also update the connector's task restart policy without
including the configuration, by using the `--task-restart-min-backoff`
and `--task-restart-max-backoff` flags. For example:

```sh
gcloud managed-kafka connectors update test-connector \
  --location=us-central1 \
  --connect-cluster=test-connect-cluster \
  --task-restart-min-backoff="60s" \
  --task-restart-max-backoff="90s"

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