卡夫卡消息未传递给消费者

I am facing a very weird problem over here. I started a Kafka consumer using sarama-cluster library in go to consume some messages from a kafka topic. But the messages are not being received by the consumer being started.

However, a very weird thing is happening. If I start another consumer parallel to it, the messages are suddenly being delivered to both the consumers.

I cannot think of a logical explanation to it. Any pointers will be appreciated.

Note : This problem started after the Kafka and Zookeeper servers were started non-gracefully.

Here is the go code of consumer for consuming the messages which is not working:

if err := consumer.Start(); err != nil {
    return err
}
updChan, err := consumer.Consume()
if err != nil {
    return err
}
go func() {
    for {
        select {
        case msg, ok := <-updChan:
            if !ok {
                return
            }
            var message liveupdater.KafkaMessage
            err := json.Unmarshal(msg.Msg, &message)
            if err != nil {
                fmt.Println(err)
            }
            err = handleMessaege(message)
            if err != nil {
                logrus.Println("encountered error:" + err.Error())
            }

            consumer.MarkProcessed(msg, string(message.Type))
        }
    }
}()

Following is the go code where consumer is receiving messages(the only difference between this and previous code is another consumer running in parallel for same topic).

consumeMessages(config)

if err := consumer.Start(); err != nil {
    return err
}
updChan, err := consumer.Consume()
if err != nil {
    return err
}
go func() {
    for {
        select {
        case msg, ok := <-updChan:
            if !ok {
                return
            }
            var message liveupdater.KafkaMessage
            err := json.Unmarshal(msg.Msg, &message)
            if err != nil {
                fmt.Println(err)
            }
            err = handleMessaege(message)
            if err != nil {
                logrus.Println("encountered error:" + err.Error())
            }

            consumer.MarkProcessed(msg, string(message.Type))
        }
    }
}()

func consumeMessages(config *rakshak_config.Config) {
    kafkaConfig := kafka.Config{Brokers: strings.Split(config.Kafka.Brokers, ",")}
    logrus.Println("brokers %s", config.Kafka.Brokers)
    hermesConsumer, err := hermes.NewConsumer(hermes.Kafka, []string{config.Kafka.Topic}, kafkaConfig)
    if err != nil {
        logrus.Println("could not get consumer through hermes %s", err)
    }
    err = hermesConsumer.Start()
    if err != nil {
        logrus.Println("could not start consumer through hermes %s", err)
    }
    conChan, err := hermesConsumer.Consume()
    if err != nil {
        logrus.Println("not able to start consumer channel %s", err)
    }
    go func() {
        for {
            select {
            case msg, ok := <-conChan:
                if !ok {
                    logrus.Println("could not consume message")
                }

                logrus.Println("kafka msg string: %s", string(msg.Msg[:]))

                hermesConsumer.MarkProcessed(msg, "")
            }
        }
    }()

Thanks in advance.