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 consola deGoogle Cloud , la CLI de Google Cloud, la biblioteca cliente, la API de Kafka administrado o las APIs de Apache Kafka de código abierto.

Roles y permisos obligatorios para ver un tema

Para 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:

• Temas de la lista: managedkafka.topics.list
• Obtén el tema: managedkafka.topics.get

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

Para obtener más información sobre el rol de IAM de 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: En esta pestaña, se proporcionan detalles generales de configuración sobre el tema, incluidos los siguientes:

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

  • Particiones: Es la cantidad de particiones en el 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 región Google Cloud en la que se encuentran el clúster y el tema.

  • Parámetros de temas no predeterminados: Son los reemplazos de configuración a nivel del tema que se establecieron para el tema y que son diferentes de los valores predeterminados en todo el clúster.

• Supervisión: En esta pestaña, se proporcionan 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 velocidad a la que se producen o envían bytes al tema. 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.

  • Registra segmentos por partición: En este gráfico, se 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 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 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 su nombre.

   Se mostrará la página de detalles del tema.

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

   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 la 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 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.

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: La dirección de arranque del clúster de Managed Service para Apache Kafka.
• TOPIC_ID: Es el ID del tema.

Este comando devuelve 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 de configuración que estableciste de forma explícita)
• Configuración estática (parámetros de configuración 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 de configuración como pares clave-valor. La marca --all devuelve todos los parámetros de configuración. Para obtener una lista solo de 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 proyecto de Google Cloud
• LOCATION: Es la ubicación del clúster.
• CLUSTER_ID: 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 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"

	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 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 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. 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

# 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

• Cómo borrar un tema

• Cómo editar un tema

• Enumera todos los grupos de consumidores

• Cómo ver un grupo de consumidores