Function TestReaderReadCompactedMessage

reader_test.go:1808–1861

Tests that the reader can read record batches from log compacted topics where the batch ends with compacted records. This test forces varying sized chunks of duplicated messages along with configuring the topic with a minimal `segment.bytes` in order to guarantee that at least 1 batch can be compac

(t *testing.T)

// guarantee that at least 1 batch can be compacted down to 0 "unread" messages
// with at least 1 "old" message otherwise the batch is skipped entirely.
func TestReaderReadCompactedMessage(t *testing.T) {
	topic := makeTopic()
	createTopicWithCompaction(t, topic, 1)
	defer deleteTopic(t, topic)

	msgs := makeTestDuplicateSequence()

	writeMessagesForCompactionCheck(t, topic, msgs)

	expectedKeys := map[string]int{}
	for _, msg := range msgs {
		expectedKeys[string(msg.Key)] = 1
	}

	// kafka 2.0.1 is extra slow
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*120)
	defer cancel()
	for {
		success := func() bool {
			r := NewReader(ReaderConfig{
				Brokers:  []string{"localhost:9092"},
				Topic:    topic,
				MinBytes: 200,
				MaxBytes: 200,
				// Speed up testing
				MaxWait: 100 * time.Millisecond,
			})
			defer r.Close()

			keys := map[string]int{}
			for {
				m, err := r.FetchMessage(ctx)
				if err != nil {
					t.Logf("can't get message from compacted log: %v", err)
					return false
				}
				keys[string(m.Key)]++

				if len(keys) == countKeys(msgs) {
					t.Logf("got keys: %+v", keys)
					return reflect.DeepEqual(keys, expectedKeys)
				}
			}
		}()
		if success {
			return
		}
		select {
		case <-ctx.Done():
			t.Fatal(ctx.Err())
		default:
		}
	}
}

// writeMessagesForCompactionCheck writes messages with specific writer configuration.
func writeMessagesForCompactionCheck(t *testing.T, topic string, msgs []Message) {