Skip to content

Commit

Permalink
Support nil compressed messages
Browse files Browse the repository at this point in the history
Per https://kafka.apache.org/documentation.html#compaction messages with null
payloads are valid when using log compaction. Make this work for all messages,
not just uncompressed ones. Should fix #634.
  • Loading branch information
eapache committed Jun 9, 2016
1 parent ee044df commit 87fb5e3
Showing 1 changed file with 3 additions and 3 deletions.
6 changes: 3 additions & 3 deletions message.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ func (m *Message) encode(pe packetEncoder) error {
if m.compressedCache != nil {
payload = m.compressedCache
m.compressedCache = nil
} else {
} else if m.Value != nil {
switch m.Codec {
case CompressionNone:
payload = m.Value
Expand Down Expand Up @@ -116,7 +116,7 @@ func (m *Message) decode(pd packetDecoder) (err error) {
// nothing to do
case CompressionGZIP:
if m.Value == nil {
return PacketDecodingError{"GZIP compression specified, but no data to uncompress"}
break
}
reader, err := gzip.NewReader(bytes.NewReader(m.Value))
if err != nil {
Expand All @@ -130,7 +130,7 @@ func (m *Message) decode(pd packetDecoder) (err error) {
}
case CompressionSnappy:
if m.Value == nil {
return PacketDecodingError{"Snappy compression specified, but no data to uncompress"}
break
}
if m.Value, err = snappyDecode(m.Value); err != nil {
return err
Expand Down

0 comments on commit 87fb5e3

Please sign in to comment.