decompressing Snappy encoded RecordBatch fails

[pdeva]

getting error:

 snappy: corrupt input (expected valid offset but got offset 20545; dst position: 0)

changing to another compression like lz4 works fine.

[pdeva]

seems the bug occurs when the batch is compressed with snappy 'framing' turned on.

this is the code from kafka-go client.
when x.framed is set to true, it will cause the decoder in kafka-protocol-rs to crash.
seems its set to true by default for all popular kafka clients including the cannonical java client.

func (c *Codec) NewWriter(w io.Writer) io.WriteCloser {
	x, _ := writerPool.Get().(*xerialWriter)
	if x != nil {
		x.Reset(w)
	} else {
		x = &xerialWriter{writer: w}
	}
	x.framed = c.Framing == Framed
	switch c.Compression {
	case FasterCompression:
		x.encode = s2.EncodeSnappy
	case BetterCompression:
		x.encode = s2.EncodeSnappyBetter
	case BestCompression:
		x.encode = s2.EncodeSnappyBest
	default:
		x.encode = snappy.Encode // aka. s2.EncodeSnappyBetter
	}
	return &writer{xerialWriter: x}
}

[tychedelia]

For compression options, we want to match the official Java client wherever possible, so this would certainly be a bug on our end.