kafka Verbraucher Registrierungsliste in zoekeeper

Question

Ich benutze C++ kafka lib zu produzieren/konsumieren Nachrichten mit Kafka - alles funktioniert gut.

Jetzt möchte ich meinen Kunden überwachen, um Unterbrechungen/Fehler zu behandeln. Ich suche eine Art Verbindungsliste aller Verbraucher.

von kafka documentation:

Consumer Id Registry

Neben dem group_id, die von allen Verbrauchern in einer Gruppe geteilt wird, wird jeder Verbraucher eine vorübergehende, einmalige consumer_id gegeben (von Die Form Hostname: UUID) zu Identifikationszwecken. Consumer-IDs sind im folgenden Verzeichnis registriert.

/consumers/[group_id]/ids/[consumer_id] --> {"topic1": #streams, ..., "topicN": #streams} (ephemeral node) 

Jedes des Verbrauchers in der Gruppe registriert unter seiner Gruppe und schafft eine ZNODE mit seinem consumer_id. Der Wert des Znode enthält eine Karte von <topic, #streams>. Diese ID wird einfach verwendet, um jede der Verbraucher, die derzeit in einer Gruppe aktiv ist, zu identifizieren. Dies ist ein ephemerer Knoten, so dass er verschwindet, wenn der Verbraucherprozess stirbt.

Aber wenn zu ls /consumers versuchen - es gibt nichts dort (meine app ist bis & läuft und die Verbraucher konsumieren Nachrichten über log)

Answer 1

In kafka 0,9-Cluster nur old-consumers(zookeeper-based offset storage) werden unter „registriert/Verbraucher/[group_id ]/ids/[consumer_id] ", new-consumer(kafka-based offset storage) wird nicht unter dem Pfad"/consumers/[group_id] "registriert. (meine Cluster-Version ist kafka 0.9.0.1)

Answer 2

Die librdkafka High-Level-KafkaConsumer auf die neuen Broker-basierte ausgewogene Verbraucher Gruppe überhaupt in Apache Kafka 0,9 und sie nicht verwenden ZooKeeper beruht.

Sie können das Skript bin/kafka-consumer-groups.sh --new-consumer .. in der Kafka-Distribution verwenden, um registrierte Verbrauchergruppen aufzulisten und zu beschreiben. Es gibt auch eine API, um diese Informationen programmatisch in der C-Schnittstelle von librdkafka zu erhalten (rd_kafka_list_groups()), und sie wird in Kürze in C++ verfügbar sein.