Wie liest man Daten von Anfang an mit der Kafka Consumer API?

Question

Bitte sagen Sie mir, wie ich die Nachrichten mit der Kafka Consumer API von Anfang an jedes Mal lesen kann, wenn ich den Consumer jar starte.

Answer 1

während der High Level mit Verbraucher eingestellt props.put("auto.offset.reset", "smallest"); in Zeiten der ConsumerConfig

Answer 2

1. https://stackoverflow.com/a/17084401/3821653

2. http://mail-archives.apache.org/mod_mbox/kafka-users/201403.mbox/%3CCAOG_4QYz2ynH45a8kXb8qw7xw4vDRRwNqMn5j9ERFxJ8RfKGCg@mail.gmail.com%3E

import kafka.utils.ZkUtils; ZkUtils.maybeDeletePath(<zkhost:zkport>, </consumers/group.id>);

Answer 3

Das Erstellen arbeitet mit dem 0.9.x-Consumer. Grundsätzlich müssen Sie beim Erstellen eines Consumers diesem Konsumenten eine Konsumentengruppen-ID mit der Eigenschaft ConsumerConfig.GROUP_ID_CONFIG zuweisen. Generieren Sie die Verbrauchergruppen-ID nach dem Zufallsprinzip, wenn Sie den Verbraucher zum Beispiel so starten. properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString()); (properties ist eine Instanz von java.util.Properties, die Sie an den Konstruktor new KafkaConsumer(properties) übergeben).

Die zufällige Generierung des Clients bedeutet, dass der neuen Verbrauchergruppe in kafka kein Offset zugeordnet ist. Was wir danach tun müssen, ist eine Richtlinie für dieses Szenario festzulegen. Da die Dokumentation für die auto.offset.reset Eigenschaft sagt:

Was tun, wenn es keine Anfänge in Kafka oder wenn der Strom auf dem Server nicht mehr existiert nicht Offset Offset ist (zB weil die Daten gelöscht wurden):

• frühestens: automatisch das die frühesten Offset zurückgesetzt Offset
• zuletzt: automatisch zurückgesetzt den Offset auf den neueste Offset
• none: throw Ausnahme für die Verbraucher, wenn keine vorherige Offset gefunden wird oder die Gruppe des Verbrauchers
• alles andere: Ausnahme zum Verbraucher werfen.

So aus den oben aufgeführten Optionen müssen wir die earliest Politik wählen, so dass die neue Verbrauchergruppe von Anfang an jedes Mal beginnt.

Ihr Code in Java, wird wie folgt aussehen:

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); 

Das einzige, was Sie brauchen es jetzt, um herauszufinden, ist, wenn mehrere Verbraucher aufweist, die auf den gleichen Verbrauchergruppe gehören, sondern verteilt, wie um eine zufällige ID zu generieren und sie zwischen diesen Instanzen zu verteilen, so dass sie alle derselben Verbrauchergruppe angehören.

Hoffe es hilft!

Answer 4

Wenn Sie die java consumer api genauer org.apache.kafka.clients.consumer.Consumer verwenden, können Sie die seek * -Methoden ausprobieren.

consumer.seekToBeginning(consumer.assignment()) 

Hier consumer.assignment() gibt alle zugeordneten Partitionen zu einem bestimmten Verbraucher und seekToBeginning wird von den frühesten für die gegebene Sammlung von Partitionen versetzt starten.

Answer 5

Eine Möglichkeit, dies zu tun, wäre eine eindeutige Gruppen-ID bei jedem Start, was bedeutet, dass Kafka Ihnen die Nachrichten in dem Thema von Anfang an senden würde. Tun Sie etwas wie dies, wenn Sie Ihre Eigenschaften für KafkaConsumer gesetzt:

properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString()); 

Die andere Option ist consumer.seekToBeginning (consumer.assignment()) zu verwenden, aber das wird nicht funktionieren, es sei denn Kafka zunächst einen Herzschlag von Ihrem Verbraucher bekommt durch den Verbraucher dazu bringen, die Poll-Methode aufzurufen. Rufen Sie also poll() auf, führen Sie dann seekToBeginning() aus und rufen Sie dann erneut poll() auf, wenn Sie alle Datensätze von Anfang an möchten. Es ist ein kleiner Hackey, aber dies scheint der zuverlässigste Weg zu sein, um es ab Version 0.9 zu machen.

// At this point, there is no heartbeat from consumer so seekToBeinning() wont work 
// So call poll() 
consumer.poll(0); 
// Now there is heartbeat and consumer is "alive" 
consumer.seekToBeginning(consumer.assingment()); 
// Now consume 
ConsumerRecords<String, String> records = consumer.poll(0); 

Answer 6

Eine mögliche Lösung ist eine Implementierung von ConsumerRebalanceListener zu verwenden, während zu einem oder mehreren Themen abonnieren möchte. Der ConsumerRebalanceListener enthält Callback-Methoden, wenn neue Partitionen einem Consumer zugewiesen oder daraus entfernt werden. Das folgende Codebeispiel veranschaulicht dies:

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); 
} 

}

Nun, wenn die Partitionen den Verbraucher zugeordnet sind, wird jede Partition von Anfang an gelesen werden.