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 consoleGoogle Cloud , a Google Cloud CLI, a biblioteca de cliente, a API Managed Kafka ou as APIs Apache Kafka de código aberto.

Papéis e permissões necessários para visualizar um tema

Para receber as permissões necessárias para ver um tópico, peça ao administrador para conceder a você o papel do IAM de Leitor do Kafka gerenciado (roles/managedkafka.viewer) no 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 necessárias, expanda a seção Permissões necessárias:

Permissões necessárias

As seguintes permissões são necessárias para acessar um tópico:

  • Listar tópicos: managedkafka.topics.list
  • Receber tema: 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 a função do IAM Leitor do Kafka gerenciado (roles/managedkafka.viewer), consulte Funções predefinidas 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 sobre a configuração do tema, incluindo:

    • 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 região Google Cloud em que o cluster e o tópico estão localizados.

    • Parâmetros de tópico não padrão: qualquer substituição de configuração no nível do tópico definida para o tópico, diferente dos padrões em todo o cluster.

  • Monitoramento: essa guia oferece gráficos visuais que mostram métricas importantes relacionadas à atividade e à performance do tema. 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 para o 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. Ela reflete a atividade e o uso geral do tema. 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: esta seção lista os grupos de consumidores inscritos no tópico. Um grupo de consumidores é um conjunto de consumidores que trabalham juntos para ler mensagens do tópico.

Conferir um assunto

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 para ver os temas.

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

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

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

gcloud

  1. No console do Google Cloud , ative o Cloud Shell.

    Ativar o Cloud Shell

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

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

    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 de configuração, como o número de partições, o fator de replicação e 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 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 conseguir 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 ver 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 definições de configuração de um tópico, incluindo definiçõ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 apenas de 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: o ID do projeto do Google Cloud
  • 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 esta amostra, 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 Go do serviço gerenciado para Apache Kafka.

Para autenticar o Serviço Gerenciado para Apache Kafka, configure o 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 esta amostra, 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 Java do serviço gerenciado para Apache Kafka.

Para autenticar o serviço gerenciado para Apache Kafka, configure o 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 esta amostra, 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 Python do serviço gerenciado para Apache Kafka.

Para autenticar o serviço gerenciado para Apache Kafka, configure o 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