func NewRecordFormatter(layout string) (*RecordFormatter, error) {
	var f RecordFormatter
	var literal []byte  
	var i int
	for len(layout) > 0 {
		i++
		c, size := utf8.DecodeRuneInString(layout)
		rawc := layout[:size]
		layout = layout[size:]
		switch c {
		default:
			literal = append(literal, rawc...)
			continue
		case '\\':
			c, n, err := parseLayoutSlash(layout)
			if err != nil {
				return nil, err
			}
			layout = layout[n:]
			literal = append(literal, c)
			continue
		case '%':
		}
		if len(layout) == 0 {
			return nil, errors.New("invalid escape sequence at end of layout string")
		}
		cNext, size := utf8.DecodeRuneInString(layout)
		if cNext == '%' || cNext == '{' || cNext == '}' {
			literal = append(literal, byte(cNext))
			layout = layout[size:]
			continue
		}
		var (
			isOpenBrace  = len(layout) > 2 && layout[1] == '{'
			handledBrace = false
			escaped      = layout[0]
		)
		layout = layout[1:]
		if len(literal) > 0 {
			l := literal
			literal = nil
			f.fns = append(f.fns, func(b []byte, _ *FetchPartition, _ *Record) []byte { return append(b, l...) })
		}
		if isOpenBrace {  
			layout = layout[1:]
		}
		switch escaped {
		default:
			return nil, fmt.Errorf("unknown escape sequence %%%s", string(escaped))
		case 'T', 'K', 'V', 'H', 'p', 'o', 'e', 'i', 'x', 'y', '[', '|', ']':
			var numfn func([]byte, int64) []byte
			if handledBrace = isOpenBrace; handledBrace {
				numfn2, n, err := parseNumWriteLayout(layout)
				if err != nil {
					return nil, err
				}
				layout = layout[n:]
				numfn = numfn2
			} else {
				numfn = writeNumAscii
			}
			switch escaped {
			case 'T':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(len(r.Topic))) })
				})
			case 'K':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(len(r.Key))) })
				})
			case 'V':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(len(r.Value))) })
				})
			case 'H':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(len(r.Headers))) })
				})
			case 'p':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(r.Partition)) })
				})
			case 'o':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, r.Offset) })
				})
			case 'e':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(r.LeaderEpoch)) })
				})
			case 'i':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, _ *Record) []byte {
					return numfn(b, atomic.AddInt64(&f.calls, 1))
				})
			case 'x':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, r.ProducerID) })
				})
			case 'y':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(r.ProducerEpoch)) })
				})
			case '[':
				f.fns = append(f.fns, func(b []byte, p *FetchPartition, _ *Record) []byte {
					return writeP(b, p, func(b []byte, p *FetchPartition) []byte { return numfn(b, p.LogStartOffset) })
				})
			case '|':
				f.fns = append(f.fns, func(b []byte, p *FetchPartition, _ *Record) []byte {
					return writeP(b, p, func(b []byte, p *FetchPartition) []byte { return numfn(b, p.LastStableOffset) })
				})
			case ']':
				f.fns = append(f.fns, func(b []byte, p *FetchPartition, _ *Record) []byte {
					return writeP(b, p, func(b []byte, p *FetchPartition) []byte { return numfn(b, p.HighWatermark) })
				})
			}
		case 't', 'k', 'v':
			var appendFn func([]byte, []byte) []byte
			if handledBrace = isOpenBrace; handledBrace {
				switch {
				case strings.HasPrefix(layout, "base64}"):
					appendFn = appendBase64
					layout = layout[len("base64}"):]
				case strings.HasPrefix(layout, "hex}"):
					appendFn = appendHex
					layout = layout[len("hex}"):]
				case strings.HasPrefix(layout, "unpack"):
					unpack, rem, err := nomOpenClose(layout[len("unpack"):])
					if err != nil {
						return nil, fmt.Errorf("unpack parse err: %v", err)
					}
					if len(rem) == 0 || rem[0] != '}' {
						return nil, fmt.Errorf("unpack missing closing } in %q", layout)
					}
					layout = rem[1:]
					appendFn, err = parseUnpack(unpack)
					if err != nil {
						return nil, fmt.Errorf("unpack formatting parse err: %v", err)
					}
				default:
					return nil, fmt.Errorf("unknown %%%s{ escape", string(escaped))
				}
			} else {
				appendFn = appendPlain
			}
			switch escaped {
			case 't':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return appendFn(b, []byte(r.Topic)) })
				})
			case 'k':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return appendFn(b, []byte(r.Key)) })
				})
			case 'v':
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return appendFn(b, []byte(r.Value)) })
				})
			}
		case 'h':
			if !isOpenBrace {
				return nil, errors.New("missing open brace sequence on %h signifying how headers are written")
			}
			handledBrace = true
			braces := 1
			at := 0
			for braces != 0 && len(layout[at:]) > 0 {
				switch layout[at] {
				case '{':
					if at > 0 && layout[at-1] != '%' {
						braces++
					}
				case '}':
					if at > 0 && layout[at-1] != '%' {
						braces--
					}
				}
				at++
			}
			if braces > 0 {
				return nil, fmt.Errorf("invalid header specification: missing closing brace in %q", layout)
			}
			spec := layout[:at-1]
			layout = layout[at:]
			inf, err := NewRecordFormatter(spec)
			if err != nil {
				return nil, fmt.Errorf("invalid header specification %q: %v", spec, err)
			}
			f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
				reuse := new(Record)
				for _, header := range r.Headers {
					reuse.Key = []byte(header.Key)
					reuse.Value = header.Value
					b = inf.AppendRecord(b, reuse)
				}
				return b
			})
		case 'd':
			handledBrace = isOpenBrace
			if !handledBrace {
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return strconv.AppendInt(b, r.Timestamp.UnixNano()/1e6, 10) })
				})
				continue
			}
			switch {
			case strings.HasPrefix(layout, "strftime"):
				tfmt, rem, err := nomOpenClose(layout[len("strftime"):])
				if err != nil {
					return nil, fmt.Errorf("strftime parse err: %v", err)
				}
				if len(rem) == 0 || rem[0] != '}' {
					return nil, fmt.Errorf("%%d{strftime missing closing } in %q", layout)
				}
				layout = rem[1:]
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return strftimeAppendFormat(b, tfmt, r.Timestamp) })
				})
			case strings.HasPrefix(layout, "go"):
				tfmt, rem, err := nomOpenClose(layout[len("go"):])
				if err != nil {
					return nil, fmt.Errorf("go parse err: %v", err)
				}
				if len(rem) == 0 || rem[0] != '}' {
					return nil, fmt.Errorf("%%d{go missing closing } in %q", layout)
				}
				layout = rem[1:]
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return r.Timestamp.AppendFormat(b, tfmt) })
				})
			default:
				numfn, n, err := parseNumWriteLayout(layout)
				if err != nil {
					return nil, fmt.Errorf("unknown %%d{ time specification in %q", layout)
				}
				layout = layout[n:]
				f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
					return writeR(b, r, func(b []byte, r *Record) []byte { return numfn(b, int64(r.Timestamp.UnixNano())/1e6) })
				})
			}
		}
		if isOpenBrace && !handledBrace {
			return nil, fmt.Errorf("unhandled open brace %q", layout)
		}
	}
	if len(literal) > 0 {
		f.fns = append(f.fns, func(b []byte, _ *FetchPartition, _ *Record) []byte { return append(b, literal...) })
	}
	return &f, nil
}
NewRecordFormatter returns a formatter for the given layout, or an error if
the layout is invalid.

