Correct use of kafka.Writer

[j4mbob]

Hi, new to go so this may be a silly question.

in my code I use a buffered channel to hold messages obtained from a kafka consumer which I then want to write out using a producer to a different broker.

for msg := range mChan.Messages {

	message := kafka.Message{Value:[]byte(msg)}

	w := &kafka.Writer{
		Addr:                   kafka.TCP(broker),
		Topic:                  topic,
		AllowAutoTopicCreation: true,
		Balancer:               &kafka.LeastBytes{},
	}

	err := w.WriteMessages(context.Background(), message)

	if err != nil {
		log.Println("failed to write messages:", err)
	}

	if err := w.Close(); err != nil {
		log.Println("failed to close writer:", err)
	}
	
}

my question is, should I be creating a new kafka.Writer like this for every message? i tried moving the creation of kafka.Writer outside the range loop thinking I can re-use the same same instance of kafka.Writer but this just results in

failed to write messages: io: read/write on closed pipe errors when i try to WriteMessages

ie:

w := &kafka.Writer{
		Addr:                   kafka.TCP(broker),
		Topic:                  topic,
		AllowAutoTopicCreation: true,
		Balancer:               &kafka.LeastBytes{},
}


for msg := range mChan.Messages {

	message := kafka.Message{Value:[]byte(msg)}

	err := w.WriteMessages(context.Background(), message)

	if err != nil {
		log.Println("failed to write messages:", err)
	}

	if err := w.Close(); err != nil {
		log.Println("failed to close writer:", err)
	}
	
}

any help would be appreciated

thanks

[gonnafaraway]

it doesn't work because you use Close in first iteration
if you read description of method Close, you will understand, that after it kafka client doesn't send any msg
just takeut out if loop and it will work!