Function produceRecordWithVersion

pkg/ingest/partition_offset_client_test.go:223–239

(ctx context.Context, t *testing.T, writeClient *kgo.Client, partitionID int32, content []byte, version int)

}

func produceRecordWithVersion(ctx context.Context, t *testing.T, writeClient *kgo.Client, partitionID int32, content []byte, version int) int64 {
	rec := &kgo.Record{
		Value:     content,
		Topic:     topicName,
		Partition: partitionID,
	}
	if version == 0 {
		rec.Headers = nil
	} else {
		rec.Headers = []kgo.RecordHeader{RecordVersionHeader(version)}
	}

	produceResult := writeClient.ProduceSync(ctx, rec)
	require.NoError(t, produceResult.FirstErr())

	return rec.Offset
}

func RecordVersionHeader(version int) kgo.RecordHeader {
	var b [4]byte