The formatter is very powerful, as such there is a lot to describe. This
documentation attempts to be as succinct as possible.

Similar to the fmt package, record formatting is based off of slash escapes
and percent "verbs" (copying fmt package lingo). Slashes are used for common
escapes,

\t \n \r \\ \xNN

printing tabs, newlines, carriage returns, slashes, and hex encoded
characters.

Percent encoding opts in to printing aspects of either a record or a fetch
partition:

%t    topic
%T    topic length
%k    key
%K    key length
%v    value
%V    value length
%h    begin the header specification
%H    number of headers
%p    partition
%o    offset
%e    leader epoch
%d    timestamp (date, formatting described below)
%x    producer id
%y    producer epoch

For AppendPartitionRecord, the formatter also undersands the following three
formatting options:

%[    partition log start offset
%|    partition last stable offset
%]    partition high watermark

The formatter internally tracks the number of times AppendRecord or
AppendPartitionRecord have been called. The special option %i prints the
iteration / call count:

%i    format iteration number (starts at 1)

Lastly, there are three escapes to print raw characters that are usually
used for formatting options:

%%    percent sign
%{    left brace (required if a brace is after another format option)
%}    right brace

Header specification

Specifying headers is essentially a primitive nested format option,
accepting the key and value escapes above:

%K    header key length
%k    header key
%V    header value length
%v    header value

