Apache Kafka consumer implemented in Golang panics when deployed to Docker

Question

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 + "\n"
            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
}

Answer 1

On line 27 of kafka.go, you are ignoring the error returned from ConsumePartition(). It's likely that it's returning an error and not a valid partition consumer, but since you're ignoring it when you attempt to use the partition consumer it's crashing.