Function newLocalClientWithTopic

topics/list_topics_test.go:48–60

(topic string, partitions int)

}

func newLocalClientWithTopic(topic string, partitions int) (*kafka.Client, func()) {
	client, shutdown := newLocalClient()
	if err := clientCreateTopic(client, topic, partitions); err != nil {
		shutdown()
		panic(err)
	}
	return client, func() {
		client.DeleteTopics(context.Background(), &kafka.DeleteTopicsRequest{
			Topics: []string{topic},
		})
		shutdown()
	}
}

func clientCreateTopic(client *kafka.Client, topic string, partitions int) error {
	_, err := client.CreateTopics(context.Background(), &kafka.CreateTopicsRequest{