Function TestConsumerPartialBatchRecovery

functional_consumer_test.go:87–131

TestConsumerPartialBatchRecovery exercises the partial-batch recovery path (#1657): the consumer's initial Fetch.Default is smaller than the produced record batch, so the broker returns a partial batch and reports its true size. The follow-up fetch must use that reported size to deliver the message

(t *testing.T)

// size. The follow-up fetch must use that reported size to deliver the
// message in a single retry instead of doubling its way up.
func TestConsumerPartialBatchRecovery(t *testing.T) {
	setupFunctionalTest(t)
	defer teardownFunctionalTest(t)

	const payloadSize = 256 * 1024

	producerConfig := NewFunctionalTestConfig()
	producerConfig.Producer.Return.Successes = true
	producerConfig.Producer.MaxMessageBytes = 2 * payloadSize

	p, err := NewSyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, producerConfig)
	assert.NoError(t, err)
	defer safeClose(t, p)

	payload := make([]byte, payloadSize)
	for i := range payload {
		payload[i] = byte(i)
	}
	_, offset, err := p.SendMessage(&ProducerMessage{Topic: "test.1", Value: ByteEncoder(payload)})
	assert.NoError(t, err)

	consumerConfig := NewFunctionalTestConfig()
	// 12 bytes is the minimum that lets the broker include the per-batch length
	// field in the partial response, so the consumer can size its retry exactly
	consumerConfig.Consumer.Fetch.Default = 12
	consumerConfig.Consumer.Fetch.Max = 2 * payloadSize

	c, err := NewConsumer(FunctionalTestEnv.KafkaBrokerAddrs, consumerConfig)
	assert.NoError(t, err)
	defer safeClose(t, c)

	pc, err := c.ConsumePartition("test.1", 0, offset)
	assert.NoError(t, err)
	defer safeClose(t, pc)

	select {
	case msg := <-pc.Messages():
		assert.Equal(t, offset, msg.Offset)
		assert.Equal(t, payload, msg.Value)
	case err := <-pc.Errors():
		t.Fatalf("consumer error: %v", err)
	case <-time.After(30 * time.Second):
		t.Fatal("timed out waiting for the large message")
	}
}

// Makes sure that messages produced by all supported client versions/
// compression codecs (except LZ4) combinations can be consumed by all