Google utilise la technologie IA pour traduire le contenu dans votre langue préférée. Les traductions générées par IA peuvent contenir des erreurs.

Créer un cluster Google Cloud Managed Service pour Apache Kafka

Un cluster Managed Service pour Apache Kafka fournit un environnement permettant de stocker et de traiter des flux de messages organisés en sujets.

Pour créer un cluster, vous pouvez utiliser la console Google Cloud , la Google Cloud CLI, la bibliothèque cliente ou l'API Managed Kafka. Vous ne pouvez pas utiliser l'API Apache Kafka Open Source pour créer un cluster.

Avant de commencer

Assurez-vous de bien maîtriser ce qui suit :

Rôles et autorisations nécessaires

Pour obtenir l'autorisation nécessaire pour créer un cluster, demandez à votre administrateur de vous accorder le rôle IAM Éditeur de cluster Kafka géré (roles/managedkafka.clusterEditor) 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 l'autorisation managedkafka.clusters.create, qui est nécessaire pour créer un cluster.

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

Autorisations du sous-réseau du cluster

Lorsque vous créez un cluster Managed Service pour Apache Kafka, vous spécifiez un ou plusieurs sous-réseaux de cloud privé virtuel (VPC) où le cluster est accessible. Lors de la création du cluster, l'agent de service Managed Kafka associe ces sous-réseaux au cluster.

L'autorisation managedkafka.clusters.create permet indirectement à un compte principal d'associer n'importe quel sous-réseau que l'agent de service est autorisé à utiliser, même si le compte principal n'est pas directement autorisé à utiliser le sous-réseau.

Propriétés d'un cluster Managed Service pour Apache Kafka

Lorsque vous créez ou mettez à jour un cluster Managed Service pour Apache Kafka, vous devez spécifier les propriétés suivantes.

Nom du cluster

Nom ou ID du cluster Managed Service pour Apache Kafka que vous créez. Pour obtenir des instructions sur la façon de nommer un cluster, consultez les consignes de dénomination des ressources Managed Service pour Apache Kafka. Le nom d'un cluster est immuable.

Emplacement

Emplacement où vous créez le cluster. L'emplacement doit être l'une des régions Google Cloud compatibles. L'emplacement d'un cluster ne peut pas être modifié par la suite. Pour obtenir la liste des emplacements disponibles, consultez Emplacements Managed Service pour Apache Kafka.

Configuration de la capacité

Capacité de calcul, de mémoire et de stockage du cluster Kafka. Lorsque vous créez un cluster, vous devez spécifier les valeurs suivantes :

vCPUs : nombre de processeurs virtuels dans le cluster. Au moins trois processeurs virtuels par cluster sont requis.
Mémoire : quantité de mémoire attribuée au cluster. Vous devez provisionner entre 1 Gio et 8 Gio par vCPU.

Par exemple, si vous créez un cluster avec six processeurs virtuels, la mémoire minimale que vous pouvez lui allouer est de 6 Gio (1 Gio par processeur virtuel) et la mémoire maximale est de 48 Gio (8 Gio par processeur virtuel).

Le service provisionne automatiquement des agents, y compris des volumes Persistent Disk pour les agents. Vous pouvez éventuellement activer l'option Espace de stockage local personnalisé, qui vous permet de spécifier la capacité du disque pour chaque agent. Pour en savoir plus, consultez Configurer la taille du disque de l'agent.

Pour savoir comment estimer la capacité dont votre cluster a besoin, consultez Planifier la taille de votre cluster Kafka.

Configuration du réseau

La configuration réseau est une liste de sous-réseaux VPC où le cluster est accessible. Pour produire ou consommer des messages, les clients doivent pouvoir accéder à l'un de ces sous-réseaux.

Voici quelques consignes pour la configuration de votre réseau :

Un cluster doit comporter au moins un sous-réseau. La valeur maximale est de 10.
Un seul sous-réseau par réseau est autorisé pour un cluster donné.
Chaque sous-réseau doit se trouver dans la même région que le cluster. Le projet et le réseau peuvent être différents.
Les adresses IP des agents et du serveur d'amorçage sont automatiquement allouées dans chaque sous-réseau. De plus, des entrées DNS pour chacune de ces adresses IP sont créées dans les réseaux VPC correspondants.
Si vous ajoutez un sous-réseau d'un autre projet, vous devez accorder des autorisations au compte de service géré par Google associé au cluster. Pour en savoir plus, consultez Connecter un cluster entre plusieurs projets.

