Heap leak when kafka server stopps

[khudaibirdin]

Hello!
I'm trying to create API-Kafka gateway in Go. And I use kafka-go.
I organized internal message buffer with bufferised channel, where api handler writes messages received.
Kafka gorutine reads message channel and send to correct topic.

there's code examples:
main gorutine:

func (k *Kafka) Start(queue *queue.Queue, workerId int, wg *sync.WaitGroup) {
	defer wg.Done()

	k.initWriter()
	
	for message := range queue.GetBuffer() {
		go func() {
			err := k.sendMessage(
				kafka.Message{
					Topic: message.Topic,
					Value: message.Data,
				},
			)
			if err != nil {
				logging.Get().Errorf("[KAFKA_%d] Error while Kafka message %s producing: %s", workerId, message.Id, err)
				return
			}
			logging.Get().Infof("[KAFKA_%d] Message %s sent to Kafka", workerId, message.Id)
		}()
	}
	k.Connection.Writer.Close()
}

writer inition

func (k *Kafka) initWriter() {
	if k.Connection.Writer != nil {
		k.Connection.Writer.Close()
	}
	k.Connection.Writer = &kafka.Writer{
		Addr:                   kafka.TCP(k.Connection.Host),
		AllowAutoTopicCreation: true,
		MaxAttempts: 2,
		WriteTimeout: 5 * time.Second,
		RequiredAcks: 1,
	}
	if k.Connection.Writer.Async {
		k.Connection.Writer.Completion = kafkaMessageResultHandler
	}
	if k.Connection.UseTLS {
		k.Connection.Writer.Transport = &kafka.Transport{
			TLS: k.Connection.TLSCertConfig,
		}
	}
}

message sending method

func (k *Kafka) sendMessage(message kafka.Message) error {
	var err error
	for i := 1; i <= k.SendRetries; i++ {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		err = k.Connection.Writer.WriteMessages(
			ctx,
			message,
		)
		cancel()
		if err == nil {
			return nil
		}
		if errors.Is(err, io.ErrClosedPipe) {
			k.initWriter()
		}
		time.Sleep(time.Millisecond * 100)
	}
	return err
}

it works great while kafka is available.
When I specially stop kafka docker container my app begin to raise heap memory and crash
How to solve the problem. I'm sure that I also could make mistake