Function TestWriteV2RecordBatch

write_test.go:194–247

(t *testing.T)

}

func TestWriteV2RecordBatch(t *testing.T) {
	if !ktesting.KafkaIsAtLeast("0.11.0") {
		t.Skip("RecordBatch was added in kafka 0.11.0")
		return
	}

	client, topic, shutdown := newLocalClientAndTopic()
	defer shutdown()

	msgs := make([]Message, 15)
	for i := range msgs {
		value := fmt.Sprintf("Sample message content: %d!", i)
		msgs[i] = Message{Key: []byte("Key"), Value: []byte(value), Headers: []Header{{Key: "hk", Value: []byte("hv")}}}
	}

	w := &Writer{
		Addr:         TCP("localhost:9092"),
		Topic:        topic,
		BatchTimeout: 100 * time.Millisecond,
		BatchSize:    5,
		Transport:    client.Transport,
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if err := w.WriteMessages(ctx, msgs...); err != nil {
		t.Errorf("Failed to write v2 messages to kafka: %v", err)
		return
	}
	w.Close()

	r := NewReader(ReaderConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   topic,
		MaxWait: 100 * time.Millisecond,
	})
	defer r.Close()

	msg, err := r.ReadMessage(context.Background())
	if err != nil {
		t.Error("Failed to read message")
		return
	}

	if string(msg.Key) != "Key" {
		t.Error("Received message's key doesn't match")
		return
	}
	if msg.Headers[0].Key != "hk" {
		t.Error("Received message header's key doesn't match")
		return
	}
}