Kafka Niveau Élevé de Consommation Récupérer Tous les Messages De Rubrique à l'Aide de l'API Java (Équivalent à --à partir de début)

Je suis en train de tester le Kafka de Haut Niveau de Consommation à l'aide de la ConsumerGroupExample code de la Kafka site. Je voudrais récupérer tous les messages existants sur le sujet appelé "test" que j'ai dans la Kafka dans la configuration du serveur. Regarder d'autres blogs, de l'automobile.décalage.la réinitialisation doit être réglé sur "plus petit" pour être en mesure d'obtenir tous les messages:

private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId)    {
    Properties props = new Properties();
    props.put("zookeeper.connect", a_zookeeper);
    props.put("group.id", a_groupId);
    props.put("auto.offset.reset", "smallest");
    props.put("zookeeper.session.timeout.ms", "10000");     

    return new ConsumerConfig(props);
}

La question, j'ai vraiment de ceci: ce qui est l'équivalent de l'api Java appel pour le Haut Niveau de Consommation, qui est l'équivalent de:

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --rubrique test, à partir de début

InformationsquelleAutor user3307283 | 2014-02-13

7

En gros, a chaque fois qu'un nouveau type de consommateur essaie de consommer un sujet, il va lire les messages depuis le début. Si vous êtes particulièrement la consommation depuis le début à chaque fois pour des fins de test, chaque fois que vous utilisez votre consommation avec une nouvelle id de groupe, il va lire les messages depuis le début. Voici comment je l'ai fait :
```
properties.put("group.id", UUID.randomUUID().toString());
```
et de lire les messages depuis le début à chaque fois!
- Merci! Besoin à des fins de test. Je suppose que c'est parce que vous pouvez réutiliser les mêmes données à des fins différentes?
- Oui, j'avais eu besoin de chacun des différents tests afin de travailler sur les mêmes données.
InformationsquelleAutor Karishma Gulati
5

Ressemble, vous devez utiliser le "faible niveau de SimpleConsumer API"

Pour la plupart des applications, le niveau élevé de consommation de l'Api est assez bon.
Certaines applications voulez fonctionnalités ne sont pas exposés à haut niveau de consommation
encore (par exemple, définir le décalage initial lors du redémarrage de la consommation). Ils peuvent
au lieu de cela, utilisez notre faible niveau SimpleConsumer Api. La logique sera un peu
plus compliqué et vous pouvez suivre l'exemple de ici.

Cet exemple travaillé pour obtenir tous les messages d'un sujet avec les arguments suivants: (à noter que le port est le Kafka de port, pas la Gardienne du port, des sujets mis en place à partir cet exemple):
```
10 my-replicated-topic 0 localhost 9092
```
Plus précisément, il existe une méthode pour obtenir readOffset qui prend kafka.l'api.OffsetRequest.EarliestTime():
```
long readOffset = getLastOffset(consumer,a_topic, a_partition, kafka.api.OffsetRequest.EarliestTime(), clientName);
```
Voici un autre poste peut fournir certains autres idées sur la façon de régler cette question: Comment obtenir les données à partir de vieux décalage de point de Kafka?
- Ce que vous avez utilisé pour l'atteindre? pour lire tous les messages du sujet.
InformationsquelleAutor pherris
2

Pour extraire les messages depuis le début, vous pouvez le faire:
```
import kafka.utils.ZkUtils;
ZkUtils.maybeDeletePath("zkhost:zkport", "/consumers/group.id");
```
puis il suffit de suivre le travail de routine...

InformationsquelleAutor Student Jack

 Properties props = new Properties(); 
 props.put("bootstrap.servers", "localhost:9092");
 props.put("auto.offset.reset", "earliest");
 props.put("group.id", UUID.randomUUID().toString());

Cette propriétés de vous aider.

InformationsquelleAutor Oleksandr Buryk

Vous devez vous connecter pour publier un commentaire.