Mettre à jour un cluster Connect

Vous pouvez modifier un cluster Connect pour mettre à jour des propriétés telles que le nombre de vCPU, la mémoire, le réseau et les libellés.

Pour modifier un cluster Connect, vous pouvez utiliser la console Google Cloud , la gcloud CLI, la bibliothèque cliente ou l'API Managed Kafka. Vous ne pouvez pas utiliser l'API Apache Kafka Open Source pour mettre à jour un cluster Connect.

Avant de commencer

Toutes les propriétés d'un cluster Connect ne sont pas modifiables. Examinez les propriétés d'un cluster Connect avant de le mettre à jour.

Rôles et autorisations requis pour modifier un cluster Connect

Pour obtenir les autorisations nécessaires pour modifier un cluster Connect, demandez à votre administrateur de vous accorder le rôle IAM Éditeur de cluster Kafka Connect géré (roles/managedkafka.connectClusterEditor) sur votre projet. Pour en savoir plus sur l'attribution de rôles, consultez Gérer l'accès aux projets, aux dossiers et aux organisations.

Ce rôle prédéfini contient les autorisations requises pour modifier un cluster Connect. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

Autorisations requises

Les autorisations suivantes sont requises pour modifier un cluster Connect :

  • Accordez à la mise à jour une autorisation de cluster Connect à l'emplacement spécifié : managedkafka.connectClusters.update
  • Accordez à la vue une autorisation de cluster Connect à l'emplacement spécifié. Cette autorisation n'est requise que pour mettre à jour un cluster Connect à l'aide de la console Google Cloud  : managedkafka.connectors.list

Vous pouvez également obtenir ces autorisations avec des rôles personnalisés ou d'autres rôles prédéfinis.

Modifier un cluster Connect

La mise à jour de certaines propriétés, telles que le processeur et la mémoire, nécessite un redémarrage du cluster.

Les redémarrages de cluster préservent les données, mais peuvent augmenter la latence. Le nombre initial de nœuds de calcul dans le cluster détermine la durée du redémarrage.

Vous pouvez modifier les propriétés suivantes du cluster Connect :

Propriété Modifiable
vCPU Oui
Mémoire Oui
Réseau Oui
Sous-réseau de nœuds de calcul Oui
Domaines DNS pouvant être résolus Oui (ajouter/supprimer)
Nom du cluster Connect Non
Cluster Kafka Non
Emplacement Non
Étiquettes Oui (ajouter/modifier/supprimer)
Secrets Oui (ajouter/supprimer)

Console

  1. Dans la console Google Cloud , accédez à la page Connecter des clusters.

    Accéder à "Connect Clusters"

  2. Cliquez sur le cluster Connect que vous souhaitez mettre à jour.

    La page Détails de la connexion du cluster s'affiche.

  3. Cliquez sur Modifier.

    La page Modifier le cluster Kafka Connect s'affiche.

  4. Apportez les modifications nécessaires aux propriétés modifiables.

  5. Cliquez sur Enregistrer.

