Configuration des propriétés du consommateur

Il existe de nombreux paramètres de configuration pour le consommateur qui contrôlent certains aspects de son comportement. Les paramètres suivants sont parmi les plus importants.

De nombreux autres paramètres de configuration sont disponibles, mais lisez la documentationApache Kafka avant de commencer à les expérimenter.

Groupes de consommateurs stables

Un groupe de consommateurs est un groupe de consommateurs qui utilisent des messages provenant d'une ou de plusieurs rubriques. Les consommateurs d'un groupe utilisent tous la même valeur pour le paramètre de configuration group.id. Si vous avez besoin de plusieurs consommateurs pour gérer votre charge de travail, vous pouvez exécuter plusieurs consommateurs dans le même groupe de consommateurs. Même si vous avez besoin d'un seul consommateur, il est normal de spécifier également une valeur pour group.id.

Chaque groupe de consommateurs dispose d'un serveur dans le cluster appelé coordinateur responsable de l'affectation de partitions aux consommateurs du groupe. Cette responsabilité est répartie dans le cluster afin d'équilibrer la charge. L'affectation des partitions aux consommateurs peut se modifier à chaque rééquilibrage du groupe.

Lorsqu'un consommateur rejoint un groupe de consommateurs, il découvre le coordinateur du groupe. Le consommateur indique ensuite au coordinateur qu'il souhaite rejoindre le groupe et le coordinateur démarre un rééquilibrage des partitions de l'ensemble du groupe pour y inclure le nouveau membre.

Les messages d'une seule partition sont traités par un consommateur de chaque groupe uniquement. Ce mécanisme garantit que les messages de chaque partition sont traités dans l'ordre. Consultez le diagramme suivant pour un exemple où une rubrique contient trois partitions et un groupe de consommateurs, qui consomme cette rubrique, contient deux consommateurs. Deux partitions sont affectées à un consommateur du groupe et une partition est affectée à l'autre consommateur.

Lorsque l'une des modifications suivantes a lieu dans un groupe de consommateurs, le groupe effectue un rééquilibrage en déplaçant l'affectation des partitions vers les membres du groupe afin de prendre en charge la modification :

• Un consommateur rejoint le groupe.
• Un consommateur quitte le groupe.
• Un consommateur est considéré comme n'étant plus en production par le coordinateur.
• De nouvelles partitions sont ajoutées à une rubrique existante.

Si vous avez un groupe de consommateurs qui est rééquilibré, sachez que tout consommateur qui a quitté le groupe voit ses engagements rejetés jusqu'à ce qu'il rejoigne le groupe. Dans ce cas, le consommateur doit rejoindre le groupe, où il lui sera affecté une autre partition que celle à partir de laquelle il consommait précédemment.

Continuité des consommateurs

Kafka détecte automatiquement les consommateurs en échec de manière à réaffecter les partitions aux consommateurs actifs. Il utilise deux mécanismes : l'interrogation et la pulsation.

Si le lot de messages renvoyés par Consumer.poll(...) est important ou si le traitement prend du temps, le délai avant que poll() ne soit appelé à nouveau peut être important ou imprévisible. Dans certains cas, il est nécessaire de configurer un long intervalle maximum entre les interrogations pour que des consommateurs ne soient pas supprimés de leurs groupes uniquement parce que le traitement des messages prend du temps. Si ce mécanisme est le seul disponible, le temps nécessaire à la détection d'un consommateur défaillant est également long.

Afin de faciliter la gestion de la continuité des consommateurs, un mécanisme de signal de présence en arrière-plan a été ajouté dans Kafka 0.10.1. Le coordinateur des groupes attend des membres des groupes qu'ils envoient régulièrement un signal de présence pour indiquer qu'ils sont toujours actifs. Une unité d'exécution de signal de présence en arrière-plan s'exécute dans le consommateur et envoie des signaux de présence réguliers au coordinateur. Si le coordinateur ne reçoit pas de signal de présence d'un membre de groupe dans délai d'attente de session, le coordinateur retire le membre du groupe et démarre un rééquilibrage du groupe. Le délai d'attente de la session peut être beaucoup plus court que l'intervalle d'interrogation maximal, de sorte que le temps nécessaire à la détection d'un consommateur défaillant peut être court, même si le traitement des messages prend beaucoup de temps.

