Function TestMixedCompressedMessages

compress/compress_test.go:218–291

(t *testing.T)

}

func TestMixedCompressedMessages(t *testing.T) {
	client, topic, shutdown := newLocalClientAndTopic()
	defer shutdown()

	offset := 0
	var values []string
	produce := func(n int, codec pkg.Codec) {
		w := &kafka.Writer{
			Addr:      kafka.TCP("127.0.0.1:9092"),
			Topic:     topic,
			Transport: client.Transport,
		}
		defer w.Close()

		if codec != nil {
			w.Compression = kafka.Compression(codec.Code())
		}

		msgs := make([]kafka.Message, n)
		for i := range msgs {
			value := fmt.Sprintf("Hello World %d!", offset)
			values = append(values, value)
			offset++
			msgs[i] = kafka.Message{Value: []byte(value)}
		}

		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := w.WriteMessages(ctx, msgs...); err != nil {
			t.Errorf("failed to produce messages: %+v", err)
		}
	}

	// produce messages that interleave uncompressed messages and messages with
	// different compression codecs.  reader should be able to properly handle
	// all of them.
	produce(10, nil)
	produce(20, new(gzip.Codec))
	produce(5, nil)
	produce(10, new(snappy.Codec))
	produce(10, new(lz4.Codec))
	produce(5, nil)

	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   []string{"127.0.0.1:9092"},
		Topic:     topic,
		Partition: 0,
		MaxWait:   10 * time.Millisecond,
		MinBytes:  1,
		MaxBytes:  1024,
	})
	defer r.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// in order to ensure proper handling of decompressing message, read at
	// offsets that we know to be in the middle of compressed message sets.