Peut-on avoir un traitement séquentiel des messages avec l'API Kafka Streams ?

Question

Nous avons quelques messages pour lesquels nous devons maintenir la séquence. Nous avons décidé d'envoyer tous les messages d'une source particulière à une partition, afin que la séquence des messages soit maintenue ( plusieurs sources peuvent produire vers la même partition mais une source ne peut pas produire vers plusieurs partitions ) et nous serons en mesure d'identifier chaque source avec sa clé.

Maintenant, nous devons consommer ces messages et effectuer un traitement. Nous effectuons plusieurs opérations indépendantes sur les messages que nous avons consommés (par exemple, les stocker dans la base de données, les transférer, etc.) Je ne sais pas si je dois utiliser l'API Kafka Streams ou l'API Consumer pour cela.

Note : Je ne peux pas avoir un grand nombre de sujets (par exemple, je ne peux pas créer un sujet pour chaque source car les sources seront nombreuses). Bien que je puisse regrouper les messages par clé pour identifier la source, pour l'utilisation de Streams, ce que je veux c'est l'ordre des messages pour une clé.

Cas d'utilisation : Je veux enregistrer ces messages dans une base de données dans l'ordre et je veux faire suivre ces messages dans l'ordre .

Alors comment puis-je traiter les messages dans l'ordre en utilisant l'API Streams ?

Answer 1

Nous avons décidé d'envoyer tous les messages d'une source particulière à une partition

Kafka garantit l'ordre au sein d'une même partition, par l'ordre d'arrivée au broker, et non par un autre champ ou par le temps. Tous les clients Kafka (Consumer, Streams, bibliothèques tierces) respectent ce fait.

En règle générale, cependant, si vous incluez l'horodatage du message dans l'événement d'insertion de la base de données, il est possible de regrouper par clé, puis d'ordonner par horodatage. Cela dépend toutefois de la base de données.

Si vous partitionnez déjà par une clé de source connue, pourquoi ne pas simplement filtrer par cette clé dans l'application de flux ? Sinon, vous devrez utiliser l'API Consumer car elle vous permet d'attribuer des partitions spécifiques (l'API Processor, peut-être, mais je ne l'ai pas utilisée).