Comment lire les données à l'aide de Kafka à la Consommation de l'API depuis le début?

Cela fonctionne avec la 0.9.x consommateur. Fondamentalement, lorsque vous créez un consommateur, vous devez affecter un groupe de consommateurs id à cette consommation à l'aide de la propriété ConsumerConfig.GROUP_ID_CONFIG. Générer la consommation de l'id de groupe au hasard chaque fois que vous démarrez le consommateur de faire quelque chose comme ce properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString()); (propriétés est une instance de java.util.Les propriétés que vous allez passer au constructeur new KafkaConsumer(properties)).

Générer le client de manière aléatoire signifie que le nouveau groupe de consommateurs qui n'ont pas de décalage associées dans kafka. Donc, ce que nous avons à faire, après c'est pour définir une stratégie pour ce scénario. La documentation pour la auto.offset.reset bien dit:

Que faire quand il n'y a pas de décalage initial de Kafka ou si le décalage actuel n'existe plus sur le serveur (par exemple, parce que les données ont été supprimées):

• plus tôt: réinitialiser automatiquement le décalage à la première décalage
• dernière: réinitialiser automatiquement le décalage de la dernière décalage
• aucun: jeter l'exception pour le consommateur si aucune décalage est constaté ou la consommation du groupe
• autre chose: jeter l'exception de la consommation.

Donc parmi les options énumérées ci-dessus, nous devons choisir earliest politique de sorte que le nouveau groupe de consommateurs qui commence dès le début à chaque fois.

Votre code en java, devrait ressembler à quelque chose comme ceci:

properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "your_client_id");
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumer = new KafkaConsumer(properties);

La seule chose que vous devez comprendre maintenant, c'est d'avoir de multiples consommateurs qui appartiennent au même groupe de consommateurs, mais sont répartis de la façon de générer un id aléatoire, et de les distribuer entre ces instances afin qu'ils appartiennent tous au même groupe de consommateurs.

Espère que cela aide!

Une solution possible est d'utiliser une mise en œuvre de ConsumerRebalanceListener lors de la souscription à un ou plusieurs sujets. Le ConsumerRebalanceListener contient des méthodes de rappel lorsque de nouvelles partitions sont attribués ou retirés à partir d'un consommateur. L'exemple de code suivant illustre cela :

public class SkillsConsumer {

private String topic;

private KafkaConsumer<String, String> consumer;

private static final int POLL_TIMEOUT = 5000;

public SkillsConsumer(String topic) {
    this.topic = topic;
    Properties properties = ConsumerUtil.getConsumerProperties();
    properties.put("group.id", "consumer-skills");
    this.consumer = new KafkaConsumer<>(properties);
    this.consumer.subscribe(Collections.singletonList(this.topic),
            new PartitionOffsetAssignerListener(this.consumer));
    }
}

public class PartitionOffsetAssignerListener implements ConsumerRebalanceListener {

private KafkaConsumer consumer;

public PartitionOffsetAssignerListener(KafkaConsumer kafkaConsumer) {
    this.consumer = kafkaConsumer;
}

@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

}

@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    //reading all partitions from the beginning
    for(TopicPartition partition : partitions)
        consumer.seekToBeginning(partition);
}

}

Maintenant, chaque fois que les partitions sont affectées à la consommation, chaque partition sera lu depuis le début.

tout en utilisant le Haut Niveau des props.put("auto.offset.reset", "smallest"); en temps de la création de la ConsumerConfig

Donc pour moi ce qui a fonctionné a été une combinaison de ce qui a été indiqué ci-dessus. Le changement majeur a été d'inclure

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

et ont généré de façon aléatoire une ID de GROUPE à chaque fois. Mais cela seul ne fonctionne pas pour moi. Pour certaines raisons, la première fois j'ai eu le consommateur, il n'a jamais eu de tous les enregistrements. J'ai dû le modifier pour le faire fonctionner -

consumer.poll(0); //without this the below statement never got any records
final ConsumerRecords<Long, String> consumerRecords = consumer.poll(Duration.ofMillis(100));

Je suis nouveau sur KAFKA et n'ai aucune idée de pourquoi cela se passe, mais pour quelqu'un d'autre encore à essayer d'obtenir ce travail, espérons que cette aide.