Afficher un sujet Google Cloud Managed Service pour Apache Kafka

Pour afficher des informations détaillées sur un sujet, vous pouvez utiliser la consoleGoogle Cloud , la Google Cloud CLI, la bibliothèque cliente, l'API Managed Kafka ou les API Apache Kafka Open Source.

Rôles et autorisations requis pour afficher un thème

Pour obtenir les autorisations nécessaires pour afficher un thème, demandez à votre administrateur de vous accorder le rôle IAM Lecteur Kafka géré (roles/managedkafka.viewer) sur votre projet. Pour en savoir plus sur l'attribution de rôles, consultez Gérer l'accès aux projets, aux dossiers et aux organisations.

Ce rôle prédéfini contient les autorisations requises pour afficher un thème. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

Autorisations requises

Les autorisations suivantes sont requises pour afficher un thème :

• Liste des thèmes : managedkafka.topics.list
• Obtenir un thème : managedkafka.topics.get

Vous pouvez également obtenir ces autorisations avec des rôles personnalisés ou d'autres rôles prédéfinis.

Pour en savoir plus sur le rôle IAM Lecteur Managed Kafka (roles/managedkafka.viewer), consultez Rôles prédéfinis Managed Service pour Apache Kafka.

Propriétés du sujet dans la console

Dans la console, vous pouvez afficher les propriétés de sujet suivantes :

• Configurations : cet onglet fournit des informations générales sur la configuration du thème, y compris les suivantes :

  • Nom : identifiant unique du thème dans le cluster.

  • Partitions : nombre de partitions dans le sujet. Les partitions divisent les données du sujet en segments pour l'évolutivité et le parallélisme.

  • Instances répliquées : nombre de copies (instances répliquées) conservées pour chaque partition afin d'assurer la redondance et la disponibilité des données.

  • Cluster : nom du cluster Managed Service pour Apache Kafka auquel appartient le sujet.

  • Région : région Google Cloud dans laquelle se trouvent le cluster et le thème.

  • Paramètres de sujet non définis par défaut : tous les paramètres de configuration au niveau du sujet qui ont été définis pour le sujet, différents des paramètres par défaut à l'échelle du cluster.

• Surveillance : cet onglet fournit des graphiques visuels qui affichent les métriques clés liées à l'activité et aux performances de la rubrique. Ces graphiques incluent les éléments suivants :

  • Nombre d'octets : graphique de séries temporelles indiquant le taux de production ou d'envoi d'octets au sujet. Indique le volume de données publiées dans le sujet au fil du temps. La métrique correspondante est managedkafka.googleapis.com/byte_in_count.

  • Nombre de requêtes : graphique de série temporelle représentant le taux de requêtes adressées à la rubrique. Elle reflète l'activité et l'utilisation globales du thème. La métrique associée est managedkafka.googleapis.com/topic_request_count.

  • Segments de journaux par partition : ce graphique affiche le nombre de segments de journaux actifs pour chaque partition du sujet. Les segments de journaux sont les fichiers physiques sur disque dans lesquels Kafka stocke les données des sujets. La métrique concernée est managedkafka.googleapis.com/log_segments.

• Groupes de consommateurs : cette section liste les groupes de consommateurs abonnés au sujet. Un groupe de consommateurs est un ensemble de consommateurs qui travaillent ensemble pour lire les messages du sujet.

Afficher un thème

Console

1. Dans la console Google Cloud , accédez à la page Clusters.

   accéder aux clusters

   Les clusters que vous avez créés dans un projet sont listés.

2. Cliquez sur le cluster dont vous souhaitez afficher les thèmes.

   La page des détails du cluster s'affiche. Sur la page de détails du cluster, les thèmes sont listés dans l'onglet Ressources.

3. Pour afficher un thème spécifique, cliquez sur son nom.

   La page des détails du thème s'affiche.

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. Exécutez la commande gcloud managed-kafka topics describe :

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

   Cette commande récupère et affiche des informations détaillées sur le thème spécifié. Ces informations incluent ses paramètres de configuration, tels que le nombre de partitions, le facteur de réplication et les éventuels remplacements de configuration au niveau du thème.

   Remplacez les éléments suivants :

   • TOPIC_ID : ID du sujet.
   • CLUSTER_ID : ID du cluster contenant le thème.
   • LOCATION_ID : emplacement du cluster.

La commande gcloud managed-kafka topics describe affiche des informations minimales sur un sujet, comme le nombre de partitions et le facteur de réplication. Pour obtenir des informations plus détaillées, y compris les attributions de partitions et l'ensemble complet des paramètres de configuration, utilisez l'outil de ligne de commande kafka-topics.sh.

CLI Kafka

Avant d'exécuter cette commande, installez les outils de ligne de commande Kafka sur une VM Compute Engine. La VM doit pouvoir accéder à un sous-réseau connecté à votre cluster Managed Service pour Apache Kafka. Suivez les instructions de la section Produire et consommer des messages avec les outils de ligne de commande Kafka.

Pour afficher des informations sur un sujet, exécutez la commande kafka-topics.sh --describe :

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

Remplacez les éléments suivants :

• BOOTSTRAP_ADDRESS : adresse d'amorçage du cluster Managed Service pour Apache Kafka.
• TOPIC_ID : ID du sujet.

Cette commande renvoie un sous-ensemble des propriétés du sujet, y compris les suivantes :

• Nombre de partitions
• Facteur de réplication
• Attributions de partitions
• Configuration dynamique (paramètres que vous avez définis de manière explicite)
• Configuration statique (paramètres appliqués au démarrage du cluster)

Pour afficher l'ensemble complet des paramètres de configuration d'un thème, y compris ceux avec des valeurs par défaut, exécutez la commande kafka-configs.sh --describe :

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

La sortie est une liste de paramètres sous forme de paires clé/valeur. L'option --all renvoie tous les paramètres de configuration. Pour obtenir la liste des paramètres de configuration dynamique uniquement, omettez l'indicateur --all.

REST

Avant d'utiliser les données de requête, effectuez les remplacements suivants :

• PROJECT_ID : ID de votre projet Google Cloud
• LOCATION : emplacement du cluster
• CLUSTER_ID : ID du cluster
• TOPIC_ID : ID du sujet

Méthode HTTP et URL :

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

Pour envoyer votre requête, développez l'une des options suivantes :

curl (Linux, macOS ou Cloud Shell)

Remarque : La commande suivante suppose que vous êtes connecté à la CLI gcloud avec votre compte utilisateur en exécutant la commande gcloud init ou gcloud auth login, ou en utilisant Cloud Shell, qui vous connecte automatiquement à la CLI gcloud. Vous pouvez exécuter la commande gcloud auth list pour vérifier quel est le compte actuellement actif.

Exécutez la commande suivante :

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)

Remarque : La commande suivante suppose que vous vous êtes connecté à la CLI gcloud avec votre compte utilisateur en exécutant la commande gcloud init ou gcloud auth login. Vous pouvez exécuter la commande gcloud auth list pour vérifier quel est le compte actuellement actif.

Exécutez la commande suivante :

$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

Vous devriez recevoir une réponse JSON de ce type :

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

Go

Avant d'essayer cet exemple, suivez les instructions de configuration pour Go dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Managed Service pour Apache Kafka en langage Go.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement 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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Java dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Java pour Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement 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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Python dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Python Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement 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}")

Étape suivante

• Supprimer un thème

• Modifier un thème

• Lister tous les groupes de consommateurs

• Afficher un groupe de consommateurs