Producer compression might not be functional?

[mtrienis]

Description

It looks like Producer compression might not be working, or it's not providing the right meta data.

How to reproduce

from confluent_kafka import KafkaException, Producer
producer = Producer({"bootstrap.servers":"localhost:9092", 'api.version.request': True, 'compression.type':'lz4'})
producer.produce('blah', 'helloworld')

bin/kafka-run-class kafka.tools.DumpLogSegments --files blah-0/00000000000000000000.log --print-data-log

Dumping /tmp/confluent.RviL8Mcp/kafka/data/blah-0/00000000000000000000.log
Starting offset: 0
offset: 0 position: 0 CreateTime: 1541440525795 isvalid: true keysize: -1 valuesize: 10 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: 0 sequence: -1 isTransactional: false headerKeys: [] payload: helloworld

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()):

confluent-kafka-0.11.4

print confluent_kafka.version()
('0.11.4', 721920)

print confluent_kafka.libversion()
('0.11.4', 722175)

• Apache Kafka broker version:

confluent 5.0.0

• Client configuration: {...}

producer = Producer({"bootstrap.servers":"localhost:9092", 'api.version.request': True, 'compression.type':'lz4'})

• Operating system:
  Linux yzhang2 4.13.0-43-generic producer config property 'acks' is not passed through #48~16.04.1-Ubuntu SMP Thu May 17 12:56:46 UTC 2018 x86_64 x86_64 x86_64 GNU/Linux

[edenhill]

It will only compress data if it actually becomes smaller than the uncompressed data.

[mtrienis]

Thanks for the quick response @edenhill, you're are correct!

[yangyu66]

@mtrienis also you should use "compression.codec" rather than ".type" for python lib.

With the C/C++, Python and Go clients, you can use batch.num.messages to set a limit on the number of messages contained in each batch. To enable compression, use compression.codec.

from https://docs.huihoo.com/apache/kafka/confluent/3.1/clients/producer.html