gcloud

  1. Dans la console Google Cloud , activez Cloud Shell.

    Activer Cloud Shell

    En bas de la console Google Cloud , une session Cloud Shell démarre et affiche une invite de ligne de commande. Cloud Shell est un environnement shell dans lequel Google Cloud CLI est déjà installé, et dans lequel des valeurs sont déjà définies pour votre projet actuel. L'initialisation de la session peut prendre quelques secondes.

  2. Exécutez la commande gcloud managed-kafka connect-clusters update :

    gcloud managed-kafka connect-clusters update CONNECT_CLUSTER_ID \
    --location=LOCATION \
    [--cpu=CPU --memory=MEMORY
     | --clear-dns-names \
     | --dns-name=DNS_NAME --clear-labels \
     | --labels=LABELS --clear-secrets \
     | --secret=SECRET [--primary-subnet=WORKER_SUBNET \
    [--async]

    Remplacez les éléments suivants :

    • CONNECT_CLUSTER_ID : ID ou nom du cluster Connect. Le nom d'un cluster Connect est immuable.
    • LOCATION : emplacement du cluster Connect. L'emplacement d'un cluster Connect est immuable.
    • CPU : nombre de processeurs virtuels pour le cluster Connect. La valeur minimale est de trois vCPU.
    • MEMORY : quantité de mémoire pour le cluster Connect. Utilisez les unités "MB", "MiB", "GB", "GiB", "TB" ou "TiB". Par exemple, "10GiB". Vous devez provisionner entre 1 Gio et 8 Gio par vCPU.

    • DNS_NAME : nom de domaine DNS du réseau du sous-réseau à rendre visible pour le cluster Connect.
    • LABELS : (facultatif) étiquettes à associer au cluster. Pour en savoir plus sur le format des libellés, consultez Libellés. Liste de paires "KEY=VALUE" (clé/valeur) de libellés à ajouter. Les clés doivent commencer par une lettre minuscule et ne contenir que des tirets (-), des traits de soulignement (_), des minuscules et des chiffres. Les valeurs ne doivent contenir que des tirets (-), des traits de soulignement (_), des minuscules et des chiffres.
    • SECRET : (facultatif) secrets à charger dans les nœuds de calcul. Vous devez fournir les versions exactes des secrets de Secret Manager. Les alias ne sont pas acceptés. Vous pouvez charger jusqu'à 32 secrets dans un même cluster. Format : projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID

    • WORKER_SUBNET : sous-réseau de nœuds de calcul pour le cluster Connect. Le sous-réseau de nœuds de calcul doit se trouver dans la même région que le cluster Connect.

      Le format du sous-réseau est projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.

Go

Avant d'essayer cet exemple, suivez les instructions de configuration pour Go dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Go pour Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les identifiants par défaut de l'application(ADC, Application Default Credentials). Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement local.

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 updateConnectCluster(w io.Writer, projectID, region, clusterID string, memoryBytes int64, labels map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-connect-cluster"
	// memoryBytes := 25769803776 // 24 GiB in bytes
	// labels := map[string]string{"environment": "production"}
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

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

	// Capacity configuration update
	capacityConfig := &managedkafkapb.CapacityConfig{
		MemoryBytes: memoryBytes,
	}

	connectCluster := &managedkafkapb.ConnectCluster{
		Name:           clusterPath,
		CapacityConfig: capacityConfig,
		Labels:         labels,
	}
	paths := []string{"capacity_config.memory_bytes", "labels"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateConnectClusterRequest{
		UpdateMask:     updateMask,
		ConnectCluster: connectCluster,
	}
	op, err := client.UpdateConnectCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateConnectCluster 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 connect cluster: %#v\n", resp)
	return nil
}

Java

Avant d'essayer cet exemple, suivez les instructions de configuration pour Java dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Java Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement local.


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.ConnectCluster;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.UpdateConnectClusterRequest;
import com.google.protobuf.FieldMask;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class UpdateConnectCluster {

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

  public static void updateConnectCluster(
      String projectId, String region, String clusterId, long memoryBytes) throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setMemoryBytes(memoryBytes).build();
    ConnectCluster connectCluster = ConnectCluster.newBuilder()
        .setName(ConnectClusterName.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
    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.updateConnectClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create(
        settingsBuilder.build())) {
      UpdateConnectClusterRequest request = UpdateConnectClusterRequest.newBuilder()
          .setUpdateMask(updateMask)
          .setConnectCluster(connectCluster).build();
      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
          .updateConnectClusterOperationCallable().futureCall(request);

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

      ConnectCluster response = future.get();
      System.out.printf("Updated connect cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaConnectClient.updateConnectCluster got err: %s\n", 
          e.getMessage());
    }
  }
}

Python

Avant d'essayer cet exemple, suivez les instructions de configuration pour Python dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Python pour Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement local.

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 ConnectCluster
from google.protobuf import field_mask_pb2

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

connect_client = ManagedKafkaConnectClient()

connect_cluster = ConnectCluster()
connect_cluster.name = connect_client.connect_cluster_path(
    project_id, region, connect_cluster_id
)
connect_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-service-for-apache-kafka/docs/connect-cluster/create-connect-cluster#properties.
request = managedkafka_v1.UpdateConnectClusterRequest(
    update_mask=update_mask,
    connect_cluster=connect_cluster,
)

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

Étape suivante

Apache Kafka® est une marque déposée d'Apache Software Foundation ou de ses sociétés affiliées aux États-Unis et/ou dans d'autres pays.