Aggiorna un argomento Google Cloud Managed Service per Apache Kafka

Dopo aver creato un argomento, puoi modificarne la configurazione per aggiornare queste proprietà: il numero di partizioni e le configurazioni degli argomenti che non utilizzano per impostazione predefinita le proprietà già impostate a livello di cluster. Puoi solo aumentare il numero di partizioni, non diminuirlo.

Per aggiornare un singolo argomento, puoi utilizzare la Google Cloud console, Google Cloud CLI, la libreria client, l'API Managed Kafka o le API Apache Kafka open source.

Ruoli e autorizzazioni richiesti per modificare un argomento

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

Questo ruolo predefinito include le autorizzazioni necessarie per modificare un argomento. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per modificare un argomento sono necessarie le seguenti autorizzazioni:

  • Aggiorna un argomento: managedkafka.topics.update

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

Per saperne di più su questo ruolo, consulta Ruoli predefiniti di Managed Service per Apache Kafka.

Modificare un argomento

Per modificare un argomento:

Console

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

    Vai a Cluster

    Vengono elencati i cluster creati in un progetto.

  2. Fai clic sul cluster a cui appartiene l'argomento che vuoi modificare.

    Si apre la pagina Dettagli cluster. Nella pagina dei dettagli del cluster, nella scheda Risorse sono elencati gli argomenti.

  3. Fai clic sull'argomento che vuoi modificare.

    Si apre la pagina Dettagli argomento.

  4. Per apportare le modifiche, fai clic su Modifica.

  5. Fai clic su Salva dopo aver apportato le modifiche.

gcloud

  1. Nella Google Cloud console, attiva Cloud Shell.

    Attiva Cloud Shell

    Nella parte inferiore della Google Cloud console 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à inclusa e installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

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

    gcloud managed-kafka topics update TOPIC_ID \
  --cluster=CLUSTER_ID \
  --location=LOCATION_ID \
  --partitions=PARTITIONS \
  --configs=CONFIGS

    Questo comando modifica la configurazione di un argomento esistente nel cluster Managed Service per Apache Kafka specificato. Puoi utilizzare questo comando per aumentare il numero di partizioni e aggiornare le impostazioni di configurazione a livello di argomento.

    Sostituisci quanto segue:

    • TOPIC_ID: l'ID dell'argomento.
    • CLUSTER_ID: l'ID del cluster contenente l'argomento.
    • LOCATION_ID: la località del cluster.
    • PARTITIONS: (facoltativo) il numero aggiornato di partizioni per l'argomento. Puoi solo aumentare il numero di partizioni, non diminuirlo.
    • CONFIGS: (facoltativo) un elenco di impostazioni di configurazione da aggiornare. Specifica come elenco separato da virgole di coppie chiave-valore. Ad esempio, retention.ms=3600000,retention.bytes=10000000.

REST

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

  • PROJECT_ID: il tuo Google Cloud ID progetto
  • LOCATION: la località del cluster
  • CLUSTER_ID: l'ID del cluster
  • TOPIC_ID: l'ID dell'argomento
  • UPDATE_MASK: i campi da aggiornare, come elenco separato da virgole di nomi completi. Esempio: partitionCount
  • PARTITION_COUNT: il numero aggiornato di partizioni per l'argomento

Metodo HTTP e URL: 

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

Corpo JSON della richiesta: 

{
  "name": "TOPIC_ID",
  "partitionCount": PARTITION_COUNT
}

Per inviare la richiesta, espandi una di queste opzioni:

Dovresti ricevere una risposta JSON simile alla seguente:

