{"action":"created","issue":{"url":"https://api.github.com/repos/segmentio/kafka-go/issues/117","repository_url":"https://api.github.com/repos/segmentio/kafka-go","labels_url":"https://api.github.com/repos/segmentio/kafka-go/issues/117/labels{/name}","comments_url":"https://api.github.com/repos/segmentio/kafka-go/issues/117/comments","events_url":"https://api.github.com/repos/segmentio/kafka-go/issues/117/events","html_url":"https://github.com/segmentio/kafka-go/issues/117","id":360449701,"node_id":"MDU6SXNzdWUzNjA0NDk3MDE=","number":117,"title":"Unable to compress messages","user":{"login":"dkoston","id":6424250,"node_id":"MDQ6VXNlcjY0MjQyNTA=","avatar_url":"https://avatars0.githubusercontent.com/u/6424250?v=4","gravatar_id":"","url":"https://api.github.com/users/dkoston","html_url":"https://github.com/dkoston","followers_url":"https://api.github.com/users/dkoston/followers","following_url":"https://api.github.com/users/dkoston/following{/other_user}","gists_url":"https://api.github.com/users/dkoston/gists{/gist_id}","starred_url":"https://api.github.com/users/dkoston/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dkoston/subscriptions","organizations_url":"https://api.github.com/users/dkoston/orgs","repos_url":"https://api.github.com/users/dkoston/repos","events_url":"https://api.github.com/users/dkoston/events{/privacy}","received_events_url":"https://api.github.com/users/dkoston/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":5,"created_at":"2018-09-14T20:47:29Z","updated_at":"2018-09-14T22:01:35Z","closed_at":null,"author_association":"NONE","body":"This is likely user error but I'm not able to compress messages.\r\n\r\nI've set them to use `snappy` with the exact code from README.md (have also tried lz4):\r\n\r\n```\r\n\tw := kafka.NewWriter(kafka.WriterConfig{\r\n\t\tBrokers: []string{\"localhost:9092\"},\r\n\t\tTopic:   \"topic-A\",\r\n\t\tCompressionCodec: snappy.NewCompressionCodec(),\r\n\t})\r\n```\r\n\r\nHowever, when the messages reach Kafka, they are not compressed:\r\n\r\n![screen shot 2018-09-14 at 4 02 59 pm](https://user-images.githubusercontent.com/6424250/45575043-b38eea80-b837-11e8-8e09-87d339c10cf6.png)\r\n\r\n\r\nCode: https://gist.github.com/dkoston/aab79446dc411c9d125a412ea20802c4\r\n\r\nPerhaps there's some broker configuration I need to do? I was under the impression that each message itself contains bits telling the broker what compression codec is used on it's contents and it's unnecessary to specify a codec at the broker level if you do so at the producer/writer.\r\n\r\nIf I add some debug to the client, I see that the messages are running through `transform()` so I'd expect them to be compressed with lz4 and marked as such:\r\n\r\n```\r\n$ go run main.go\r\n2018/09/14 15:41:29 Using snappy for message compression\r\n2018/09/14 15:41:29 Writing 10 messages to topic-A\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 [kafka-go]: compressing\r\n2018/09/14 15:41:30 Reading 10 messages from topic-A as group1\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 15:41:42 [kafka-go]: not compressed: 0\r\n2018/09/14 16:03:36 |----------|--------------|---------------|\r\n2018/09/14 16:03:36 | Messages | Uncompressed | Compressed    |\r\n2018/09/14 16:03:36 |----------|--------------|---------------|\r\n2018/09/14 16:03:36 |       10 |           10 |             0 |\r\n2018/09/14 16:03:36 |----------|--------------|---------------|\r\n```\r\n\r\n\r\nI added:\r\n\r\n```\r\ndiff --git a/message.go b/message.go\r\nindex 2d8b775..12ed4c2 100644\r\n--- a/message.go\r\n+++ b/message.go\r\n@@ -5,6 +5,7 @@ import (\r\n        \"bytes\"\r\n        \"fmt\"\r\n        \"time\"\r\n+       \"log\"\r\n )\r\n\r\n // Message is a data structure representing kafka messages.\r\n@@ -54,9 +55,12 @@ func (msg Message) message() message {\r\n\r\n func (msg Message) encode() (Message, error) {\r\n        if msg.CompressionCodec == nil {\r\n+               log.Printf(\"[kafka-go]: not compressing\")\r\n                return msg, nil\r\n        }\r\n\r\n+       log.Printf(\"[kafka-go]: compressing\")\r\n+\r\n        var err error\r\n        msg.Value, err = transform(msg.Value, msg.CompressionCodec.Encode)\r\n        return msg, err\r\n@@ -65,6 +69,7 @@ func (msg Message) encode() (Message, error) {\r\n func (msg Message) decode() (Message, error) {\r\n        c := msg.message().Attributes & compressionCodecMask\r\n        if c == CompressionNoneCode {\r\n+               log.Printf(\"[kafka-go]: not compressed: %v\", c)\r\n                return msg, nil\r\n        }\r\n\r\n@@ -73,6 +78,8 @@ func (msg Message) decode() (Message, error) {\r\n                return msg, fmt.Errorf(\"codec %d not imported.\", msg.CompressionCodec)\r\n        }\r\n\r\n+       log.Printf(\"[kafka-go]: compressed\")\r\n+\r\n        var err error\r\n        msg.Value, err = transform(msg.Value, codec.Decode)\r\n        return msg, err\r\n```\r\n\r\n\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/segmentio/kafka-go/issues/comments/421495809","html_url":"https://github.com/segmentio/kafka-go/issues/117#issuecomment-421495809","issue_url":"https://api.github.com/repos/segmentio/kafka-go/issues/117","id":421495809,"node_id":"MDEyOklzc3VlQ29tbWVudDQyMTQ5NTgwOQ==","user":{"login":"abraithwaite","id":3791413,"node_id":"MDQ6VXNlcjM3OTE0MTM=","avatar_url":"https://avatars0.githubusercontent.com/u/3791413?v=4","gravatar_id":"","url":"https://api.github.com/users/abraithwaite","html_url":"https://github.com/abraithwaite","followers_url":"https://api.github.com/users/abraithwaite/followers","following_url":"https://api.github.com/users/abraithwaite/following{/other_user}","gists_url":"https://api.github.com/users/abraithwaite/gists{/gist_id}","starred_url":"https://api.github.com/users/abraithwaite/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/abraithwaite/subscriptions","organizations_url":"https://api.github.com/users/abraithwaite/orgs","repos_url":"https://api.github.com/users/abraithwaite/repos","events_url":"https://api.github.com/users/abraithwaite/events{/privacy}","received_events_url":"https://api.github.com/users/abraithwaite/received_events","type":"User","site_admin":false},"created_at":"2018-09-14T22:01:35Z","updated_at":"2018-09-14T22:01:35Z","author_association":"MEMBER","body":"Hmm, you definitely should have those attributes set.\r\n\r\nhttps://github.com/segmentio/kafka-go/blob/48c37f796910d8154479aaa04ade5843ae4f55d0/writer.go#L268-L271\r\n\r\nMaybe there is something funky going on on the broker, but it's my understanding that all compression handling is done in the clients.\r\n\r\nThe command line client almost certainly decompresses though."}}
{"actor":{"display_login":"abraithwaite"}}