Cette rubrique explique comment créer un connecteur de réception Tablestore afin d'exporter les données d'une rubrique d'une instance ApsaraMQ for Kafka vers Tablestore.
Prérequis
-
ApsaraMQ for Kafka
Vous avez activé la fonctionnalité de connecteur pour l'instance. Pour plus d'informations, consultez la page Activer les connecteurs.
Vous avez créé une rubrique source pour l'instance. Pour plus d'informations, consultez la section Étape 1 : Créer une rubrique.
-
Tablestore
Vous avez activé Tablestore et créé une instance. Pour plus d'informations, consultez la page Activer Tablestore et créer une instance.
Notes
Vous ne pouvez exporter les données d'une rubrique source d'une instance ApsaraMQ for Kafka vers Tablestore que si les deux services se trouvent dans la même région. Pour connaître les limites relatives aux connecteurs, consultez la section Limites d'utilisation.
-
Lors de la création d'un connecteur, ApsaraMQ for Kafka crée automatiquement un rôle lié au service.
Si aucun rôle lié au service n'existe, ApsaraMQ for Kafka en crée un automatiquement, ce qui permet à ApsaraMQ for Kafka d'exporter les données vers Tablestore.
Si un rôle lié au service existe déjà, ApsaraMQ for Kafka n'en crée pas de nouveau.
Pour plus d'informations sur les rôles liés au service, consultez la page Rôles liés au service.
Procédure
Cette rubrique décrit la procédure d'utilisation d'un connecteur de réception Tablestore pour exporter les données d'une rubrique source d'une instance ApsaraMQ for Kafka vers Tablestore.
-
Facultatif : Créez les rubriques et le group requis pour le connecteur de réception Tablestore.
Si vous n'avez pas besoin de personnaliser les rubriques et le group, ignorez cette étape et sélectionnez Auto à l'étape suivante.
ImportantCertaines rubriques requises par le connecteur de réception Tablestore doivent utiliser le moteur de stockage Local. Pour les instances ApsaraMQ for Kafka de version majeure 0.10.2, vous ne pouvez pas créer manuellement des rubriques utilisant le moteur de stockage Local. Ces rubriques ne peuvent être créées qu'automatiquement.
-
Vérifiez le résultat
Créer des rubriques pour le connecteur de réception Tablestore
Dans la console ApsaraMQ for Kafka, créez manuellement les cinq rubriques requises pour le connecteur de réception Tablestore : une rubrique d'offset de tâche, une rubrique de configuration de tâche, une rubrique d'état de tâche, une rubrique de file d'attente de lettres mortes et une rubrique de données d'erreur. Le nombre de partitions requis et le moteur de stockage varient selon chaque rubrique. Pour plus d'informations, consultez le tableau Liste des paramètres du service source.
Connectez-vous à la console ApsaraMQ for Kafka.
-
Sur la page Overview, sélectionnez une région dans la section Resource Distribution.
ImportantVous devez créer les rubriques dans la même région que votre application, c'est-à-dire la région où l'instance ECS est déployée. Les rubriques ne peuvent pas être utilisées entre différentes régions. Par exemple, si une rubrique est créée dans la région Chine (Pékin), le producteur et le consommateur de messages doivent également s'exécuter sur une instance ECS dans la région Chine (Pékin).
Sur la page Instances, cliquez sur le nom de l'instance cible.
Dans le volet de navigation de gauche, cliquez sur Topics.
Sur la page Topics, cliquez sur Create Topic.
-
Dans le panneau Create Topic, définissez les propriétés de la rubrique, puis cliquez sur OK.
Paramètre
Description
Exemple
Name
Nom de la rubrique.
RemarqueDans Kafka, les noms de rubrique tels que
xxx_xxxetxxx.xxxsont considérés comme identiques. Une erreur est signalée si vous tentez de créer une rubrique avec un nom dupliqué.demo
Description
Brève description de la rubrique.
test demo
Partitions
Nombre de partitions dans la rubrique.
12
Storage Engine
RemarqueActuellement, vous pouvez sélectionner un type de moteur de stockage uniquement pour les instances non Serverless de l'édition Professionnelle. Pour les autres instances, ce paramètre n'est pas pris en charge et est défini sur Cloud Storage par défaut.
Moteur de stockage pour les messages de la rubrique.
ApsaraMQ for Kafka prend en charge les deux moteurs de stockage suivants.
Cloud Storage : La couche sous-jacente utilise des disques Alibaba Cloud. Ce moteur de stockage offre une faible latence, des performances élevées, une durabilité élevée et une fiabilité élevée. Il utilise un mécanisme distribué à trois réplicas. Si l'Instance Edition de l'instance est Standard (High Write), le moteur de stockage ne peut être que Cloud Storage.
Local Storage : Utilise l'algorithme de réplication ISR (In-Sync Replica) natif de Kafka et un mécanisme distribué à trois réplicas.
Cloud Storage
Message Type
Type de messages de la rubrique.
Normal Message : Par défaut, les messages ayant la même clé sont distribués vers la même partition, et les messages au sein d'une partition sont stockés dans l'ordre de leur envoi. Si une machine du cluster tombe en panne, les messages peuvent devenir désordonnés. Lorsque vous définissez le Storage Engine sur Cloud Storage, l'option Normal Message est sélectionnée par défaut.
Partitionally Ordered Message : Par défaut, les messages ayant la même clé sont distribués vers la même partition, et les messages au sein d'une partition sont stockés dans l'ordre de leur envoi. Si une machine du cluster tombe en panne, l'ordre des messages est toujours garanti au sein de la partition. Toutefois, l'envoi de messages vers certaines partitions peut échouer. Les opérations normales reprennent après la récupération des partitions. Lorsque vous définissez le Storage Engine sur Local Storage, l'option Partitionally Ordered Message est sélectionnée par défaut.
Normal Message
Log Cleanup Policy
Politique de nettoyage des journaux de la rubrique.
Lorsque vous définissez le Storage Engine sur Local Storage, vous devez configurer la Log Cleanup Policy. Vous pouvez définir le moteur de stockage sur Local Storage uniquement pour les instances de l'édition Professionnelle. Les instances de l'édition Standard ne sont pas prises en charge.
ApsaraMQ for Kafka prend en charge les deux politiques de nettoyage suivantes.
Delete : Politique de nettoyage des messages par défaut. Si la capacité du disque est suffisante, les messages sont conservés pendant la durée de rétention maximale. Si la capacité du disque est insuffisante (généralement lorsque l'utilisation du disque dépasse 85 %), les anciens messages sont supprimés prématurément pour garantir la disponibilité du service.
Compact : Utilise la politique de nettoyage Kafka Log Compaction. La politique Log Compaction garantit que pour les messages ayant la même clé, la dernière valeur est toujours conservée. Cela s'applique principalement aux scénarios tels que la récupération de l'état après une panne du système ou le rechargement du cache après un redémarrage du système. Par exemple, lors de l'utilisation de Kafka Connect ou de Confluent Schema Registry, vous devez utiliser une rubrique Kafka Compact pour stocker l'état du système ou les informations de configuration.
ImportantLes rubriques Compact sont généralement utilisées uniquement dans certains composants de l'écosystème, tels que Kafka Connect ou Confluent Schema Registry. Ne définissez pas cette propriété pour les rubriques dans d'autres scénarios d'envoi et de réception de messages. Pour plus d'informations, consultez la bibliothèque de démonstration ApsaraMQ for Kafka.
Compact
Tag
Tags de la rubrique.
demo
Une fois la rubrique créée, elle s'affiche dans la liste sur la page Topics.
Créer un group pour le connecteur de réception Tablestore
Vous pouvez créer manuellement un group pour la tâche de synchronisation des données du connecteur de réception Tablestore dans la console ApsaraMQ for Kafka. Le nom de ce group doit être connect-Nom de la tâche. Pour plus d'informations, consultez la section Paramètres du service source.
Connectez-vous à la console ApsaraMQ for Kafka.
Sur la page Overview, sélectionnez une région dans la section Resource Distribution.
Sur la page Instances, cliquez sur le nom de l'instance cible.
Dans le volet de navigation de gauche, cliquez sur Groups.
Sur la page Groups, cliquez sur Create Group.
-
Dans le panneau Create Group, saisissez un nom pour le group dans la zone de texte Group ID, saisissez une brève description dans la zone de texte Description, ajoutez des tags au group, puis cliquez sur OK.
Une fois le group créé, il s'affiche dans la liste sur la page Groups.
Créer et déployer un connecteur de sortie Tablestore
Créez et déployez un connecteur de sortie Tablestore pour synchroniser les données d'une instance ApsaraMQ for Kafka vers une table Tablestore.
Connectez-vous à la console ApsaraMQ for Kafka.
Sur la page Overview, sélectionnez une région dans la section Resource Distribution.
Dans le volet de navigation de gauche, cliquez sur Connectors.
Sur la page Connectors, sélectionnez l'instance à laquelle le connecteur appartient dans la liste déroulante Select Instance puis cliquez sur Create Connector.
-
Dans l'assistant Create Connector, effectuez les étapes suivantes.
-
Sous l'onglet Configure Basic Information, configurez les paramètres suivants puis cliquez sur Next.
Paramètre
Description
Exemple
Name
Nom du connecteur. Le nom doit respecter les exigences suivantes :
-
Le nom peut comporter jusqu'à 48 caractères et ne peut contenir que des chiffres, des lettres minuscules et des traits d'union (-). Il ne peut pas commencer par un trait d'union (-).
-
Le nom doit être unique au sein d'une instance ApsaraMQ for Kafka.
La tâche de synchronisation des données du connecteur utilise un consumer group nommé
connect-task-name. Si vous ne créez pas ce consumer group manuellement, le système le crée automatiquement.kafka-ts-sink
Instance
Par défaut, le nom et l'ID de l'instance s'affichent.
demo alikafka_post-cn-st21p8vj****
-
-
Sous l'onglet Configure Source Service, définissez Data Source sur ApsaraMQ for Kafka, configurez les paramètres suivants puis cliquez sur Next.
RemarqueSi vous avez déjà créé le topic et le group requis, sélectionnez Manual pour la création des ressources et saisissez leurs informations. Sinon, sélectionnez Auto.
Tableau 1. Paramètres du service source
Paramètre
Description
Exemple
Data Source Topic
Topic source pour la synchronisation des données.
ts-test-input
Consumer Thread Concurrency
Nombre de threads de consommateur simultanés pour le topic source. La valeur par défaut est 6. Valeurs possibles :
-
1
-
2
-
3
-
6
-
12
6
Consumer Offset
Position de départ pour la consommation des messages. Valeurs possibles :
-
Earliest Offset : Consomme les messages à partir du décalage le plus ancien.
-
Latest Offset : Consomme les messages à partir du décalage le plus récent.
Earliest Offset
VPC ID
ID du réseau privé virtuel (VPC) dans lequel s'exécute la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Ce paramètre correspond par défaut au VPC de l'instance ApsaraMQ for Kafka et ne nécessite aucune configuration.
vpc-bp1xpdnd3l***
vSwitch ID
ID du vSwitch dans lequel s'exécute la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Le vSwitch doit se trouver dans le même VPC que l'instance ApsaraMQ for Kafka. Ce paramètre correspond par défaut au vSwitch spécifié lors du déploiement de l'instance ApsaraMQ for Kafka.
vsw-bp1d2jgg81***
Failure Handling Policy
Politique de gestion des échecs d'envoi de messages dans une partition de topic. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Valeurs possibles :
-
Continue Subscription : Poursuit l'abonnement à la partition de topic où l'erreur s'est produite et imprime les journaux d'erreurs.
-
Stop Subscription : Arrête l'abonnement à la partition de topic où l'erreur s'est produite et imprime les journaux d'erreurs.
Remarque-
Pour plus d'informations sur la consultation des journaux, consultez Gérer les connecteurs.
-
Pour plus d'informations sur la recherche de solutions basées sur les codes d'erreur, consultez Codes d'erreur.
Continue Subscription
Resource Creation Method
Méthode de création des topics et du group requis. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment.
-
Auto
-
Manual
Auto
Connector Consumer Group
group de consommateurs utilisé par la tâche de synchronisation des données. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Le nom du group doit être au format connect-task-name.
connect-cluster-kafka-ots-sink
Task Offset Topic
Topic qui stocke les décalages des consommateurs. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment.
-
Topic : Le nom du topic doit commencer par connect-offset.
-
Partitions : Le nombre de partitions pour le topic doit être supérieur à 1.
-
Moteur de stockage : Le moteur de stockage pour le topic doit être Local Storage.
-
cleanup.policy: La politique de nettoyage des journaux pour le topic doit êtrecompact.
connect-offset-kafka-ots-sink
Task Configuration Topic
Topic qui stocke les configurations des tâches. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment.
-
Topic : Le nom du topic doit commencer par connect-config.
-
Partitions : Le nombre de partitions pour le topic doit être 1.
-
Moteur de stockage : Le moteur de stockage pour le topic doit être Local Storage.
-
cleanup.policy: La politique de nettoyage des journaux pour le topic doit êtrecompact.
connect-config-kafka-ots-sink
Task Status Topic
Topic qui stocke l'état des tâches. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment.
-
Topic : Le nom du topic doit commencer par connect-status.
-
Partitions : Le nombre recommandé de partitions est 6.
-
Moteur de stockage : Le moteur de stockage pour le topic doit être Local Storage.
-
cleanup.policy: La politique de nettoyage des journaux pour le topic doit êtrecompact.
connect-status-kafka-ots-sink
Dead-letter Queue Topic
Topic utilisé pour stocker les données d'erreur du framework Kafka Connect. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Pour économiser les ressources de topic, vous pouvez utiliser le même topic pour la dead-letter queue et le Error Data Topic.
-
Topic : Le nom du topic doit commencer par connect-error.
-
Partitions : Le nombre recommandé de partitions est 6.
-
Moteur de stockage : Le moteur de stockage pour le topic peut être Local Storage ou Cloud Storage.
connect-error-kafka-ots-sink
Error Data Topic
Topic utilisé pour stocker les données d'erreur du connecteur de sortie. Ce paramètre s'affiche après avoir cliqué sur Configure Runtime Environment. Pour économiser les ressources de topic, vous pouvez utiliser le même topic pour le Dead-letter Queue Topic et le topic de données d'erreur.
-
Topic : Le nom du topic doit commencer par connect-error.
-
Partitions : Le nombre recommandé de partitions est 6.
-
Moteur de stockage : Le moteur de stockage pour le topic peut être Local Storage ou Cloud Storage.
connect-error-kafka-ots-sink
-
-
Sous l'onglet Configure Destination Service, définissez Destination Service sur Tablestore, configurez les paramètres suivants puis cliquez sur Create.
Paramètre
Description
Exemple
Instance Name
Nom de l'instance Tablestore.
k00eny67****
Automatically Create Destination Table
Indique s'il faut créer automatiquement une table dans Tablestore.
-
Yes : Une table est automatiquement créée dans Tablestore pour stocker les données synchronisées en fonction du nom de table configuré.
-
No : Une table existante est utilisée pour stocker les données synchronisées.
Yes
Destination Table Name
Nom de la table qui stocke les données synchronisées. Si vous définissez Automatically Create Destination Table sur No, le nom de la table doit être identique à celui d'une table existante dans l'instance Tablestore.
kafka_table
Tablestore
Type de la table qui stocke les données synchronisées.
-
Wide Column Model
-
TimeSeries Model
Wide Column Model
Message Key Format
Format d'entrée de la clé du message. Les valeurs possibles sont String et JSON. La valeur par défaut est JSON. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model.
-
String : La clé du message est analysée directement sous forme de chaîne.
-
JSON : La clé du message doit être au format JSON.
String
Message Value Format
Format d'entrée de la valeur du message. Les valeurs possibles sont String et JSON. La valeur par défaut est JSON. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model.
-
String : La valeur du message est analysée directement sous forme de chaîne.
-
JSON : La valeur du message doit être au format JSON.
String
JSON Message Field Conversion
Méthode de traitement des champs dans un message JSON. Ce paramètre s'affiche si vous définissez Message Key Format ou Message Value Format sur JSON. Valeurs possibles :
-
Write All as String : Convertit tous les champs au type String dans Tablestore.
-
Automatically Identify Field Types : Convertit les champs String et Boolean du corps du message JSON aux types String et Boolean correspondants dans Tablestore. Les types de données Integer et Float du corps du message JSON sont convertis au type Double dans Tablestore.
Write All as String
Primary Key Mode
Spécifie le mode de clé primaire. Vous pouvez extraire les clés primaires de la table à partir de différentes parties des enregistrements de message ApsaraMQ for Kafka, y compris les coordonnées (topic, partition et offset), la clé et la valeur. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model. La valeur par défaut est kafka.
-
kafka : Utilise <connect_topic>_<connect_partition> et <connect_offset> comme clés primaires de la table de données.
-
record_key : Utilise les champs de la clé d'enregistrement comme clés primaires de la table de données.
-
record_value : Utilise les champs de la valeur d'enregistrement comme clés primaires de la table de données.
kafka
Primary Key Column Names
Noms des colonnes de clé primaire de la table de données et leurs types de données correspondants. Les types de données String et Integer sont pris en charge. Cela signifie que les champs de la clé ou de la valeur d'enregistrement qui correspondent aux noms de colonne configurés sont utilisés comme clés primaires pour la table de données.
Ce paramètre s'affiche si vous définissez Message Key Format sur JSON et Primary Key Mode sur record_key, ou si vous définissez Message Value Format sur JSON et Primary Key Mode sur record_value.
Cliquez sur Create pour ajouter un nom de colonne. Vous pouvez configurer un maximum de quatre noms de colonne.
None
Write Mode
Spécifie le mode d'écriture. Les valeurs possibles sont put et update. La valeur par défaut est put. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur Wide Column Model.
-
put : Écrase les données existantes.
-
update : Met à jour les données existantes.
put
Delete Mode
Si un enregistrement de message ApsaraMQ for Kafka contient des valeurs nulles, vous pouvez choisir de supprimer des lignes ou des colonnes d'attributs. Ce paramètre s'affiche lorsque Primary Key Mode est défini sur record_key. Valeurs possibles :
-
none : Valeur par défaut. Aucune suppression n'est autorisée.
-
row : Autorise la suppression de lignes.
-
column : Autorise la suppression de colonnes d'attributs.
-
row_and_column : Autorise la suppression de lignes et de colonnes d'attributs.
Le comportement de suppression dépend du mode d'écriture :
-
Si Write Mode est put, tout mode de suppression entraîne des écritures par écrasement dans la table de données Tablestore, même lorsque la valeur contient des champs nuls.
-
Si Write Mode est update et que le mode de suppression est none ou row, les enregistrements dont tous les champs de valeur sont nuls sont traités comme des données incorrectes. Si seuls certains champs de valeur sont nuls, le connecteur ignore les champs nuls et écrit les champs non nuls dans la table de données Tablestore. Si le mode de suppression est column ou row_and_column, le connecteur supprime les lignes et les colonnes d'attributs pour les champs nuls, puis écrit les données dans la table de données Tablestore.
None
Metric Name Field
Ce champ est mappé au champ de nom de mesure (_m_name) dans le modèle TimeSeries de Tablestore. Le nom de la mesure spécifie le nom de la quantité physique ou de la métrique de surveillance pour les données de séries temporelles, telles que la température ou la vitesse. Ce champ ne peut pas être vide. Ce paramètre s'affiche lorsque vous sélectionnez le modèle TimeSeries pour Tablestore.
measurement
Data Source Field
Mappez ce champ au champ de source de données (_data_source) dans le modèle TimeSeries de Tablestore. Ce champ sert d'identifiant pour la source des données de séries temporelles, tel qu'un nom de machine ou un ID d'appareil, et peut être vide. Ce paramètre s'affiche lorsque le modèle TimeSeries est sélectionné pour Tablestore.
source
Tag Field
Utilisez un ou plusieurs champs comme champs de tag (_tags) pour un modèle TimeSeries de Tablestore. Chaque tag est une paire clé-valeur de chaînes. La clé est le nom du champ configuré et la valeur est le contenu du champ. Les tags font partie des métadonnées de la série temporelle. Une série temporelle est identifiée de manière unique par la combinaison de son nom de métrique, de sa source de données et de ses tags. Les tags peuvent être vides. Ce paramètre s'affiche lorsque Tablestore est sélectionné comme modèle de série temporelle.
tag1, tag2
Timestamp Field
Mappe ce champ au champ d'horodatage (_time) dans le modèle TimeSeries de Tablestore. Il représente le point dans le temps pour cette ligne de données de série temporelle, tel que le moment où une quantité physique est générée. Lors de l'écriture des données dans Tablestore, le champ d'horodatage est converti en microsecondes pour l'écriture et le stockage. Ce paramètre s'affiche lorsque vous sélectionnez le modèle TimeSeries pour Tablestore.
time
Timestamp Unit
Configurez ce paramètre en fonction du champ d'horodatage réel. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur TimeSeries Model. Valeurs possibles :
-
SECONDS
-
MILLISECONDS
-
MICROSECONDS
-
NANOSECONDS
MILLISECONDS
Whether to Map All Non-primary Key Fields
Indique s'il faut mapper tous les champs non-clés primaires en tant que champs de données. Les champs non-clés primaires sont les champs qui ne sont pas déjà mappés en tant que nom de métrique, source de données, tag ou horodatage. Ce paramètre s'affiche uniquement lorsque Tablestore est défini sur TimeSeries Model. Valeurs possibles :
-
Yes : Les champs sont automatiquement mappés et leurs types de données sont déterminés. Les types numériques sont tous convertis au type Double.
-
No : Vous devez spécifier les champs et les types à mapper.
Yes
Configure Mapping for All Non-primary Key Fields
Types de champs correspondant aux noms de champs non-clés primaires de la table de séries temporelles. Les types de données Double, Integer, String, Binary et Boolean sont pris en charge. Ce paramètre s'affiche si vous définissez Whether to Map All Non-primary Key Fields sur No.
String
Une fois le connecteur créé, vous pouvez le consulter sur la page Connectors.
-
-
Une fois le connecteur créé, localisez-le sur la page Connectors et cliquez sur Deploy dans la colonne Actions.
Cliquez sur OK.
Sur la page Connectors, localisez le connecteur cible et cliquez sur Test dans la colonne Actions.
-
Dans le panneau Send Message, envoyez un message de test.
-
Définissez Sending Method sur Console.
Dans la zone de texte Message Key, saisissez la clé du message. Par exemple, demo.
Dans la zone de texte Message Content, saisissez le contenu du message de test. Par exemple, {"key": "test"}.
-
Définissez Send to Specified Partition pour indiquer si le message doit être envoyé à une partition spécifique.
Cliquez sur Yes et saisissez l'ID de partition dans la zone de texte Partition ID. Par exemple, 0. Pour consulter l'ID de partition, consultez la rubrique View partition status.
Cliquez sur No pour ne pas spécifier de partition.
Définissez Sending Method sur Docker. Exécutez la commande Docker indiquée dans la section Run the Docker container to produce a sample message pour envoyer le message.
Définissez Sending Method sur SDK. Sélectionnez un SDK correspondant au langage ou au framework requis, ainsi qu'un type de connexion, pour envoyer les messages.
-
Connectez-vous à la console Tablestore.
Sur la page Overview, cliquez sur le nom de l'instance ou sur Instances dans la colonne Actions.
Sous l'onglet Instance Details, repérez la table cible dans la section Tables.
Cliquez sur le nom de la table. Sur la page Table Manage, accédez à l'onglet Data Management pour consulter les données.
Envoyer un message de test
Après le déploiement du connecteur de destination Tablestore, vous pouvez envoyer un message à la rubrique source dans ApsaraMQ for Kafka afin de vérifier que les données sont synchronisées vers Tablestore.
Consulter les données de la table
Après avoir envoyé un message à la rubrique source de données dans ApsaraMQ for Kafka, consultez les données dans la table Tablestore pour confirmer leur réception.