Function prepareDockerTestEnvironment

functional_test.go:143–262

(ctx context.Context, env *testEnvironment)

}

func prepareDockerTestEnvironment(ctx context.Context, env *testEnvironment) error {
	const expectedBrokers = 5

	Logger.Println("bringing up docker-based test environment")

	// Always (try to) tear down first.
	if err := tearDownDockerTestEnvironment(ctx, env); err != nil {
		return fmt.Errorf("failed to tear down existing env: %w", err)
	}

	if version, ok := os.LookupEnv("KAFKA_VERSION"); ok {
		env.KafkaVersion = version
	} else {
		env.KafkaVersion = "3.5.1"
	}
	// docker compose v2.17.0 or newer required for `--wait-timeout` support
	args := []string{"compose", "up", "-d", "--quiet-pull", "--timestamps", "--wait", "--wait-timeout", "600"}
	v, _ := ParseKafkaVersion(env.KafkaVersion)
	// use zookeeper for kafka < 4
	if !v.IsAtLeast(V4_0_0_0) {
		args = append([]string{"compose", "--profile", "zookeeper"}, args[1:]...)
	}
	c := exec.Command("docker", args...)
	c.Stdout = os.Stdout
	c.Stderr = os.Stderr
	c.Env = append(os.Environ(), fmt.Sprintf("KAFKA_VERSION=%s", env.KafkaVersion))
	err := c.Run()
	if err != nil {
		return fmt.Errorf("failed to run docker compose to start test environment: %w", err)
	}

	if err := setupToxiProxies(env, "http://localhost:8474"); err != nil {
		return fmt.Errorf("failed to setup toxiproxies: %w", err)
	}

	dialCheck := func(addr string, timeout time.Duration) error {
		conn, err := net.DialTimeout("tcp", addr, timeout)
		if err != nil {
			return err
		}
		return conn.Close()
	}

	config := NewFunctionalTestConfig()
	config.Net.DialTimeout = 1 * time.Second
	config.Net.ReadTimeout = 1 * time.Second
	config.Net.WriteTimeout = 1 * time.Second
	config.ClientID = "sarama-tests"

	// wait for the kafka brokers to come up
	allBrokersUp := false

	Logger.Printf("waiting for kafka %s brokers to come up...\n", env.KafkaVersion)

mainLoop:
	for i := 0; i < 90 && !allBrokersUp; i++ {
		if i > 0 {
			Logger.Printf("still waiting for kafka %s brokers to come up...\n", env.KafkaVersion)