Timestamp in ProducerMessage is incompatible with compression

[AlbinOS]

Versions

Sarama Version: master
Kafka Version: 0.10.0.1
Go Version: 1.7

Configuration

Default for Sarama and Kafka

Logs

Sarama producer :

ERRO[0000] Failed to write entry:kafka: Failed to produce message to topic cyclop: kafka server: Unexpected (unknown?) server error.

Kafka :

java.lang.IllegalStateException: Compressed message has magic value 0 but inner message has magic value 1

Problem Description

If I use the following Producer configuration, message are not accepted by Kafka :

    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForLocal       // Only wait for the leader to ack
    config.Producer.Compression = sarama.CompressionSnappy   // Compress messages
    config.Producer.Flush.Frequency = 500 * time.Millisecond // Flush batches every 500ms
    config.Version = sarama.V0_10_0_0

    producer, err := sarama.NewAsyncProducer(brokerList, config)
    if err != nil {
        return nil, err
    }

       producer.Input() <- &sarama.ProducerMessage{
        Topic:     topic,
        Value:     sarama.StringEncoder("foo"),
        Timestamp: time.Now(),
    }

According to @eapache in #750, the problem lies in the usage of compression in the producer compression.

By removing config.Producer.Compression = sarama.CompressionSnappy, messages are indeed accepted by Kafka and the timestamp is well set in the message.

[AlbinOS]

Duplicate of #757, sorry for this, I was opening this in reaction of #750 latest comments without looking at the issue list.