Kafka 0.8.2.1 come leggere dall'argomento __consumer_offsets

Question

Sto cercando di scoprire quale offset compensa i miei attuali utenti di alto livello. Io uso Kafka 0.8.2.1, con no "offset.storage" impostato nel server.properties di Kafka - che, penso, significa che gli offset sono memorizzati in Kafka. (Ho anche verificato che nessun offset sono memorizzati in Zookeeper controllando questo percorso nella shell Zk: /consumatori/consumer_group_name/offset/TOPIC_NAME/partition_number)

ho cercato di ascoltare i __consumer_offsets argomento per vedere quale consumatore di risparmiare quale valore di offset, ma non ha funzionato ...

ho provato la seguente:

creato un file di configurazione per la console dei consumatori come segue:

=> more kafka_offset_consumer.config 

exclude.internal.topics=false 

e provato due versioni degli script di consumo console:

#1: 
bin/kafka-console-consumer.sh --consumer.config kafka_offset_consumer.config --topic __consumer_offsets --zookeeper localhost:2181 

#2 
./bin/kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 0 --broker-list localhost:9092 --formatter "kafka.server.OffsetManager\$OffsetsMessageFormatter" --consumer.config kafka_offset_consumer.config 

Né lavorato - ma appena ci si siede, ma non stampa nulla, anche se i consumatori sono attivamente consumando/risparmio offset.

Mi mancano altre configurazioni/proprietà?

grazie!

Marina

Answer 1

mi sono imbattuto in questa domanda quando si cerca di consumare anche dalle __consumer_offsets argomento. sono riuscito a capirlo per diverse versioni Kafka e ho pensato di condividere quello che avrei trovato

Per Kafka 0.8.2.x

#Create consumer config 
echo "exclude.internal.topics=false" > /tmp/consumer.config 
#Consume all offsets 
./kafka-console-consumer.sh --consumer.config /tmp/consumer.config \ 
--formatter "kafka.server.OffsetManager\$OffsetsMessageFormatter" \ 
--zookeeper localhost:2181 --topic __consumer_offsets --from-beginning 

Per Kafka 0.9.xx e 0.10.0.0

#Create consumer config 
echo "exclude.internal.topics=false" > /tmp/consumer.config 
#Consume all offsets 
./kafka-console-consumer.sh --consumer.config /tmp/consumer.config \ 
--formatter "kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter" \ 
--zookeeper localhost:2181 --topic __consumer_offsets --from-beginning 

Per 0.11.0.0

#Create consumer config 
echo "exclude.internal.topics=false" > /tmp/consumer.config 
#Consume all offsets 
./kafka-console-consumer.sh --consumer.config /tmp/consumer.config \ 
--formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \ 
--zookeeper localhost:2181 --topic __consumer_offsets --from-beginning 

Answer 2

Se si aggiunge --from-beginning è dovrebbe molto probabilmente vi dà alcuni risultati, almeno lo ha fatto quando mi sono provato. E se non si fornisce quell'argomento ma si leggono più messaggi (e si attivano i commit di offset) mentre si ascolta quel consumatore, questo dovrebbe anche visualizzare i messaggi lì.

Answer 3

Ok, ho capito qual era il problema. Il mio Kafka è stato effettivamente utilizzato Zookeeper come lo stoccaggio offset, non Kafka .... La ragione per cui non rilevare che fin da subito era perché stavo controllando il contenuto ZK in modo non corretto:

stavo facendo

ls /consumers/consumer_group_name/offsets/topic_name/partition_number 

e non vedere nulla lì. Invece ho avuto al contenuto 'get' - che ha fatto spettacolo offset corretti per i miei consumatori, come quella qui sotto:

get /consumers/consumer_group_name/offsets/topic_name/partition_number 
185530404 
cZxid = 0x70789ad05 
ctime = Mon Nov 23 17:49:46 GMT 2015 
mZxid = 0x7216cdc5c 
mtime = Thu Dec 03 20:18:57 GMT 2015 
pZxid = 0x70789ad05 
cversion = 0 
dataVersion = 3537384 
aclVersion = 0 
ephemeralOwner = 0x0 
dataLength = 9 
numChildren = 0