Après avoir créé le cluster, vous pouvez mettre à jour la liste des sous-réseaux. Pour en savoir plus sur la mise en réseau, consultez Configurer la mise en réseau pour Managed Service pour Apache Kafka.

L'agent de service Kafka géré utilise les sous-réseaux du cluster. L'agent est autorisé à utiliser les sous-réseaux via le rôle IAM Agent du service Kafka géré (roles/managedkafka.serviceAgent).

Clusters publics

Vous pouvez éventuellement configurer votre cluster en tant que cluster public pour permettre aux applications clientes de s'y connecter et d'exécuter des opérations Kafka (y compris l'envoi et la réception de messages) sur l'Internet public. Pour activer l'accès public, vous devez spécifier une ou plusieurs plages d'adresses IP sources autorisées.

Lorsque vous configurez des plages d'adresses IP sources autorisées, les règles suivantes s'appliquent :

Toutes les plages d'adresses IP sources autorisées doivent être spécifiées au format CIDR IPv4 (par exemple, 203.0.113.0/24 ou 198.51.100.5/32).
La taille de chaque sous-réseau CIDR doit être comprise entre /16 et /32.
Les plages CIDR ne doivent pas se chevaucher.
Toutes les plages d'adresses IP sources autorisées doivent être routables publiquement. Les plages d'adresses IP privées, telles que 10.0.0.0/8 (RFC 1918), sont refusées.
Vous pouvez spécifier jusqu'à 500 plages d'adresses IP sources autorisées.
Les adresses IPv6 ne sont pas compatibles.

Avant d'activer l'accès public pour un cluster, nous vous recommandons de lire Connecter des clients à un cluster public pour découvrir comment la fonctionnalité d'accès public est implémentée.

Étiquettes

Les étiquettes sont des paires clé/valeur qui vous aident à organiser et à identifier vos ressources. Les libellés vous permettent de classer les ressources en fonction de l'environnement. Exemples : "env:production" et "owner:data-engineering".

Vous pouvez filtrer et rechercher des ressources en fonction de leurs étiquettes. Par exemple, supposons que vous disposiez de plusieurs clusters Managed Service pour Apache Kafka pour différents services. Vous pouvez configurer et rechercher des clusters avec l'étiquette "department:marketing" pour trouver celui qui vous intéresse.

Configuration du rééquilibrage

Ce paramètre détermine si le service rééquilibre automatiquement les instances répliquées de partition entre les agents.

Les modes disponibles sont les suivants :

Rééquilibrage automatique lors du scaling à la hausse : lorsque cette option est activée, le service déclenche automatiquement un rééquilibrage des instances répliquées lorsque vous effectuez un scaling à la hausse du cluster. Ce mode permet de maintenir une répartition uniforme de la charge, mais peut affecter temporairement les performances pendant l'opération de rééquilibrage.
Aucun rééquilibrage : lorsque cette option est activée, le service ne rééquilibre pas automatiquement les instances répliquées.

Pour en savoir plus, consultez Rééquilibrer les partitions de l'agent.

Chiffrement

Managed Service pour Apache Kafka peut chiffrer les messages avecGoogle-owned and Google-managed encryption keys (par défaut) ou des clés de chiffrement gérées par le client (CMEK). Tous les messages sont chiffrés au repos et en transit. Le type de chiffrement d'un cluster est immuable.

Google-owned and Google-managed encryption keys sont utilisées par défaut. Ces clés sont créées, gérées et stockées entièrement par Google Cloud dans son infrastructure.

Les CMEK sont des clés de chiffrement que vous gérez à l'aide de Cloud Key Management Service. Cette fonctionnalité vous permet de mieux contrôler les clés utilisées pour chiffrer les données au repos dans les services Google Cloud compatibles. L'utilisation de CMEK entraîne des coûts supplémentaires liés à Cloud Key Management Service. Pour utiliser une clé CMEK, votre trousseau de clés doit se trouver au même emplacement que les ressources avec lesquelles vous l'utilisez. Pour en savoir plus, consultez Configurer le chiffrement des messages.

Configuration mTLS

Vous pouvez éventuellement configurer mTLS comme méthode d'authentification alternative qui utilise des certificats clients. La configuration comprend les éléments suivants :

Pools d'autorités de certification : liste de 1 à 10 pools Certificate Authority Service (CAS) auxquels le cluster fait confiance pour l'authentification du client.
Règles de mappage des principaux SSL : propriété d'agent ssl.principal.mapping.rules facultative, mais recommandée, permettant de simplifier les noms de principaux de certificat longs pour les utiliser dans les LCA Kafka.

Pour en savoir plus sur mTLS, consultez Configurer l'authentification mTLS.

Créer un cluster

Avant de créer un cluster, consultez la documentation sur les propriétés de cluster.

La création d'un cluster prend généralement 20 à 30 minutes.

Pour créer un cluster, procédez comme suit :

Console

Dans la console Google Cloud , accédez à la page Clusters.

accéder aux clusters
Sélectionnez Créer.

La page Créer un cluster Kafka s'ouvre.
Dans le champ Nom du cluster, saisissez une chaîne.

Pour en savoir plus sur la dénomination d'un cluster, consultez les consignes de dénomination des ressources Managed Service pour Apache Kafka.
Dans le champ Emplacement, saisissez un emplacement compatible.

Pour en savoir plus sur les emplacements compatibles, consultez Emplacements compatibles avec Managed Service pour Apache Kafka.
Pour Configuration de la capacité, saisissez des valeurs pour Mémoire et vCPU.

Pour en savoir plus sur le dimensionnement d'un cluster, consultez Planifier la taille de votre cluster Kafka.
Facultatif. Pour spécifier la capacité du disque de l'agent, procédez comme suit :
1. Cliquez sur le bouton Activer la taille de l'espace de stockage local personnalisée pour l'activer.
  
  Important : Une fois que vous avez cliqué sur Enregistrer, vous ne pouvez plus désactiver ce paramètre pour ce cluster.
2. Dans le champ Espace de stockage local personnalisé, saisissez la taille du disque par agent, en Gio.
Si vous laissez l'option Activer la taille de l'espace de stockage local personnalisée désactivée, la capacité de disque par défaut est provisionnée. Pour en savoir plus, consultez Configurer la taille du disque de l'agent.
Pour Configuration du réseau, saisissez les informations suivantes :
1. Projet : projet dans lequel se trouve le sous-réseau. Le sous-réseau doit se trouver dans la même région que le cluster, mais le projet peut être différent.
2. Réseau : réseau auquel le sous-réseau est connecté.
3. Sous-réseau : nom du sous-réseau.
4. Chemin d'URI du sous-réseau : ce champ est renseigné automatiquement. Vous pouvez également saisir le chemin d'accès au sous-réseau ici. Le nom du sous-réseau doit être au format suivant : projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.
5. Cliquez sur OK.
Facultatif : pour ajouter d'autres sous-réseaux, cliquez sur Ajouter un sous-réseau connecté.

Vous pouvez ajouter d'autres sous-réseaux, jusqu'à un maximum de 10.
Facultatif : Activez l'accès public à votre cluster.
1. Dans la section Configuration du cluster public, sous Le cluster est accessible, sélectionnez Dans un réseau privé et via l'Internet public.
2. Dans le champ Liste des plages d'adresses IPv4 sources autorisées, séparées par une virgule, au format CIDR, saisissez les plages d'adresses IP de vos applications clientes au format CIDR IPv4 (par exemple, 203.0.113.0/24). Les adresses IPv6 ne sont pas acceptées.
Facultatif : Définissez une clé de chiffrement gérée par le client (CMEK).
1. Sous Chiffrement, sélectionnez Clé Cloud KMS.
2. Pour Type de clé, sélectionnez Cloud KMS.
3. Dans le champ Sélectionner une clé gérée par le client, sélectionnez ou saisissez la CMEK.
  
  Si cette option ne s'affiche pas, assurez-vous d'avoir activé l'API Cloud KMS pour le projet.
Pour en savoir plus, consultez Configurer le chiffrement des messages.
Cliquez sur Créer.

gcloud

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 de shell dans lequel la Google Cloud CLI est déjà installée, et dans lequel des valeurs sont déjà définies pour votre projet actuel. L'initialisation de la session peut prendre quelques secondes.

Avant d'utiliser les données de la commande ci-dessous, effectuez les remplacements suivants :

PROJECT_ID : ID du projet.
LOCATION : emplacement du cluster.
CLUSTER_ID : ID du cluster.
CPU_COUNT : nombre de processeurs virtuels pour le cluster.
MEMORY : quantité de mémoire pour le cluster. Exemple : 3GiB.
SUBNET_ID : ID du sous-réseau auquel se connecter. Exemple : default.
ENCRYPTION_KEY : ID de la clé de chiffrement gérée par le client à utiliser pour le cluster. Cette opération est facultative.
LABELS : libellés à associer au cluster.
ALLOWED_SOURCE_IP_RANGES : plages d'adresses IP sources autorisées pour le cluster public. Cette opération est facultative.

Exécutez la commande suivante :

Linux, macOS ou Cloud Shell

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

Windows (PowerShell)

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

Windows (cmd.exe)

gcloud managed-kafka clusters create CLUSTER_ID ^
    --location=LOCATION ^
    --cpu=CPU_COUNT ^
    --memory=MEMORY ^
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
    --auto-rebalance ^
    --encryption-key=ENCRYPTION_KEY ^
    --async ^
    --labels=LABELS ^
    --public-cluster ^
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

Vous devriez obtenir un résultat semblable à celui-ci :

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: create
name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID

Stockez le OPERATION_ID pour suivre progress.

REST

Avant d'utiliser les données de requête, effectuez les remplacements suivants :

PROJECT_ID : ID de votre projet Google Cloud
LOCATION : emplacement du cluster.
CLUSTER_ID : ID du cluster.
CPU_COUNT : nombre de processeurs virtuels pour le cluster.
MEMORY : quantité de mémoire pour le cluster, en octets. Exemple : 3221225472.
SUBNET_ID : ID du sous-réseau auquel se connecter. Exemple : default.

Méthode HTTP et URL :

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID

Corps JSON de la requête :

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

Pour envoyer votre requête, développez l'une des options suivantes :

curl (Linux, macOS ou Cloud Shell)

Remarque : La commande suivante suppose que vous êtes connecté à la CLI gcloud avec votre compte utilisateur en exécutant la commande gcloud init ou gcloud auth login, ou en utilisant Cloud Shell, qui vous connecte automatiquement à la CLI gcloud. Vous pouvez exécuter la commande gcloud auth list pour vérifier quel est le compte actuellement actif.

Enregistrez le corps de la requête dans un fichier nommé request.json, puis exécutez la commande suivante :

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)

