Aggiorna un cluster Google Cloud Managed Service per Apache Kafka

Puoi modificare un cluster Managed Service per Apache Kafka per aggiornare proprietà come le dimensioni del cluster (numero di vCPU e memoria), l'elenco delle subnet connesse, la configurazione del ribilanciamento automatico e la configurazione mTLS.

Per modificare un cluster, puoi utilizzare la console Google Cloud , Google Cloud CLI, la libreria client o l'API Managed Kafka. Non puoi utilizzare l'API Apache Kafka open source per aggiornare un cluster.

L'aggiornamento di alcune proprietà, come il conteggio di vCPU e la memoria, potrebbe richiedere il riavvio del cluster da parte del servizio. Il cluster viene riavviato un broker alla volta. Durante questo processo, le richieste ai singoli broker potrebbero non andare a buon fine, ma questi errori sono temporanei. Le librerie client di uso comune gestiscono automaticamente questi errori.

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per aggiornare un cluster, chiedi all'amministratore di concederti il ruolo IAM Editor cluster Kafka gestito (roles/managedkafka.clusterEditor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per aggiornare un cluster. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per aggiornare un cluster sono necessarie le seguenti autorizzazioni:

• Modifica un cluster: managedkafka.clusters.update

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Il ruolo Editor cluster Kafka gestito non consente di creare, eliminare o modificare argomenti e gruppi di consumatori nei cluster Managed Service per Apache Kafka. Né consente l'accesso al piano dati per pubblicare o utilizzare messaggi all'interno dei cluster. Per saperne di più su questo ruolo, consulta Ruoli predefiniti di Managed Service per Apache Kafka.

Ridimensionamento di un cluster

Se aggiorni il numero di vCPU o la memoria di un cluster, si applicano le seguenti regole:

• Il rapporto complessivo tra vCPU e memoria del cluster deve sempre rimanere compreso tra 1:1 e 1:8.

• Se esegui lo scale down, devono essere presenti almeno 1 vCPU e 1 GiB di memoria per ogni broker esistente. Il numero di broker non diminuisce mai.

• Se esegui l'upgrade e la modifica comporta l'aggiunta di nuovi broker, la vCPU media e la memoria per broker non possono diminuire di oltre il 10% rispetto alle medie prima dell'aggiornamento.

  Ad esempio, se provi ad aumentare le dimensioni di un cluster da 45 vCPU (3 broker) a 48 vCPU (4 broker), l'operazione non va a buon fine. Questo perché la vCPU media per broker diminuisce da 15 a 12, con una riduzione del 20%, che supera il limite del 10%.

Per saperne di più, vedi Aggiornare le dimensioni del cluster.

Modifica di un cluster

Per modificare un cluster:

Console

1. Nella console Google Cloud , vai alla pagina Cluster.

   Vai a Cluster

2. Nell'elenco dei cluster, fai clic su quello di cui vuoi modificare le proprietà.

   Viene visualizzata la pagina dei dettagli del cluster.

3. Nella pagina dei dettagli del cluster, fai clic su Modifica.

4. Modifica le proprietà in base alle esigenze. Le seguenti proprietà di un cluster sono modificabili dalla console:

   • Memoria
   • vCPU
   • Subnet
   • Configurazione del ribilanciamento
   • Configurazione mTLS
   • Etichette

5. Fai clic su Salva.

gcloud

1. Nella console Google Cloud , attiva Cloud Shell.

   Attiva Cloud Shell

   Nella parte inferiore della console Google Cloud viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

2. Esegui il comando gcloud managed-kafka clusters update:

   gcloud managed-kafka clusters update CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU \
       --memory=MEMORY \
       --subnets=SUBNETS \
       --auto-rebalance \
       --labels=LABELS
   

   Sostituisci quanto segue:

   • CLUSTER_ID: l'ID o il nome del cluster. Non puoi aggiornare questo valore.
   • LOCATION: la posizione del cluster. Non puoi aggiornare questo valore.
   • CPU: Il numero di CPU virtuali per il cluster.
   • MEMORY: La quantità di memoria per il cluster. Utilizza le unità "MB", "MiB", "GB", "GiB", "TB" o "TiB". Ad esempio, "10 GiB".
   • SUBNETS: l'elenco delle subnet a cui connettersi. Utilizza le virgole per separare più valori di subnet.
   • auto-rebalance: consente il ribilanciamento automatico delle partizioni degli argomenti tra i broker quando cambia il numero di CPU nel cluster. Questa opzione è attiva per impostazione predefinita.
   • LABELS: le etichette da associare al cluster.

Se utilizzi il flag --async con il comando, il sistema invia la richiesta di aggiornamento e restituisce immediatamente una risposta, senza attendere il completamento dell'operazione. Con il flag --async, puoi continuare con altre attività mentre l'aggiornamento del cluster avviene in background. Se non utilizzi il flag --async, il sistema attende il completamento dell'operazione prima di restituire una risposta. Devi attendere il completamento dell'aggiornamento del cluster prima di poter continuare con altre attività.

REST

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

• PROJECT_ID: il tuo Google Cloud ID progetto
• LOCATION: la posizione del cluster
• CLUSTER_ID: l'ID del cluster
• UPDATE_MASK: quali campi aggiornare, come un elenco separato da virgole di nomi completi. Esempio: capacityConfig.vcpuCount,capacityConfig.memoryBytes
• CPU_COUNT: il numero di vCPU per il cluster
• MEMORY: la quantità di memoria per il cluster, in byte
• SUBNET_ID: l'ID subnet della subnet a cui connettersi

Metodo HTTP e URL:

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

Corpo JSON della richiesta:

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

Per inviare la richiesta, espandi una di queste opzioni:

curl (Linux, macOS o Cloud Shell)

Nota: il seguente comando presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login oppure utilizzando Cloud Shell, che consente di accedere automaticamente all'interfaccia a riga di comando gcloud. Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente:

curl -X PATCH \
     -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/CLUSTER_ID?updateMask=UPDATE_MASK"

PowerShell (Windows)

Nota: il comando seguente presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login . Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK" | Select-Object -Expand Content

Dovresti ricevere una risposta JSON simile alla seguente:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID",
  "partitionCount": PARTITION_COUNT,
  "replicationFactor": REPLICATION_FACTOR
}

Nel corpo della richiesta, includi solo i campi che stai aggiornando, come specificato nel parametro di query UPDATE_MASK. Per aggiungere una subnet, aggiungi una nuova voce a networkConfigs.

Go

Prima di provare questo esempio, segui le istruzioni di configurazione di Go in Installare le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Go di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione(ADC). Per saperne di più, vedi Configura ADC per un ambiente di sviluppo locale.

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
}

Java

Prima di provare questo esempio, segui le istruzioni di configurazione di Java in Installare le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Java di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura ADC per un ambiente di sviluppo locale.

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

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in Installare le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Python di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, vedi Configura ADC per un ambiente di sviluppo locale.

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

Limitazioni

Dopo aver creato un cluster Managed Service per Apache Kafka, non puoi aggiornare le seguenti proprietà:

• Nome del cluster
• La località del cluster
• Il tipo di crittografia

Anche se non puoi modificare il tipo di crittografia, puoi ruotare le chiavi di crittografia.

Passaggi successivi

• Monitora il cluster.

• Pianifica le dimensioni del cluster.

• Crea un argomento all'interno del cluster.

• Visualizza il cluster.

• Elimina il cluster.

Apache Kafka® è un marchio registrato di Apache Software Foundation o delle sue affiliate negli Stati Uniti e/o in altri paesi.