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 se establecen de forma predeterminada en las propiedades ya establecidas a nivel del clúster. Solo puedes aumentar la cantidad de particiones, no disminuirla.

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

Roles y permisos obligatorios para editar un tema

Para 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 Kafka administrados(roles/managedkafka.topicEditor) en tu 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:

  • Actualiza un tema: managedkafka.topics.update

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

Para obtener más información sobre este rol, consulta Roles predefinidos de Managed Service para Apache Kafka.

Cómo editar un tema

Para editar un tema, sigue estos pasos:

Console

  1. En la 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 enumeran los temas.

  3. Haz clic en el tema que deseas editar.

    Se abrirá la página Detalles del tema.

  4. Para realizar cambios, haz clic en Editar.

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

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. Ejecuta el comando gcloud managed-kafka topics update:

    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: Opcional: Es la cantidad actualizada de particiones del tema. Solo puedes aumentar la cantidad de particiones, no disminuirla.
    • CONFIGS: Opcional: Es una lista de parámetros de configuración que se actualizarán. Se debe especificar como una lista de pares clave-valor separados por comas. Por ejemplo, retention.ms=3600000,retention.bytes=10000000

    3. REST

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

    • PROJECT_ID: El ID de tu proyecto de Google Cloud
    • LOCATION: Es la ubicación del clúster.
    • CLUSTER_ID: ID del clúster
    • TOPIC_ID: El ID del tema
    • UPDATE_MASK: Indica qué campos se actualizarán, como una lista separada por comas de nombres completamente calificados. Ejemplo: partitionCount
    • PARTITION_COUNT: Es la cantidad actualizada de particiones del 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 en Go.

    Para autenticarte en Managed Service for 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 Java de Managed Service for Apache Kafka.

    Para autenticarte en el servicio administrado 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. Si deseas obtener más información, consulta la documentación de referencia de la API de Python de Managed Service for Apache Kafka.

    Para autenticarte en el servicio administrado 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 los mensajes en archivos de segmentos de registro. De forma predeterminada, Kafka borra los archivos de segmento después de un período de retención o cuando una partición supera un umbral de tamaño de datos. Puedes cambiar este comportamiento si habilitas 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 de en el 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 registros:

  • 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 pueden almacenar por partición. Si los datos de una partición superan este valor, Kafka descarta los archivos de segmento más antiguos.

Para actualizar estos parámetros de configuración, usa gcloud CLI o Kafka CLI:

gcloud

Para establecer la retención de mensajes, ejecuta el comando gcloud managed-kafka topics update.

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 pueden almacenar 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 acceder a una subred conectada a tu clúster de Managed Service para Apache Kafka. Sigue las instrucciones en Produce and consume messages with the Kafka command-line tools.

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: 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 pueden almacenar 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 para cada clave. La compactación de registros está inhabilitada de forma predeterminada. Para habilitar la compactación de registros en un tema, establece la configuración cleanup.policy en "compact", de la siguiente manera:

gcloud

Ejecuta el comando gcloud managed-kafka topics update:

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 acceder a una subred conectada a tu clúster de Managed Service para Apache Kafka. Sigue las instrucciones en Produce and consume messages with the Kafka command-line tools.

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: 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 la configuración de temas para el almacenamiento remoto, como remote.storage.enable.

  • No puedes anular la configuración 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