Function createTopicWithCompaction

reader_test.go:1931–1972

(t *testing.T, topic string, partitions int)

}

func createTopicWithCompaction(t *testing.T, topic string, partitions int) {
	t.Helper()

	t.Logf("createTopic(%s, %d)", topic, partitions)

	conn, err := Dial("tcp", "localhost:9092")
	require.NoError(t, err)
	defer conn.Close()

	controller, err := conn.Controller()
	require.NoError(t, err)

	conn, err = Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
	require.NoError(t, err)

	conn.SetDeadline(time.Now().Add(10 * time.Second))

	err = conn.CreateTopics(TopicConfig{
		Topic:             topic,
		NumPartitions:     partitions,
		ReplicationFactor: 1,
		ConfigEntries: []ConfigEntry{
			{
				ConfigName:  "cleanup.policy",
				ConfigValue: "compact",
			},
			{
				ConfigName:  "segment.bytes",
				ConfigValue: "200",
			},
		},
	})
	if err != nil {
		if !errors.Is(err, TopicAlreadyExists) {
			require.NoError(t, err)
		}
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	waitForTopic(ctx, t, topic)
}

// The current behavior of the Reader is to retry OffsetOutOfRange errors
// indefinitely, which results in programs hanging in the event of a topic being