here is my attempt in implementing a simple microservice that is supposed to read messages from a kafka server and send it via HTTP. It works fine when i run it from the terminal, but when deployed on to docker it panics with
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x40 pc=0x7b6345]
goroutine 12 [running]:
main.kafkaRoutine.func1(0xc420174060, 0x0, 0x0)
/go/src/github.com/deathcore666/ProperConsumerServiceYo/kafka.go:36 +0x95
created by main.kafkaRoutine
/go/src/github.com/deathcore666/ProperConsumerServiceYo/kafka.go:32 +0x1ad
kafka.go lines 32 and 36 are the ones where go func(pc sarama.PartitionConsumer) function sits.
I am relatively new to programming, so any help would be appreciated. Thank you!
main.go:
func main() {
var (
listen = flag.String("listen", ":8080", "HTTP listen address")
proxy = flag.String("proxy", "", "Optional comma-separated list of URLs to proxy uppercase requests")
)
flag.Parse()
logger := log.NewLogfmtLogger(os.Stderr)
var svc KafkaService
svc = kafkaService{}
svc = proxyingMiddleware(context.Background(), *proxy, logger)(svc)
svc = loggingMiddleware(logger)(svc)
consumehandler := httptransport.NewServer(
makeConsumeEndpoint(svc),
decodeConsumeRequest,
encodeResponse,
)
http.Handle("/consume", consumehandler)
logger.Log("msg", "HTTP", "addr", *listen)
logger.Log("err", http.ListenAndServe(*listen, nil))}
service.go:
package main
import (
"context"
"errors"
"time"
)
//KafkaService yolo
type KafkaService interface {
Consume(context.Context, string) (string, error)
}
//ErrEmpty yolo
var ErrEmpty = errors.New("No topic provided")
type kafkaService struct{}
//Consumer logic implemented here
func (kafkaService) Consume(_ context.Context, topic string) (string, error) {
if topic == "" {
return "", ErrEmpty
}
var inChan = make(chan string)
var readyChan = make(chan struct{})
var result string
var brokers = []string{"192.168.88.208:9092"}
//var brokersLocal = []string{"localhost:9092"}
go kafkaRoutine(inChan, topic, brokers)
go func() {
for {
select {
case msg := <-inChan:
result = result + msg + "
"
case <-time.After(time.Second * 1):
readyChan <- struct{}{}
}
}
}()
<-readyChan
close(inChan)
return result, nil
}
//ServiceMiddleware is a chainable thing for the service
type ServiceMiddleware func(KafkaService) KafkaService
kafka.go:
package main
import (
"fmt"
"time"
"github.com/Shopify/sarama"
)
func kafkaRoutine(inChan chan string, topic string, brokers []string) {
config := sarama.NewConfig()
config.Consumer.Return.Errors = true
consumer, err := sarama.NewConsumer(brokers, config)
if err != nil {
panic(err)
}
topics, _ := consumer.Topics()
if !(containsTopic(topics, topic)) {
inChan <- "There is no such a topic"
fmt.Println("kafkaroutine exited")
return
}
partitionList, err := consumer.Partitions(topic)
for _, partition := range partitionList {
pc, _ := consumer.ConsumePartition(topic, partition, sarama.OffsetOldest)
go func(pc sarama.PartitionConsumer) {
loop:
for {
select {
case msg := <-pc.Messages():
inChan <- string(msg.Value)
case <-time.After(time.Second * 1):
break loop
}
}
}(pc)
}
fmt.Println("Kafka GoRoutine exited")
}
func containsTopic(topics []string, topic string) bool {
for _, v := range topics {
if topic == v {
return true
}
}
return false
}