Vous pouvez configurer l'intervalle d'interrogation maximal avec la propriété max.poll.interval.ms et le délai d'attente de session avec la propriété session.timeout.ms. Vous n'avez pas besoin d'utiliser ces paramètres, sauf s'il faut plus de 5 minutes pour traiter un lot de messages.

Gestion des positions

Pour chaque groupe de consommateurs, Kafka maintient le décalage validé pour chaque partition consommée. Lorsqu'un consommateur traite un message, ce dernier n'est pas supprimé de la partition. Au lieu de cela, il met simplement à jour son décalage actuel à l'aide d'un processus appelé validation du décalage.

Event Streams conserve les informations de validation de position pendant 7 jours.

Contrôle de la vitesse de consommation des messages

Si vous rencontrez des problèmes de gestion des messages dus à l'afflux de messages, vous pouvez définir une option de consommation pour contrôler la vitesse de consommation des messages. Utilisez fetch.max.bytes et max.poll.records pour contrôler la quantité de données qu'un appel de poll() peut renvoyer.

Gestion du rééquilibrage des consommateurs

Lorsque des consommateurs sont ajoutés à un groupe ou en sont retirés, un rééquilibrage du groupe a lieu et les consommateurs ne sont pas en mesure de consommer des messages. Il en résulte que tous les consommateurs d'un groupe de consommateurs sont indisponibles pendant une courte période.

Si vous êtes informé par le rappel "on partitions revoked", utilisez un ConsumerRebalanceListener pour valider manuellement les décalages (si vous n'utilisez pas l'auto-commit) et pour mettre en pause la suite du traitement jusqu'à ce que vous soyez informé de la réussite du rééquilibrage par l'utilisation du rappel "on partition assigned".

Fragments de code

Ces fragments de code sont à un niveau élevé pour illustrer les concepts impliqués. Pour des exemples complets, voir les exemples Event Streams dans GitHub.

Pour connecter un consommateur à Event Streams, vous devez créer des données d'identification de service. Pour plus d'informations sur l'obtention de ces données d'identification, voir Connexion à Event Streams.

Dans le code de consommateur, vous devez d'abord générer l'ensemble des propriétés de configuration. Toutes les connexions à Event Streams sont sécurisées par l'utilisation de TLS et d'une authentification par mot de passe, vous avez donc besoin d'au moins ces propriétés. Remplacez BOOTSTRAP_ENDPOINTS, USER et PASSWORD par les informations d'identification de votre propre service :

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

Pour consommer les messages, vous devez également spécifier des désérialiseurs pour les clés et les valeurs, comme dans l'exemple suivant.

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Ces désérialiseurs doivent correspondre aux sérialiseurs utilisés par les producteurs.

Utilisez ensuite un KafkaConsumer pour consommer des messages, chaque message étant représenté par un ConsumerRecord. La méthode la plus courante pour consommer des messages consiste à placer le consommateur dans un groupe de consommateurs en définissant l'ID de groupe, puis d'appeler subscribe() afin d'obtenir une liste de rubriques. Le consommateur est affecté à certaines partitions à consommer, mais, s'il y a plus de consommateurs dans le groupe que de partitions dans la rubrique, il se peut que le consommateur ne soit affecté à aucune partition. Ensuite, appelez poll() dans une boucle pour réceptionner un lot de messages à traiter, chaque message étant représenté par un ConsumerRecord.

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
while (true) {
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records)
     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

Cette boucle de consommation s'exécute indéfiniment, mais elle peut être interrompue à partir d'un autre fil d'exécution en appelant Consumer.wakeup() pour obtenir un arrêt en bon ordre.

Pour valider manuellement des positions, vous devez commencer par définir le paramètre de configuration enable.auto.commit sur false. Ensuite, utilisez Consumer.commmitSync() ou Consumer.commitAsync() pour mettre à jour régulièrement la validation des positions des consommateurs. Par souci de simplicité, cet exemple traite les enregistrements de chaque partition et valide la dernière position séparément.

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

Gestion des exceptions

Toute application puissante qui utilise le client Kafka a besoin de gérer les exceptions relatives à certaines situations prévisibles. Dans certains cas, les exceptions ne sont pas lancées directement car certaines méthodes sont asynchrones et fournissent leurs résultats à l'aide d'un Future ou d'un rappel. Consultez un exemple de code dans GitHub qui présente des exemples complets.

Traitez la liste d'exceptions suivante dans votre code: