Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

pkg/kgo: use codecType explicitly #697

Merged
merged 1 commit into from
May 26, 2024
+12 −12
Merged

pkg/kgo: use codecType explicitly #697

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions pkg/kgo/compression.go
View file
Original file line number Diff line number Diff line change
@@ -259,10 +259,10 @@ type zstdDecoder struct {
}

func (d *decompressor) decompress(src []byte, codec byte) ([]byte, error) {
switch codec {
case 0:
switch codecType(codec) {
case codecNone:
return src, nil
case 1:
case codecGzip:
ungz := d.ungzPool.Get().(*gzip.Reader)
defer d.ungzPool.Put(ungz)
if err := ungz.Reset(bytes.NewReader(src)); err != nil {
@@ -273,12 +273,12 @@ func (d *decompressor) decompress(src []byte, codec byte) ([]byte, error) {
return nil, err
}
return out.Bytes(), nil
case 2:
case codecSnappy:
if len(src) > 16 && bytes.HasPrefix(src, xerialPfx) {
return xerialDecode(src)
}
return s2.Decode(nil, src)
case 3:
case codecLZ4:
unlz4 := d.unlz4Pool.Get().(*lz4.Reader)
defer d.unlz4Pool.Put(unlz4)
unlz4.Reset(bytes.NewReader(src))
@@ -287,7 +287,7 @@ func (d *decompressor) decompress(src []byte, codec byte) ([]byte, error) {
return nil, err
}
return out.Bytes(), nil
case 4:
case codecZstd:
unzstd := d.unzstdPool.Get().(*zstdDecoder)
defer d.unzstdPool.Put(unzstd)
return unzstd.inner.DecodeAll(src, nil)
12 changes: 6 additions & 6 deletions pkg/kgo/record_formatter.go
View file
Original file line number Diff line number Diff line change
@@ -444,16 +444,16 @@ func NewRecordFormatter(layout string) (*RecordFormatter, error) {
layout = layout[len("compression}"):]
f.fns = append(f.fns, func(b []byte, _ *FetchPartition, r *Record) []byte {
return writeR(b, r, func(b []byte, r *Record) []byte {
switch r.Attrs.CompressionType() {
case 0:
switch codecType(r.Attrs.CompressionType()) {
case codecNone:
return append(b, "none"...)
case 1:
case codecGzip:
return append(b, "gzip"...)
case 2:
case codecSnappy:
return append(b, "snappy"...)
case 3:
case codecLZ4:
return append(b, "lz4"...)
case 4:
case codecZstd:
return append(b, "zstd"...)
default:
return append(b, "unknown"...)

Unchanged files with check annotations Beta

View file
pkg/kgo/consumer.go
case assignPurgeMatching:
// This is slightly different than invalidate in that
// we invalidate whole topics.
loadOffsets.keepFilter(func(t string, p int32) bool {

Check warning on line 1030 in pkg/kgo/consumer.go

GitHub Actions / golangci-lint on amd64 

unused-parameter: parameter 'p' seems to be unused, consider removing or renaming it as _ (revive)
_, ok := assignments[t]
return !ok // assignments are topics to purge -- do NOT keep the topic if it is being purged
})
View file
pkg/kgo/consumer_direct_test.go
r := StringRecord("v")
r.Partition = int32(which % 3)
which++
cl.Produce(ctx, r, func(r *Record, err error) {

Check warning on line 295 in pkg/kgo/consumer_direct_test.go

GitHub Actions / golangci-lint on amd64 

unused-parameter: parameter 'r' seems to be unused, consider removing or renaming it as _ (revive)
if err == context.Canceled {
exit.Store(true)
}
r := StringRecord("v")
r.Topic = ts[which%len(ts)]
which++
cl.Produce(ctx, r, func(r *Record, err error) {

Check warning on line 375 in pkg/kgo/consumer_direct_test.go

GitHub Actions / golangci-lint on amd64 

unused-parameter: parameter 'r' seems to be unused, consider removing or renaming it as _ (revive)
if err == context.Canceled {
exit.Store(true)
}