Reconnect to kafka from Goroutine

I want to write a message to kafka using Golang. I have the following code.

package kafkaK

import (
	"context"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/snappy"
	"time"
)

var writer *kafka.Writer

func Configure() (w *kafka.Writer, err error) {
	dialer := &kafka.Dialer{
		Timeout:  10 * time.Second,
		ClientID: "123",
	}

	config := kafka.WriterConfig{
		Brokers:          []string{"localhost:9092"},
		Topic:            "test",
		Balancer:         &kafka.LeastBytes{},
		Dialer:           dialer,
		WriteTimeout:     10 * time.Second,
		ReadTimeout:      10 * time.Second,
		CompressionCodec: snappy.NewCompressionCodec(),
	}
	w = kafka.NewWriter(config)
	writer = w
	return w, nil
}

func Push(parent context.Context, key, value []byte) (err error) {
	message := kafka.Message{
		Key:   key,
		Value: value,
		Time:  time.Now(),
	}
	return writer.WriteMessages(parent, message)
}

I write to kafka in separate Goroutines.

func sendMessage(message string) {
	err := kafkaK.Push(context.Background(), nil, []byte(message))
	if err != nil {
		fmt.Println(err)
	}
}

The call looks like

go sendMessage("message #" + strconv.Itoa(i))

1. It happens that kafka is unavailable, and I would like to reconnect to it. How to do it correctly if different Goroutines use the same kafka object, and only they know that an error has occurred and can initiate reconnection. Perhaps there is some other approach?
2. Which library is better to use for working with kafka?

I think it's possible to use a channel or context, but I'm new to go, so I don't really understand how it can be implemented yet

First I want to point out that you're using a deprecated interface in kafka-go. This documentation specifies that you can use kafkago.Writer{} to supply the config to connect and write to you kafka cluster. Now to answer your questions:

> It happens that kafka is unavailable, and I would like to reconnect to it. How to do it correctly if different Goroutines use the same kafka object, and only they know that an error has occurred and can initiate reconnection. Perhaps there is some other approach?

With a high level library like kafka-go, you don't have to worry about handling connections or transient errors like temporary unavailability of your kafka cluster. When you use the WriteMessages method on kafka writer, it buffers your messages and flushes periodically after connecting to your brokers using the config you provide to it. So, as long as the config is correct kafka-go will handle the rest for you.

> Which library is better to use for working with kafka?

kafka-go is definitely widely used and actively maintained for reading and writing to kafka.

Also, back to your example, this is how I would refactor it:

package kafkak
import (
"context"
"fmt"
"github.com/segmentio/kafka-go"
"sync"
"time"
)
// Define an interface so you can mock it for testing.
type kproducer interface {
WriteMessages(ctx context.Context, msgs ...kafka.Message) error
}
type KafkaK struct {
p      kproducer
dataCh chan kafka.Message // Write messages to this channel so a separate go routine can process them
errCh  chan error // Fetch errors from this channel
stopCh chan struct{}
}
func NewKafkaK(p kproducer) *KafkaK {
k := &KafkaK{
p:      p,
dataCh: make(chan kafka.Message),
errCh:  make(chan error),
stopCh: make(chan struct{}),
}
go k.processLoop()
return k
}
func (k *KafkaK) Push(key, value string) {
m := kafka.Message{
Key:   []byte(key),
Value: []byte(value),
Time:  time.Now(),
}
k.dataCh <- m
}
func (k *KafkaK) processLoop() {
for {
select {
case msg := <-k.dataCh:
ctx, _ := context.WithDeadline(context.Background(), time.Now().Add(time.Second*5))
err := k.p.WriteMessages(ctx, msg)
k.errCh <- err
case <-k.stopCh:
break
}
}
}
func (k *KafkaK) Stop() {
close(k.stopCh)
close(k.errCh)
}
func (k *KafkaK) Errs() <-chan error {
return k.errCh
}