Function produceRecords

fetch_test.go:16–47

(t *testing.T, n int, addr net.Addr, topic string, compression compress.Codec)

)

func produceRecords(t *testing.T, n int, addr net.Addr, topic string, compression compress.Codec) []Record {
	conn, err := (&Dialer{
		Resolver: &net.Resolver{},
	}).DialLeader(context.Background(), addr.Network(), addr.String(), topic, 0)

	if err != nil {
		t.Fatal("failed to open a new kafka connection:", err)
	}
	defer conn.Close()

	msgs := makeTestSequence(n)
	if compression == nil {
		_, err = conn.WriteMessages(msgs...)
	} else {
		_, err = conn.WriteCompressedMessages(compression, msgs...)
	}
	if err != nil {
		t.Fatal(err)
	}

	records := make([]Record, len(msgs))
	for offset, msg := range msgs {
		records[offset] = Record{
			Offset:  int64(offset),
			Key:     NewBytes(msg.Key),
			Value:   NewBytes(msg.Value),
			Headers: msg.Headers,
		}
	}

	return records
}

func TestClientFetch(t *testing.T) {
	client, topic, shutdown := newLocalClientAndTopic()