Cómo ver un tema de Google Cloud Managed Service para Apache Kafka

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

Roles y permisos necesarios para ver un tema

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

Permisos necesarios

Se requieren los siguientes permisos para ver un tema:

• Enumerar temas: managedkafka.topics.list
• Obtener tema: managedkafka.topics.get

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

Para obtener más información sobre el rol de IAM Visualizador de Kafka administrado (roles/managedkafka.viewer), consulta Roles predefinidos de Managed Service para Apache Kafka.

Propiedades del tema en la consola

En la consola, puedes ver las siguientes propiedades del tema:

• Configuraciones: Esta pestaña proporciona detalles de configuración generales sobre el tema, incluidos los siguientes:

  • Nombre: Es el identificador único del tema dentro del clúster.

  • Particiones: Es la cantidad de particiones del tema. Las particiones dividen los datos del tema en segmentos para la escalabilidad y el paralelismo.

  • Réplicas: Es la cantidad de copias (réplicas) que se mantienen para cada partición para garantizar la redundancia y la disponibilidad de los datos.

  • Clúster: Es el nombre del clúster de Managed Service para Apache Kafka al que pertenece el tema.

  • Región: Es la Google Cloud región en la que se encuentran el clúster y el tema.

  • Parámetros del tema no predeterminados: Son las anulaciones de configuración a nivel del tema que se establecieron para el tema, diferentes de los valores predeterminados de todo el clúster.

• Supervisión: Esta pestaña proporciona gráficos visuales que muestran métricas clave relacionadas con la actividad y el rendimiento del tema. Estos gráficos incluyen lo siguiente:

  • Recuento de bytes: Es un gráfico de series temporales que muestra la tasa a la que se producen o envían bytes al tema. Esto indica el volumen de datos publicados en el tema a lo largo del tiempo. La métrica correspondiente es managedkafka.googleapis.com/byte_in_count.

  • Recuento de solicitudes: Es un gráfico de series temporales que representa la tasa de solicitudes realizadas al tema. Refleja la actividad y el uso generales del tema. La métrica relacionada es managedkafka.googleapis.com/topic_request_count.

  • Segmentos de registro por partición: Este gráfico muestra la cantidad de segmentos de registro activos para cada partición dentro del tema. Los segmentos de registro son los archivos físicos en el disco en los que Kafka almacena los datos del tema. La métrica pertinente es managedkafka.googleapis.com/log_segments.

• Grupos de consumidores: En esta sección, se enumeran los grupos de consumidores que están suscritos al tema. Un grupo de consumidores es un conjunto de consumidores que trabajan juntos para leer mensajes del tema.

Cómo ver un tema

Console

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

   Ir a los clústeres

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

2. Haz clic en el clúster para el que deseas ver los temas.

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

3. Para ver un tema específico, haz clic en el nombre del tema.

   Se muestra la página de detalles del tema.

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 describe comando:

   gcloud managed-kafka topics describe TOPIC_ID \
     --cluster=CLUSTER_ID --location=LOCATION_ID
   

   Este comando recupera y muestra detalles completos sobre el tema especificado. Esta información incluye su configuración, como la cantidad de particiones, el factor de replicación y cualquier anulación de 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.

El comando gcloud managed-kafka topics describe muestra información mínima sobre un tema, como el recuento de particiones y el factor de replicación. Para obtener información más detallada, incluidas las asignaciones de particiones y el conjunto completo de parámetros de configuración, usa la herramienta de línea de comandos de kafka-topics.sh.

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.

Para ver los detalles sobre un tema, ejecuta el comando kafka-topics.sh --describe:

kafka-topics.sh --describe \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --topic TOPIC_ID

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.

Este comando muestra un subconjunto de las propiedades del tema, incluidas las siguientes:

• Recuento de particiones
• Factor de replicación
• Asignaciones de particiones
• Configuración dinámica (parámetros que estableciste de forma explícita)
• Configuración estática (parámetros que se aplican cuando se inicia el clúster)

Para ver el conjunto completo de parámetros de configuración de un tema, incluidos los parámetros con valores predeterminados, ejecuta el comando kafka-configs.sh --describe:

kafka-configs.sh --describe \
--bootstrap-server=BOOTSTRAP_ADDRESS \
--command-config client.properties \
--entity-type topics \
--entity-name TOPIC_ID \
--all

El resultado es una lista de parámetros como pares clave-valor. La marca --all muestra todos los parámetros de configuración. Para obtener una lista de solo los parámetros de configuración dinámicos, omite la marca --all.

REST

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

• PROJECT_ID: El ID de tu Google Cloud proyecto
• LOCATION: La ubicación del clúster
• CLUSTER_ID: El ID del clúster
• TOPIC_ID: El ID del tema

Método HTTP y URL:

GET https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC_ID

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.

Ejecuta el siguiente comando:

curl -X GET \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/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.

Ejecuta el siguiente comando:

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

Invoke-WebRequest `
    -Method GET `
    -Headers $headers `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/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
}

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 getTopic(w io.Writer, projectID, region, clusterID, topicID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	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)
	req := &managedkafkapb.GetTopicRequest{
		Name: topicPath,
	}
	topic, err := client.GetTopic(ctx, req)
	if err != nil {
		return fmt.Errorf("client.GetTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Got 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 java.io.IOException;

public class GetTopic {

  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";
    getTopic(projectId, region, clusterId, topicId);
  }

  public static void getTopic(String projectId, String region, String clusterId, String topicId)
      throws Exception {
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      // This operation is being handled synchronously.
      Topic topic =
          managedKafkaClient.getTopic(TopicName.of(projectId, region, clusterId, topicId));
      System.out.println(topic.getAllFields());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.getTopic 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 Managed Service para Apache Kafka para Python.

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

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# topic_id = "my-topic"

client = managedkafka_v1.ManagedKafkaClient()

topic_path = client.topic_path(project_id, region, cluster_id, topic_id)
request = managedkafka_v1.GetTopicRequest(
    name=topic_path,
)

try:
    topic = client.get_topic(request=request)
    print("Got topic:", topic)
except NotFound as e:
    print(f"Failed to get topic {topic_id} with error: {e.message}")

Próximos pasos

• Borra un tema.

• Edita un tema.

• Enumera todos los grupos de consumidores.

• Visualiza un grupo de consumidores.