{
  "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": "update",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Vai

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 a Managed Service per Apache Kafka, configura le credenziali predefinite dell'applicazione(ADC). Per saperne di più, consulta Configurare le credenziali 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 updateTopic(w io.Writer, projectID, region, clusterID, topicID string, partitionCount int32, configs map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	// partitionCount := 20
	// configs := map[string]string{"min.insync.replicas":"1"}
	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)
	topicPath := fmt.Sprintf("%s/topics/%s", clusterPath, topicID)
	TopicConfig := managedkafkapb.Topic{
		Name:           topicPath,
		PartitionCount: partitionCount,
		Configs:        configs,
	}
	paths := []string{"partition_count", "configs"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateTopicRequest{
		UpdateMask: updateMask,
		Topic:      &TopicConfig,
	}
	topic, err := client.UpdateTopic(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Updated topic: %#v\n", topic)
	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 a Managed Service per Apache Kafka, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare le credenziali ADC per un ambiente di sviluppo locale.

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.Topic;
import com.google.cloud.managedkafka.v1.TopicName;
import com.google.cloud.managedkafka.v1.UpdateTopicRequest;
import com.google.protobuf.FieldMask;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class UpdateTopic {

  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 topicId = "my-topic";
    int partitionCount = 200;
    Map<String, String> configs =
        new HashMap<String, String>() {
          {
            put("min.insync.replicas", "1");
          }
        };
    updateTopic(projectId, region, clusterId, topicId, partitionCount, configs);
  }

  public static void updateTopic(
      String projectId,
      String region,
      String clusterId,
      String topicId,
      int partitionCount,
      Map<String, String> configs)
      throws Exception {
    Topic topic =
        Topic.newBuilder()
            .setName(TopicName.of(projectId, region, clusterId, topicId).toString())
            .setPartitionCount(partitionCount)
            .putAllConfigs(configs)
            .build();
    String[] paths = {"partition_count", "configs"};
    FieldMask updateMask = FieldMask.newBuilder().addAllPaths(Arrays.asList(paths)).build();
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      UpdateTopicRequest request =
          UpdateTopicRequest.newBuilder().setUpdateMask(updateMask).setTopic(topic).build();
      // This operation is being handled synchronously.
      Topic response = managedKafkaClient.updateTopic(request);
      System.out.printf("Updated topic: %s\n", response.getName());
    } catch (IOException | ApiException 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 a Managed Service per Apache Kafka, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare le credenziali ADC per un ambiente di sviluppo locale.

from google.api_core.exceptions import NotFound
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"
# topic_id = "my-topic"
# partition_count = 20
# configs = {"min.insync.replicas": "1"}

client = managedkafka_v1.ManagedKafkaClient()

topic = managedkafka_v1.Topic()
topic.name = client.topic_path(project_id, region, cluster_id, topic_id)
topic.partition_count = partition_count
topic.configs = configs
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.extend(["partition_count", "configs"])

# For a list of editable fields, one can check https://cloud.google.com/managed-kafka/docs/create-topic#properties.
request = managedkafka_v1.UpdateTopicRequest(
    update_mask=update_mask,
    topic=topic,
)

try:
    response = client.update_topic(request=request)
    print("Updated topic:", response)
except NotFound as e:
    print(f"Failed to update topic {topic_id} with error: {e.message}")

Configura la conservazione dei messaggi

Kafka archivia i messaggi nei file di segmenti di log. Per impostazione predefinita, Kafka elimina i file di segmenti dopo un periodo di conservazione o quando una partizione supera una soglia di dimensioni dei dati. Puoi modificare questo comportamento abilitando la compattazione dei log. Se la compattazione dei log è abilitata, Kafka conserva solo l'ultimo valore per ogni chiave.

Google Cloud Managed Service per Apache Kafka utilizza l'archiviazione a livelli, il che significa che i segmenti di log completati vengono archiviati in remoto anziché nello spazio di archiviazione locale. Per saperne di più sull' archiviazione a livelli, consulta Archiviazione a livelli nella documentazione di Apache Kafka.

Imposta i valori di conservazione

Se la compattazione dei log non è abilitata, le seguenti impostazioni controllano la modalità di archiviazione dei file di segmenti di log da parte di Kafka:

  • retention.ms: la durata massima di salvataggio dei file di segmenti, in millisecondi.
  • retention.bytes: il numero massimo di byte da archiviare per partizione. Se i dati in una partizione superano questo valore, Kafka elimina i file di segmenti precedenti.

Per aggiornare queste impostazioni, utilizza gcloud CLI o Kafka CLI:

gcloud

Per impostare la conservazione dei messaggi, esegui il gcloud managed-kafka topics update comando.

gcloud managed-kafka topics update TOPIC_ID \
  --cluster=CLUSTER_ID \
  --location=LOCATION_ID \
  --configs=retention.ms=RETENTION_PERIOD,retention.bytes=MAX_BYTES

Sostituisci quanto segue:

  • TOPIC_ID: l'ID dell'argomento.
  • CLUSTER_ID: l'ID del cluster contenente l' argomento.
  • LOCATION_ID: la località del cluster.
  • RETENTION_PERIOD: la durata massima di archiviazione dei file di segmenti, in millisecondi.
  • MAX_BYTES: il numero massimo di byte da archiviare per partizione.

Kafka CLI

Prima di eseguire questo comando, installa gli strumenti da riga di comando di Kafka su una VM di Compute Engine. La VM deve essere in grado di raggiungere una subnet connessa al cluster Managed Service per Apache Kafka. Segui le istruzioni riportate in Produci e consuma messaggi con gli strumenti da riga di comando di Kafka.

Esegui il comando kafka-configs.sh:

kafka-configs.sh --alter \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --entity-type topics \
  --entity-name TOPIC_ID \
  --add-config retention.ms=RETENTION_PERIOD,retention.bytes=MAX_BYTES

Sostituisci quanto segue:

  • BOOTSTRAP_ADDRESS: l' indirizzo di bootstrap del cluster Managed Service per Apache Kafka.
  • TOPIC_ID: l'ID dell'argomento.
  • RETENTION_PERIOD: la durata massima di archiviazione dei file di segmenti, in millisecondi.
  • MAX_BYTES: il numero massimo di byte da archiviare per partizione.

Abilita la compattazione dei log

Se la compattazione dei log è abilitata, Kafka archivia solo l'ultimo messaggio per ogni chiave. La compattazione dei log è disabilitata per impostazione predefinita. Per abilitare la compattazione dei log per un argomento, imposta la configurazione cleanup.policy su "compact", come segue:

gcloud

Esegui il gcloud managed-kafka topics update comando.

gcloud managed-kafka topics update TOPIC_ID \
  --cluster=CLUSTER_ID \
  --location=LOCATION_ID \
  --configs=cleanup.policy=compact

Sostituisci quanto segue:

  • TOPIC_ID: l'ID dell'argomento.
  • CLUSTER_ID: l'ID del cluster contenente l' argomento.
  • LOCATION_ID: la località del cluster.

Kafka CLI

Prima di eseguire questo comando, installa gli strumenti da riga di comando di Kafka su una VM di Compute Engine. La VM deve essere in grado di raggiungere una subnet connessa al cluster Managed Service per Apache Kafka. Segui le istruzioni riportate in Produci e consuma messaggi con gli strumenti da riga di comando di Kafka.

Esegui il comando kafka-configs.sh:

kafka-configs.sh --alter \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --entity-type topics \
  --entity-name TOPIC_ID \
  --add-config cleanup.policy=compact

Sostituisci quanto segue:

  • BOOTSTRAP_ADDRESS: l' indirizzo di bootstrap del cluster Managed Service per Apache Kafka.
  • TOPIC_ID: l'ID dell'argomento.

Limitazioni

  • Non puoi sostituire le configurazioni degli argomenti per l'archiviazione remota, ad esempio remote.storage.enable.

  • Non puoi sostituire le configurazioni degli argomenti per i file di segmenti di log, ad esempio segment.bytes.

  • L'abilitazione della compattazione dei log per un argomento disabilita implicitamente l'archiviazione a livelli per quell'argomento. Tutti i file di log per l'argomento vengono archiviati localmente.

Passaggi successivi