Ver um tópico do serviço gerenciado do Google Cloud para Apache Kafka

Para conferir informações detalhadas sobre um único tópico, use o Google Cloud console, a Google Cloud CLI, a biblioteca de cliente, a API Managed Kafka ou as APIs de código aberto do Apache Kafka.

Papéis e permissões necessários para visualizar um tópico

Para receber as permissões necessárias para visualizar um tópico, peça ao administrador para conceder a você o papel do IAM de Leitor do Managed Kafka (roles/managedkafka.viewer) no seu projeto. Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Esse papel predefinido contém as permissões necessárias para visualizar um tópico. Para acessar as permissões exatas que são necessárias, expanda a seção Permissões necessárias:

Permissões necessárias

As permissões a seguir são necessárias para visualizar um tópico:

  • Listar tópicos: managedkafka.topics.list
  • Receber tópico: managedkafka.topics.get

Essas permissões também podem ser concedidas com funções personalizadas ou outros papéis predefinidos.

Para mais informações sobre o papel do IAM de Leitor do Managed Kafka (roles/managedkafka.viewer), consulte Papéis predefinidos do Serviço Gerenciado para Apache Kafka.

Propriedades do tópico no console

No console, é possível conferir as seguintes propriedades do tópico:

  • Configurações: essa guia fornece detalhes gerais de configuração sobre o tópico, incluindo o seguinte:

    • Nome: o identificador exclusivo do tópico no cluster.

    • Partições: o número de partições no tópico. As partições dividem os dados do tópico em segmentos para escalonabilidade e paralelismo.

    • Réplicas: o número de cópias (réplicas) mantidas para cada partição para garantir a redundância e a disponibilidade dos dados.

    • Cluster: o nome do cluster do Serviço Gerenciado para Apache Kafka a que o tópico pertence.

    • Região: A Google Cloud região em que o cluster e o tópico estão localizados.

    • Parâmetros de tópico não padrão: todas as substituições de configuração no nível do tópico que foram definidas para o tópico, diferentes dos padrões de todo o cluster.

  • Monitoramento: essa guia fornece gráficos visuais que mostram as principais métricas relacionadas à atividade e ao desempenho do tópico. Esses gráficos incluem o seguinte:

    • Contagem de bytes: um gráfico de série temporal que mostra a taxa em que os bytes são produzidos ou enviados ao tópico. Isso indica o volume de dados publicados no tópico ao longo do tempo. A métrica correspondente é managedkafka.googleapis.com/byte_in_count.

    • Contagem de solicitações: um gráfico de série temporal que representa a taxa de solicitações feitas ao tópico. Ele reflete a atividade e o uso geral do tópico. A métrica relacionada é managedkafka.googleapis.com/topic_request_count.

    • Segmentos de registro por partição: esse gráfico mostra o número de segmentos de registro ativos para cada partição no tópico. Os segmentos de registro são os arquivos físicos no disco em que o Kafka armazena os dados do tópico. A métrica relevante é managedkafka.googleapis.com/log_segments.

  • Grupos de consumidores: essa seção lista os grupos de consumidores que estão inscritos no tópico. Um grupo de consumidores é um conjunto de consumidores que trabalham juntos para ler mensagens do tópico.

Visualizar um tópico

Console

  1. No Google Cloud console, acesse a página Clusters.

    Acessar Clusters

    Os clusters criados em um projeto são listados.

  2. Clique no cluster em que você quer conferir os tópicos.

    A página de detalhes do cluster é exibida. Na página de detalhes do cluster, na guia Recursos, os tópicos são listados.

  3. Para visualizar um tópico específico, clique no nome dele.

    A página de detalhes do tópico é exibida.

gcloud

  1. Noconsol, ative o Cloud Shell. Google Cloud

    Ativar o Cloud Shell

    Na parte de baixo do Google Cloud consol, uma sessão do Cloud Shell é iniciada e exibe um prompt de linha de comando. O Cloud Shell é um ambiente shell com a Google Cloud CLI já instalada e com valores já definidos para o projeto atual. A inicialização da sessão pode levar alguns segundos.

  2. Execute o gcloud managed-kafka topics describe comando:

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

    Esse comando busca e mostra detalhes abrangentes sobre o tópico especificado. Essas informações incluem as configurações, como o número de partições, o fator de replicação e todas as substituições de configuração no nível do tópico.

    Substitua:

    • TOPIC_ID: o ID do tópico.
    • CLUSTER_ID: o ID do cluster que contém o tópico.
    • LOCATION_ID: o local do cluster.

O comando gcloud managed-kafka topics describe mostra informações mínimas sobre um tópico, como a contagem de partições e o fator de replicação. Para receber informações mais detalhadas, incluindo atribuições de partição e o conjunto completo de configurações, use a ferramenta de linha de comando kafka-topics.sh.

CLI do Kafka

Antes de executar esse comando, instale as ferramentas de linha de comando do Kafka em uma VM do Compute Engine. A VM precisa acessar uma sub-rede conectada ao cluster do Serviço Gerenciado para Apache Kafka. Siga as instruções em Produzir e consumir mensagens com as ferramentas de linha de comando do Kafka.

Para conferir detalhes sobre um tópico, execute o comando kafka-topics.sh --describe:

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

Substitua:

Esse comando retorna um subconjunto das propriedades do tópico, incluindo o seguinte:

  • Contagem de partições
  • Fator de replicação
  • Atribuições de partição
  • Configuração dinâmica (configurações definidas explicitamente)
  • Configuração estática (configurações aplicadas quando o cluster é iniciado)

Para conferir o conjunto completo de configurações de um tópico, incluindo configurações com valores padrão, execute o 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

A saída é uma lista de configurações como pares de chave-valor. A flag --all retorna todas as configurações. Para receber uma lista de apenas configurações dinâmicas, omita a flag --all.

REST

Antes de usar os dados da solicitação abaixo, faça as substituições a seguir:

  • PROJECT_ID: seu Google Cloud ID do projeto
  • LOCATION: o local do cluster
  • CLUSTER_ID: o ID do cluster
  • TOPIC_ID: o ID do tópico

Método HTTP e URL: 

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

Para enviar a solicitação, expanda uma destas opções:

Você receberá uma resposta JSON semelhante a esta:

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

Go

Antes de testar este exemplo, siga as instruções de configuração do Go em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API do Serviço Gerenciado para Apache Kafka para Go.

Para autenticar no Serviço Gerenciado para Apache Kafka, configure as Application Default Credentials(ADC). Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento 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 testar este exemplo, siga as instruções de configuração do Java em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API do Serviço Gerenciado para Apache Kafka para Java.

Para autenticar no Serviço Gerenciado para Apache Kafka, configure as Application Default Credentials. Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento 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 testar este exemplo, siga as instruções de configuração do Python em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Serviço Gerenciado para Apache Kafka Python.

Para autenticar no Serviço Gerenciado para Apache Kafka, configure as Application Default Credentials. Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento 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}")

A seguir