Actualiza un tema de Google Cloud Managed Service para Apache Kafka

Después de crear un tema, puedes editar su configuración para actualizar estas propiedades: la cantidad de particiones y las configuraciones del tema que no tienen como valor predeterminado las propiedades ya establecidas a nivel del clúster. Solo puedes aumentar la cantidad de particiones, no puedes disminuirla.

Para actualizar un solo tema, puedes usar la Google Cloud consola, la Google Cloud CLI, la biblioteca cliente, la API de Managed Kafka o las APIs de Apache Kafka de código abierto.

Roles y permisos necesarios para editar un tema

Si quieres obtener los permisos que necesitas para editar un tema, pídele a tu administrador que te otorgue el rol de IAM Editor de temas de Managed Kafka(roles/managedkafka.topicEditor) en el proyecto. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para editar un tema. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para editar un tema:

  • Actualizar un tema: managedkafka.topics.update

También puedes obtener estos permisos con roles personalizados o otros roles predefinidos.

Edita un tema

Para editar un tema, sigue estos pasos:

Console

  1. En la Google Cloud consola de Google Cloud, ve a la página Clústeres.

    Ir a los clústeres

    Se muestran los clústeres que creaste en un proyecto.

  2. Haz clic en el clúster al que pertenece el tema que quieres editar.

    Se abrirá la página Detalles del clúster. En la página de detalles del clúster, en la pestaña Recursos, se muestran los temas.

  3. Haz clic en el tema que quieres editar.

    Se abrirá la página Detalles del tema.

  4. Para realizar las ediciones, haz clic en Editar.

  5. Haz clic en Guardar después de los cambios.

gcloud

  1. En la Google Cloud consola de Google Cloud, activa Cloud Shell.

    Activa Cloud Shell

    En la parte inferior de la Google Cloud consola de Google Cloud, se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.

  2. Ejecuta el gcloud managed-kafka topics update comando:

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

    Este comando modifica la configuración de un tema existente en el clúster de Managed Service para Apache Kafka especificado. Puedes usar este comando para aumentar la cantidad de particiones y actualizar la configuración a nivel del tema.

    Reemplaza lo siguiente:

    • TOPIC_ID: Es el ID del tema.
    • CLUSTER_ID: Es el ID del clúster que contiene el tema.
    • LOCATION_ID: Es la ubicación del clúster.
    • PARTITIONS: Es la cantidad actualizada de particiones para el tema (opcional). Solo puedes aumentar la cantidad de particiones, no puedes disminuirla.
    • CONFIGS: Es una lista de parámetros de configuración para actualizar (opcional). Especifica como una lista separada por comas de pares clave-valor. Por ejemplo, retention.ms=3600000,retention.bytes=10000000.

REST

Antes de usar cualquiera de los datos de solicitud a continuación, realiza los siguientes reemplazos:

  • PROJECT_ID: tu Google Cloud ID del proyecto
  • LOCATION: la ubicación del clúster
  • CLUSTER_ID: el ID del clúster
  • TOPIC_ID: El ID del tema
  • UPDATE_MASK: qué campos actualizar, como una lista separada por comas de nombres completamente calificados Ejemplo: partitionCount
  • PARTITION_COUNT: la cantidad actualizada de particiones para el tema

Método HTTP y URL: 

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

Cuerpo JSON de la solicitud: 

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

Para enviar tu solicitud, expande una de estas opciones:

Deberías recibir una respuesta JSON similar a la que se muestra a continuación:

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

Go

Antes de probar este ejemplo, sigue las instrucciones de configuración de Go en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Managed Service para Apache Kafka para Go.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación(ADC). Para obtener más información, consulta Configura ADC para un entorno de desarrollo 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 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

Antes de probar este ejemplo, sigue las instrucciones de configuración de Java en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Managed Service para Apache Kafka para Java.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.

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

Antes de probar este ejemplo, sigue las instrucciones de configuración de Python en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Python de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.

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 retención de mensajes

