Sarama: How to close the partitionConsumer.Messages() channel when no more messages are coming ?

[mandeepm91]

I am using Kafka as a staging area for my data flow pipeline. I am new to both Go as well as Kafka. My Topic has 100 partitions each having a million messages. And there are no new messages coming in. I trigger the consumer only after all the messages in my topic are ready. For my consumer, I need to read all the partitions of the topic one by one. I am using a for loop for that, iterating over the results of consumer.Partitions(myTopic). However, I could not understand how can I terminate the inner for loop of reading the messages:

for {
    select {
    case msg := <-partitionConsumer.Messages():
        consumed++
        // processing
    case <-signals:
        break ConsumerLoop
    }
}

The loop terminates only when an interrupt is received. I need to end the loop when all the messages in the input are exhausted. How can I accomplish that ? Also, I was under the impression that once the messages are read from the partition, we can no longer access those messages again. However, whenever I run my program, I can always fetch those messages. I pass the offset as 0 always though. Is this normal behaviour or am I doing something wrong ?

[eapache]

I need to end the loop when all the messages in the input are exhausted. How can I accomplish that?

Kafka doesn't have the concept of a topic or partition being exhausted; it was designed for systems where new messages are constantly arriving and so the question is meaningless. The best you can do is wait for the configured Consumer.MaxWaitTime time to pass, and if no messages have arrived within that time assume you're done. Or, if you know in advance how many messages there are, you can count the messages you process and stop when you've processed that many.

[eapache]

Also, I was under the impression that once the messages are read from the partition, we can no longer access those messages again.

Not true. Kafka retains messages according to some configurable rules (by default I believe simply 7 days). All retained messages are always available by offset. If you ask for the same offset you will get the same message back. If you want to stop a consumer, then start where you left off, you have to save the offset in between.