For example, "%H %h{%k %v }" will print the number of headers, and then each
header key and value with a space after each.

Verb modifiers

Most of the previous verb specifications can be modified by adding braces
with a given modifier, e.g., "%V{ascii}". All modifiers are described below.

Numbers

All number verbs accept braces that control how the number is printed:

%v{ascii}       the default, print the number as ascii

%v{hex64}       print 16 hex characters for the number
%v{hex32}       print 8 hex characters for the number
%v{hex16}       print 4 hex characters for the number
%v{hex8}        print 2 hex characters for the number
%v{hex4}        print 1 hex characters for the number
%v{hex}         print as many hex characters as necessary for the number

%v{big64}       print the number in big endian uint64 format
%v{big32}       print the number in big endian uint32 format
%v{big16}       print the number in big endian uint16 format
%v{big8}        alias for byte

%v{little64}    print the number in little endian uint64 format
%v{little32}    print the number in little endian uint32 format
%v{little16}    print the number in little endian uint16 format
%v{little8}     alias for byte

%v{byte}        print the number as a single byte

All numbers are truncated as necessary per each given format.

Timestamps

Timestamps can be specified in three formats: plain number formatting,
native Go timestamp formatting, or strftime formatting. Number formatting is
follows the rules above using the millisecond timestamp value. Go and
strftime have further internal format options:

%d{go##2006-01-02T15:04:05Z07:00##}
%d{strftime[%F]}

An arbitrary amount of pounds, braces, and brackets are understood before
beginning the actual timestamp formatting. For Go formatting, the format is
simply passed to the time package's AppendFormat function. For strftime, all
"man strftime" options are supported.

Text

Topics, keys, and values have "base64", "hex", and "unpack" formatting
options:

%t{hex}
%k{unpack{<bBhH>iIqQc.$}}
%v{base64}

Unpack formatting is inside of enclosing pounds, braces, or brackets, the
same way that timestamp formatting is understood. The syntax roughly follows
Python's struct packing/unpacking rules:

x    pad character (does not parse input)
<    parse what follows as little endian
>    parse what follows as big endian

b    signed byte
B    unsigned byte
h    int16  ("half word")
H    uint16 ("half word")
i    int32
I    uint32
q    int64  ("quad word")
Q    uint64 ("quad word")

c    any character
.    alias for c
s    consume the rest of the input as a string
$    match the end of the line (append error string if anything remains)

Unlike python, a '<' or '>' can appear anywhere in the format string and
affects everything that follows. It is possible to switch endianness
multiple times. If the parser needs more data than available, or if the more
input remains after '$', an error message will be appended.
package kgo
import (
	"bufio"
	"bytes"
	"encoding/base64"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"strconv"
	"strings"
	"sync/atomic"
	"time"
	"unicode/utf8"
	"unsafe"
)
type RecordFormatter struct {
	fns   []func([]byte, *FetchPartition, *Record) []byte
	calls int64
}
func (f *RecordFormatter) AppendRecord(b []byte, r *Record) []byte {
	for _, fn := range f.fns {
		b = fn(b, nil, r)
	}
	return b
}
func (f *RecordFormatter) AppendPartitionRecord(b []byte, p *FetchPartition, r *Record) []byte {
	for _, fn := range f.fns {
		b = fn(b, p, r)
	}
	return b
}