Crea un tema de Google Cloud Managed Service para Apache Kafka

En Managed Service para Apache Kafka, los mensajes se organizan en temas. Un tema se compone de particiones. Una partición es una secuencia ordenada e inmutable de registros que pertenece a un solo agente dentro de un clúster de Kafka. Debes crear un tema para publicar o consumir mensajes.

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

Antes de comenzar

Primero debes crear un clúster antes de crear un tema. Asegúrate de haber configurado lo siguiente:

• Un clúster de Managed Service para Apache Kafka.

Roles y permisos necesarios para crear un tema

Para obtener los permisos que necesitas para crear un tema, pídele a tu administrador que te otorgue el rol de IAM Editor de temas de Kafka administrado (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 crear un tema. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para crear un tema:

• Crear un tema: managedkafka.topics.create

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

Propiedades de un tema de Managed Service para Apache Kafka

Cuando creas o actualizas un tema de Managed Service para Apache Kafka, debes especificar las siguientes propiedades.

Nombre del tema

Es el nombre del tema de Managed Service para Apache Kafka que estás creando. Si necesitas ayuda para asignarle un nombre a un tema, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka. El nombre de un tema es inmutable.

Recuento de particiones

La cantidad de particiones del tema. Puedes editar un tema para aumentar el recuento de particiones, pero no puedes disminuirlo. Aumentar la cantidad de particiones para un tema que usa una clave puede cambiar la forma en que se distribuyen los mensajes.

Factor de replicación

La cantidad de réplicas para cada partición. Si no especificas el valor, se usa el factor de replicación predeterminado del clúster.

Un factor de replicación más alto puede mejorar la coherencia de los datos en caso de fallas del agente, ya que los datos se replican en varios agentes. Para los entornos de producción, se recomienda un factor de replicación de 3 o más. Las cantidades más altas de réplicas aumentan el almacenamiento local y los costos de transferencia de datos para el tema. Sin embargo, no aumentan los costos de almacenamiento persistente. El factor de replicación no puede exceder la cantidad de agentes disponibles.

Nota: Los clústeres de Managed Service para Apache Kafka tienen los valores de configuración del agente como min.insync.replicas=2 y default.replication.factor=3. Si estableces un factor de replicación inferior a 3 para el tema, también debes establecer una anulación de configuración del tema como min.insync.replicas=1.

Otros parámetros

También puedes establecer otros parámetros de configuración de Apache Kafka a nivel del tema. Estos se especifican como pares key=value que anulan los valores predeterminados del clúster.

Las configuraciones relacionadas con los temas tienen un valor predeterminado del servidor y una anulación opcional por tema. El formato es una lista separada por comas de pares KEY=VALUE, donde KEY es el nombre de la propiedad de configuración del tema de Kafka y VALUE es la configuración requerida.Estos pares clave-valor te ayudan a anular los valores predeterminados del clúster. Algunos ejemplos son flush.ms=10 y compression.type=producer.

Precaución. Si anulas el período de retención predeterminado (retention.ms), evita configurarlo como demasiado corto (por ejemplo, menos de 48 horas). Un tiempo de retención más corto puede reducir los costos de almacenamiento, pero aumenta el riesgo de pérdida de datos si los consumidores se retrasan o si se producen interrupciones del servicio. Asegúrate de que el período de retención sea lo suficientemente largo como para cubrir posibles demoras en el procesamiento de nivel inferior o en las ventanas de resolución de incidentes.

Para obtener una lista de todos los parámetros de configuración admitidos a nivel del tema, consulta Parámetros de configuración a nivel del tema en la documentación de Apache Kafka.

Crea un tema

Antes de crear un tema, revisa las propiedades del tema.

Console

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

   Ir a los clústeres

2. Haz clic en el clúster para el que deseas crear un tema.

   Se abrirá la página Detalles del clúster.

3. En la página de detalles del clúster, haz clic en Crear tema.

   Se abrirá la página Crear tema de Kafka.

4. En Nombre del tema, ingresa una cadena.

5. En Recuento de particiones, ingresa la cantidad de particiones que deseas o conserva el valor predeterminado.

6. En Factor de replicación, ingresa el factor de replicación que deseas o conserva el valor predeterminado.

7. (Opcional) Para modificar cualquier configuración del tema, agrégala como pares clave-valor separados por comas en el campo Configuraciones.

8. Haz clic en Crear.

gcloud

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

   Activa Cloud Shell

   En la parte inferior de la Google Cloud consola de, 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 create comando:

   gcloud managed-kafka topics create TOPIC_ID \
       --cluster=CLUSTER_ID --location=LOCATION_ID \
       --partitions=PARTITIONS \
       --replication-factor=REPLICATION_FACTOR \
       --configs=CONFIGS
   

   Reemplaza lo siguiente:

   • TOPIC_ID: El nombre del tema.
   • CLUSTER: El nombre del clúster en el que deseas crear el tema.
   • LOCATION: La región del clúster.
   • PARTITIONS: La cantidad de particiones del tema.
   • REPLICATION_FACTOR: El factor de replicación para el tema.
   • CONFIGS: Parámetros opcionales a nivel del tema. Especifica como pares clave-valor separados por comas. Por ejemplo, compression.type=producer.

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-topics.sh de la siguiente manera:

kafka-topics.sh --create --if-not-exists \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --topic TOPIC_ID \
  --partitions PARTITIONS \
  --replication-factor REPLICATION_FACTOR

Reemplaza lo siguiente:

• BOOTSTRAP_ADDRESS: La dirección de arranque del clúster de Managed Service para Apache Kafka.

• TOPIC_ID: El nombre del tema.

• PARTITIONS: La cantidad de particiones del tema.

• REPLICATION_FACTOR: El factor de replicación del tema.

REST

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

• PROJECT_ID: tu Google Cloud ID de proyecto
• LOCATION: la ubicación del clúster
• CLUSTER_ID: el ID del clúster
• TOPIC_ID: El ID del tema
• PARTITION_COUNT: la cantidad de particiones del tema
• REPLICATION_FACTOR: la cantidad de réplicas de cada partición

Método HTTP y URL:

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

Cuerpo JSON de la solicitud:

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

Para enviar tu solicitud, expande una de estas opciones:

curl (Linux, macOS o Cloud Shell)

Nota: Con el siguiente comando, se supone que accediste a gcloud CLI con tu cuenta de usuario a través de la ejecución de gcloud init o gcloud auth login , o a través del uso de Cloud Shell, que accede de forma automática a gcloud CLI . Para comprobar la cuenta activa actual, ejecuta gcloud auth list.

Guarda el cuerpo de la solicitud en un archivo llamado request.json, y ejecuta el siguiente comando:

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/CLUSTER_ID/topics?topicId=TOPIC_ID"

PowerShell (Windows)

Nota: El siguiente comando supone que accediste a gcloud CLI con tu cuenta de usuario a través de la ejecución de gcloud init o gcloud auth login . Para comprobar la cuenta activa actual, ejecuta gcloud auth list.

Guarda el cuerpo de la solicitud en un archivo llamado request.json, y ejecuta el siguiente comando:

$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/CLUSTER_ID/topics?topicId=TOPIC_ID" | Select-Object -Expand Content

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

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

Terraform

Puedes usar un recurso de Terraform para crear un tema.

resource "google_managed_kafka_topic" "default" {
  project            = data.google_project.default.project_id # Replace this with your project ID in quotes
  topic_id           = "my-topic-id"
  cluster            = google_managed_kafka_cluster.default.cluster_id
  location           = "us-central1"
  partition_count    = 2
  replication_factor = 3
}

Si deseas obtener más información para aplicar o quitar una configuración de Terraform, consulta los comandos básicos de Terraform.

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"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func createTopic(w io.Writer, projectID, region, clusterID, topicID string, partitionCount, replicationFactor int32, configs map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	// partitionCount := 10
	// replicationFactor := 3
	// 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,
		ReplicationFactor: replicationFactor,
		Configs:           configs,
	}

	req := &managedkafkapb.CreateTopicRequest{
		Parent:  clusterPath,
		TopicId: topicID,
		Topic:   topicConfig,
	}
	topic, err := client.CreateTopic(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Created topic: %s\n", topic.Name)
	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.ClusterName;
import com.google.cloud.managedkafka.v1.CreateTopicRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.Topic;
import com.google.cloud.managedkafka.v1.TopicName;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreateTopic {

  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 = 100;
    int replicationFactor = 3;
    Map<String, String> configs =
        new HashMap<String, String>() {
          {
            put("min.insync.replicas", "2");
          }
        };
    createTopic(projectId, region, clusterId, topicId, partitionCount, replicationFactor, configs);
  }

  public static void createTopic(
      String projectId,
      String region,
      String clusterId,
      String topicId,
      int partitionCount,
      int replicationFactor,
      Map<String, String> configs)
      throws Exception {
    Topic topic =
        Topic.newBuilder()
            .setName(TopicName.of(projectId, region, clusterId, topicId).toString())
            .setPartitionCount(partitionCount)
            .setReplicationFactor(replicationFactor)
            .putAllConfigs(configs)
            .build();
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      CreateTopicRequest request =
          CreateTopicRequest.newBuilder()
              .setParent(ClusterName.of(projectId, region, clusterId).toString())
              .setTopicId(topicId)
              .setTopic(topic)
              .build();
      // This operation is being handled synchronously.
      Topic response = managedKafkaClient.createTopic(request);
      System.out.printf("Created topic: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.createTopic 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 AlreadyExists
from google.cloud import managedkafka_v1

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# topic_id = "my-topic"
# partition_count = 10
# replication_factor = 3
# 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.replication_factor = replication_factor
# For a list of configs, one can check https://kafka.apache.org/documentation/#topicconfigs
topic.configs = configs

request = managedkafka_v1.CreateTopicRequest(
    parent=client.cluster_path(project_id, region, cluster_id),
    topic_id=topic_id,
    topic=topic,
)

try:
    response = client.create_topic(request=request)
    print("Created topic:", response.name)
except AlreadyExists as e:
    print(f"Failed to create topic {topic.name} with error: {e.message}")

Próximos pasos

• Supervisa tu tema

• Haz una lista de tus temas

• Visualiza un tema

• Edita un tema

• Borra un tema