Mettre à jour un cluster Connect

Vous pouvez modifier un cluster Connect pour mettre à jour des propriétés telles que le nombre de processeurs virtuels, 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 pour l'emplacement spécifié : managedkafka.connectClusters.update
• Accordez l'autorisation "Afficher un 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.

Pour en savoir plus sur ce rôle, consultez Rôles prédéfinis de Managed Service pour Apache Kafka.

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 mettre à jour les propriétés suivantes du cluster Connect :

Property (Propriété)	Modifiable
Processeurs virtuels	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 Connecter les détails 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. 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. 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 processeurs virtuels.
   • 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 processeur virtuel.
   • DNS_NAME : nom de domaine DNS du réseau du sous-réseau à rendre visible pour le cluster Connect.
   • LABELS : (facultatif) libellés à associer au cluster. Pour en savoir plus sur le format des libellés, consultez Libellés. Liste de paires 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 workers. Les versions exactes des secrets de Secret Manager doivent être fournies. Les alias ne sont pas acceptés. Jusqu'à 32 secrets peuvent être chargés 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 Managed Service pour Apache Kafka en langage Go.

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 (
	"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 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.

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

• Lister tous les clusters Connect

• Afficher les détails d'un cluster Connect

• Supprimer un cluster Connect

• Présentation des connecteurs

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