I am making a simple Telegram bot that would read messages from a local Kafka server and print it out to a chat. Both zookeeper and kafka server config files are at their defaults. Console consumer works. The problem rises when I try to consume messages from code using Golang Sarama package. Before I added these lines:
case err := <-pc.Errors():
log.Panic(err)
the program only printed the messages once, after which it would stall.
Now it panics prinitng this to the log:
kafka: error while consuming test1/0: kafka: broker not connected
Here's the code:
type kafkaResponse struct {
telega *tgbotapi.Message
message []byte
}
type kafkaRequest struct {
telega *tgbotapi.Message
topic string
}
var kafkaBrokers = []string{"localhost:9092"}
func main() {
//channels for request response
var reqChan = make(chan kafkaRequest)
var respChan = make(chan kafkaResponse)
//starting kafka client routine to listen to topic channnel
go consumer(reqChan, respChan, kafkaBrokers)
//bot thingy here
bot, err := tgbotapi.NewBotAPI(token)
if err != nil {
log.Panic(err)
}
bot.Debug = true
log.Printf("Authorized on account %s", bot.Self.UserName)
u := tgbotapi.NewUpdate(0)
u.Timeout = 60
updates, err := bot.GetUpdatesChan(u)
for {
select {
case update := <-updates:
if update.Message == nil {
continue
}
switch update.Message.Text {
case "Topic: test1":
topic := "test1"
reqChan <- kafkaRequest{update.Message, topic}
}
case response := <-respChan:
bot.Send(tgbotapi.NewMessage(response.telega.Chat.ID, string(response.message)))
}
}
here's the consumer.go:
func consumer(reqChan chan kafkaRequest, respChan chan kafkaResponse, brokers []string) {
config := sarama.NewConfig()
config.Consumer.Return.Errors = true
// Create new consumer
consumer, err := sarama.NewConsumer(brokers, config)
if err != nil {
panic(err)
}
defer func() {
if err := consumer.Close(); err != nil {
panic(err)
}
}()
select {
case request := <-reqChan:
//get all partitions on the given topic
partitionList, err := consumer.Partitions(request.topic)
if err != nil {
fmt.Println("Error retrieving partitionList ", err)
}
initialOffset := sarama.OffsetOldest
for _, partition := range partitionList {
pc, _ := consumer.ConsumePartition(request.topic, partition, initialOffset)
go func(pc sarama.PartitionConsumer) {
for {
select {
case message := <-pc.Messages():
respChan <- kafkaResponse{request.telega, message.Value}
case err := <-pc.Errors():
log.Panic(err)
}
}
}(pc)
}
}
}