Kafka almacena mensajes en archivos de segmentos de registro. De forma predeterminada, Kafka borra los archivos de segmentos después de un período de retención o cuando una partición supera un umbral de tamaño de datos. Para cambiar este comportamiento, habilita la compactación de registros. Si la compactación de registros está habilitada, Kafka solo conserva el valor más reciente de cada clave.

Google Cloud Managed Service para Apache Kafka usa almacenamiento por niveles, lo que significa que los segmentos de registro completados se almacenan de forma remota, en lugar del almacenamiento local. Para obtener más información sobre el almacenamiento por niveles, consulta Almacenamiento por niveles en la documentación de Apache Kafka.

Establece los valores de retención

Si la compactación de registros no está habilitada, la siguiente configuración controla cómo Kafka almacena los archivos de segmentos de registro:

  • retention.ms: Es la duración máxima para guardar archivos de segmentos, en milisegundos.
  • retention.bytes: Es la cantidad máxima de bytes que se almacenarán por partición. Si los datos de una partición superan este valor, Kafka descarta los archivos de segmentos más antiguos.

Para actualizar esta configuración, usa gcloud CLI o la CLI de Kafka:

gcloud

Para establecer la retención de mensajes, ejecuta el 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

Reemplaza lo siguiente:

  • TOPIC_ID: Es el ID del tema.
  • CLUSTER_ID: Es el ID del clúster que contiene el tema.
  • LOCATION_ID: Es la ubicación del clúster.
  • RETENTION_PERIOD: Es la cantidad máxima de tiempo para almacenar archivos de segmentos, en milisegundos.
  • MAX_BYTES: Es la cantidad máxima de bytes que se almacenarán por partición.

CLI de Kafka

Antes de ejecutar este comando, instala las herramientas de línea de comandos de Kafka en una VM de Compute Engine. La VM debe poder comunicarse con una subred conectada a tu clúster de Managed Service para Apache Kafka. Sigue las instrucciones en Produce y consume mensajes con las herramientas de línea de comandos de Kafka.

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

Reemplaza lo siguiente:

  • BOOTSTRAP_ADDRESS: Es la dirección de arranque del clúster de Managed Service para Apache Kafka.
  • TOPIC_ID: Es el ID del tema.
  • RETENTION_PERIOD: Es la cantidad máxima de tiempo para almacenar archivos de segmentos, en milisegundos.
  • MAX_BYTES: Es la cantidad máxima de bytes que se almacenarán por partición.

Habilita la compactación de registros

Si la compactación de registros está habilitada, Kafka solo almacena el mensaje más reciente de cada clave. La compactación de registros está inhabilitada de forma predeterminada. Para habilitar la compactación de registros para un tema, establece la configuración de cleanup.policy en "compact", de la siguiente manera:

gcloud

Ejecuta el gcloud managed-kafka topics update comando.

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

Reemplaza lo siguiente:

  • TOPIC_ID: Es el ID del tema.
  • CLUSTER_ID: Es el ID del clúster que contiene el tema.
  • LOCATION_ID: Es la ubicación del clúster.

CLI de Kafka

Antes de ejecutar este comando, instala las herramientas de línea de comandos de Kafka en una VM de Compute Engine. La VM debe poder comunicarse con una subred conectada a tu clúster de Managed Service para Apache Kafka. Sigue las instrucciones en Produce y consume mensajes con las herramientas de línea de comandos de Kafka.

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

Reemplaza lo siguiente:

  • BOOTSTRAP_ADDRESS: Es la dirección de arranque del clúster de Managed Service para Apache Kafka.
  • TOPIC_ID: Es el ID del tema.

Limitaciones

  • No puedes anular las configuraciones de temas para el almacenamiento remoto, como remote.storage.enable.

  • No puedes anular las configuraciones de temas para los archivos de segmentos de registro, como segment.bytes.

  • Si habilitas la compactación de registros para un tema, se inhabilita implícitamente el almacenamiento por niveles para ese tema. Todos los archivos de registro del tema se almacenan de forma local.

Próximos pasos