Remarque : La commande suivante suppose que vous vous êtes connecté à la CLI gcloud avec votre compte utilisateur en exécutant la commande gcloud init ou gcloud auth login. Vous pouvez exécuter la commande gcloud auth list pour vérifier quel est le compte actuellement actif.

Enregistrez le corps de la requête dans un fichier nommé request.json, puis exécutez la commande suivante :

$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

Vous devriez recevoir une réponse JSON de ce type :

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

Vous pouvez utiliser une ressource Terraform pour créer un cluster.

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

Pour savoir comment appliquer ou supprimer une configuration Terraform, consultez Commandes Terraform de base.

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

	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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Java décrites 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.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

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 de Managed Service pour Apache Kafka.

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}")

Surveiller l'opération de création du cluster

Vous ne pouvez exécuter la commande suivante que si vous avez utilisé la gcloud CLI pour créer le cluster.

La création d'un cluster prend généralement 20 à 30 minutes. Pour suivre la progression de la création du cluster, la commande gcloud managed-kafka clusters create utilise une opération de longue durée (LRO), que vous pouvez surveiller à l'aide de la commande suivante :
```
gcloud managed-kafka operations describe OPERATION_ID \
    --location=LOCATION
```
Remplacez les éléments suivants :
- OPERATION_ID par la valeur de l'ID d'opération de la section précédente.
- LOCATION par la valeur de l'emplacement de la section précédente.

Résoudre les problèmes

Pour en savoir plus sur la résolution des erreurs liées à la création d'un cluster, consultez Erreurs de création de cluster Kafka.

Étape suivante

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.