Comment obtenir les données à partir de vieux décalage de point de Kafka?

Les consommateurs appartiennent toujours à un groupe et, pour chaque partition, la Gardienne d'animaux permet de suivre les progrès de ce groupe de consommateurs dans la partition.

Pour aller chercher depuis le début, vous pouvez supprimer toutes les données associées avec les progrès que Hussain référencé

ZkUtils.maybeDeletePath(${zkhost:zkport}", "/consumers/${group.id}");

Vous pouvez également spécifier le décalage de la partition que vous voulez, comme spécifié dans la base/src/main/scala/kafka/outils/UpdateOffsetsInZK.scala

ZkUtils.updatePersistentPath(zkClient, topicDirs.consumerOffsetDir + "/" + partition, offset.toString)

Cependant, le décalage n'est pas le temps indexées, mais vous savez pour chaque partition est une séquence.

Si votre message contient un timestamp (et méfiez-vous que cet horodatage n'a rien à voir avec le moment Kafka reçu votre message), vous pouvez essayer de faire un indexeur qui tente de récupérer une entrée dans les étapes en incrémentant le décalage par N, et de stocker le n-uplet (rubrique X, partie 2, offset 100, timestamp) quelque part.

Lorsque vous souhaitez récupérer les entrées à partir d'un certain moment dans le temps, vous pouvez demander une recherche binaire pour l'ébauche de votre index jusqu'à ce que vous trouviez l'entrée que vous voulez et de chercher à partir de là.

Consulter la doc à propos de kafka config : http://kafka.apache.org/08/configuration.html pour votre requête sur les plus petits et les plus grandes valeurs de paramètre d'offset.

BTW, Tout en explorant de kafka, je me demandais comment faire pour relire tous les messages d'un consommateur. Je veux dire, si un groupe de consommateurs a interrogé tous les messages et il veut ré-obtenir celles-ci.

La façon dont il peut être atteint est de supprimer les données de la gardienne. Utilisez la kafka.utils.ZkUtils classe pour supprimer un nœud sur la gardienne. Ci-dessous est son utilisation :

ZkUtils.maybeDeletePath(${zkhost:zkport}", "/consumers/${group.id}");

Kafka Protocole Doc est une grande source de jouer avec de requête/réponse/Offsets/Messages:
https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+De+La+Kafka+Protocole de
vous utilisez Simple Consommateur exemple où le code suivant démontrer l'état:

FetchRequest req = new FetchRequestBuilder()

        .clientId(clientName)

        .addFetch(a_topic, a_partition, readOffset, 100000) 

        .build();

FetchResponse fetchResponse = simpleConsumer.fetch(req);

ensemble readOffset pour démarrer le décalage initial de. mais vous devez vérifier que le max offset ainsi que ci-dessus donnera limitée des décalages de compter comme par FetchSize dans le dernier param de addFetch méthode.

avez-vous essayé?

bin/kafka-console-consumer.sh --bootstrap-serveur localhost:9092 --rubrique test, à partir de début

Il serait d'imprimer tous les messages de la rubrique "test" dans cet exemple.

Plus de détails à partir de ce lien https://kafka.apache.org/quickstart