Function prepareTestTopics

functional_test.go:335–441

(ctx context.Context, env *testEnvironment)

}

func prepareTestTopics(ctx context.Context, env *testEnvironment) error {
	Logger.Println("creating test topics")
	var testTopicNames []string
	for topic := range testTopicDetails {
		testTopicNames = append(testTopicNames, topic)
	}

	Logger.Println("Creating topics")
	config := NewFunctionalTestConfig()
	config.Metadata.Retry.Max = 5
	config.Metadata.Retry.Backoff = 10 * time.Second
	config.ClientID = "sarama-prepareTestTopics"

	client, err := NewClient(env.KafkaBrokerAddrs, config)
	if err != nil {
		return fmt.Errorf("failed to connect to kafka: %w", err)
	}
	defer client.Close()

	controller, err := client.Controller()
	if err != nil {
		return fmt.Errorf("failed to connect to kafka controller: %w", err)
	}
	defer controller.Close()

	// Start by deleting the test topics (if they already exist)
	{
		request := NewDeleteTopicsRequest(config.Version, testTopicNames, time.Minute)
		deleteRes, err := controller.DeleteTopics(request)
		if err != nil {
			return fmt.Errorf("failed to delete test topics: %w", err)
		}
		for topic, topicErr := range deleteRes.TopicErrorCodes {
			if !isTopicNotExistsErrorOrOk(topicErr) {
				return fmt.Errorf("failed to delete topic %s: %w", topic, topicErr)
			}
		}
	}

	// wait for the topics to _actually_ be gone - the delete is not guaranteed to be processed
	// synchronously
	{
		var topicsOk bool
		request := NewMetadataRequest(config.Version, testTopicNames)
		for i := 0; i < 600 && !topicsOk; i++ {
			time.Sleep(100 * time.Millisecond)
			md, err := controller.GetMetadata(request)
			if err != nil {
				return fmt.Errorf("failed to get metadata for test topics: %w", err)
			}

			if len(md.Topics) == len(testTopicNames) {
				topicsOk = true
				for _, topicsMd := range md.Topics {
					if !isTopicNotExistsErrorOrOk(topicsMd.Err) {
						topicsOk = false
					}
				}