[VOL-4291] Rw-core updates for gRPC migration
Change-Id: I8d5a554409115b29318089671ca4e1ab3fa98810
diff --git a/vendor/github.com/Shopify/sarama/message.go b/vendor/github.com/Shopify/sarama/message.go
index 7c54748..fd0d1d9 100644
--- a/vendor/github.com/Shopify/sarama/message.go
+++ b/vendor/github.com/Shopify/sarama/message.go
@@ -6,15 +6,15 @@
)
const (
- //CompressionNone no compression
+ // CompressionNone no compression
CompressionNone CompressionCodec = iota
- //CompressionGZIP compression using GZIP
+ // CompressionGZIP compression using GZIP
CompressionGZIP
- //CompressionSnappy compression using snappy
+ // CompressionSnappy compression using snappy
CompressionSnappy
- //CompressionLZ4 compression using LZ4
+ // CompressionLZ4 compression using LZ4
CompressionLZ4
- //CompressionZSTD compression using ZSTD
+ // CompressionZSTD compression using ZSTD
CompressionZSTD
// The lowest 3 bits contain the compression codec used for the message
@@ -42,7 +42,7 @@
}[int(cc)]
}
-//Message is a kafka message type
+// Message is a kafka message type
type Message struct {
Codec CompressionCodec // codec used to compress the message contents
CompressionLevel int // compression level
@@ -85,7 +85,6 @@
payload = m.compressedCache
m.compressedCache = nil
} else if m.Value != nil {
-
payload, err = compress(m.Codec, m.CompressionLevel, m.Value)
if err != nil {
return err
@@ -147,18 +146,12 @@
// for future metrics about the compression ratio in fetch requests
m.compressedSize = len(m.Value)
- switch m.Codec {
- case CompressionNone:
- // nothing to do
- default:
- if m.Value == nil {
- break
- }
-
+ if m.Value != nil && m.Codec != CompressionNone {
m.Value, err = decompress(m.Codec, m.Value)
if err != nil {
return err
}
+
if err := m.decodeSet(); err != nil {
return err
}