Function TestReaderTruncatedResponse

reader_test.go:1730–1799

Tests that the reader can handle messages where the response is truncated due to reaching MaxBytes. If MaxBytes is too small to fit 1 record then it will never truncate, so we start from a small message size and increase it until we are sure truncation has happened at some point.

(t *testing.T)

// we start from a small message size and increase it until we are sure
// truncation has happened at some point.
func TestReaderTruncatedResponse(t *testing.T) {
	topic := makeTopic()
	createTopic(t, topic, 1)
	defer deleteTopic(t, topic)

	readerMaxBytes := 100
	batchSize := 4
	maxMsgPadding := 5
	readContextTimeout := 10 * time.Second

	var msgs []Message
	// The key of each message
	n := 0
	// `i` is the amount of padding per message
	for i := 0; i < maxMsgPadding; i++ {
		bb := bytes.Buffer{}
		for x := 0; x < i; x++ {
			_, err := bb.WriteRune('0')
			require.NoError(t, err)
		}
		padding := bb.Bytes()
		// `j` is the number of times the message repeats
		for j := 0; j < batchSize*4; j++ {
			msgs = append(msgs, Message{
				Key:   []byte(fmt.Sprintf("%05d", n)),
				Value: padding,
			})
			n++
		}
	}

	wr := NewWriter(WriterConfig{
		Brokers:   []string{"localhost:9092"},
		BatchSize: batchSize,
		Async:     false,
		Topic:     topic,
		Balancer:  &LeastBytes{},
	})
	err := wr.WriteMessages(context.Background(), msgs...)
	require.NoError(t, err)

	ctx, cancel := context.WithTimeout(context.Background(), readContextTimeout)
	defer cancel()
	r := NewReader(ReaderConfig{
		Brokers:  []string{"localhost:9092"},
		Topic:    topic,
		MinBytes: 1,
		MaxBytes: readerMaxBytes,
		// Speed up testing
		MaxWait: 100 * time.Millisecond,
	})
	defer r.Close()

	expectedKeys := map[string]struct{}{}
	for _, k := range msgs {
		expectedKeys[string(k.Key)] = struct{}{}
	}
	keys := map[string]struct{}{}