<commit_before>//+build functional

package sarama

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	toxiproxy "github.com/Shopify/toxiproxy/client"
)

const (
	uncomittedMsgJar = "https://github.com/FrancoisPoinsot/simplest-uncommitted-msg/releases/download/0.1/simplest-uncommitted-msg-0.1-jar-with-dependencies.jar"
)

var (
	testTopicDetails = map[string]*TopicDetail{
		"test.1": {
			NumPartitions:     1,
			ReplicationFactor: 3,
		},
		"test.4": {
			NumPartitions:     4,
			ReplicationFactor: 3,
		},
		"test.64": {
			NumPartitions:     64,
			ReplicationFactor: 3,
		},
		"uncommitted-topic-test-4": {
			NumPartitions:     1,
			ReplicationFactor: 3,
		},
	}

	FunctionalTestEnv *testEnvironment
)

func TestMain(m *testing.M) {
	// Functional tests for Sarama
	//
	// You can either set TOXIPROXY_ADDR, which points at a toxiproxy address
	// already set up with 21801-21805 bound to zookeeper and 29091-29095
	// bound to kafka. Alternatively, if TOXIPROXY_ADDR is not set, we'll try
	// and use Docker to bring up a 5-node zookeeper cluster & 5-node kafka
	// cluster, with toxiproxy configured as above.
	//
	// In either case, the following topics will be deleted (if they exist) and
	// then created/pre-seeded with data for the functional test run:
	//     * uncomitted-topic-test-4
	//     * test.1
	//     * test.4
	//     * test.64
	os.Exit(testMain(m))
}

func testMain(m *testing.M) int {
	ctx := context.Background()
	var env testEnvironment

	if os.Getenv("DEBUG") == "true" {
		Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
	}

	usingExisting, err := existingEnvironment(ctx, &env)
	if err != nil {
		panic(err)
	}
	if !usingExisting {
		err := prepareDockerTestEnvironment(ctx, &env)
		if err != nil {
			_ = tearDownDockerTestEnvironment(ctx, &env)
			panic(err)
		}
		defer tearDownDockerTestEnvironment(ctx, &env) // nolint:errcheck
	}
	if err := prepareTestTopics(ctx, &env); err != nil {
		panic(err)
	}
	FunctionalTestEnv = &env
	return m.Run()
}

type testEnvironment struct {
	ToxiproxyClient  *toxiproxy.Client
	Proxies          map[string]*toxiproxy.Proxy
	KafkaBrokerAddrs []string
	KafkaVersion     string
}

func prepareDockerTestEnvironment(ctx context.Context, env *testEnvironment) error {
	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 {
		// We have cp-5.5.0 as the default in the docker-compose file, so that's kafka 2.5.0.
		env.KafkaVersion = "2.5.0"
	}

	// the mapping of confluent platform docker image versions -> kafka versions can be
	// found here: https://docs.confluent.io/current/installation/versions-interoperability.html
	var confluentPlatformVersion string
	switch env.KafkaVersion {
	case "2.5.0":
		confluentPlatformVersion = "5.5.0"
	case "2.4.1":
		confluentPlatformVersion = "5.4.2"
	default:
		return fmt.Errorf("don't know what confluent platform version to use for kafka %s", env.KafkaVersion)
	}

	c := exec.Command("docker-compose", "up", "-d")
	c.Stdout = os.Stdout
	c.Stderr = os.Stderr
	c.Env = append(os.Environ(), fmt.Sprintf("CONFLUENT_PLATFORM_VERSION=%s", confluentPlatformVersion))
	err := c.Run()
	if err != nil {
		return fmt.Errorf("failed to run docker-compose to start test enviroment: %w", err)
	}

	// Set up toxiproxy Proxies
	env.ToxiproxyClient = toxiproxy.NewClient("localhost:8474")
	env.Proxies = map[string]*toxiproxy.Proxy{}
	for i := 1; i <= 5; i++ {
		proxyName := fmt.Sprintf("kafka%d", i)
		proxy, err := env.ToxiproxyClient.CreateProxy(
			proxyName,
			fmt.Sprintf("0.0.0.0:%d", 29090+i),
			fmt.Sprintf("kafka-%d:%d", i, 29090+i),
		)
		if err != nil {
			return fmt.Errorf("failed to create toxiproxy: %w", err)
		}
		env.Proxies[proxyName] = proxy
		env.KafkaBrokerAddrs = append(env.KafkaBrokerAddrs, fmt.Sprintf("127.0.0.1:%d", 29090+i))
	}

	// the mapping of confluent platform docker image vesions -> kafka versions can be
	// found here: https://docs.confluent.io/current/installation/versions-interoperability.html
	// We have cp-5.5.0 in the docker-compose file, so that's kafka 2.5.0.
	env.KafkaVersion = "2.5.0"

	// Wait for the kafka broker to come up
	allBrokersUp := false
	for i := 0; i < 45 && !allBrokersUp; i++ {
		Logger.Println("waiting for kafka brokers to come up")
		time.Sleep(1 * time.Second)
		config := NewConfig()
		config.Version, err = ParseKafkaVersion(env.KafkaVersion)
		if err != nil {
			return err
		}
		config.Net.DialTimeout = 1 * time.Second
		config.Net.ReadTimeout = 1 * time.Second
		config.Net.WriteTimeout = 1 * time.Second
		config.ClientID = "sarama-tests"
		brokersOk := make([]bool, len(env.KafkaBrokerAddrs))
	retryLoop:
		for j, addr := range env.KafkaBrokerAddrs {
			client, err := NewClient([]string{addr}, config)
			if err != nil {
				continue
			}<commit_msg>Remove redundant hard-coding of env.KafkaVersion<commit_after>//+build functional

package sarama

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	toxiproxy "github.com/Shopify/toxiproxy/client"
)

const (
	uncomittedMsgJar = "https://github.com/FrancoisPoinsot/simplest-uncommitted-msg/releases/download/0.1/simplest-uncommitted-msg-0.1-jar-with-dependencies.jar"
)

var (
	testTopicDetails = map[string]*TopicDetail{
		"test.1": {
			NumPartitions:     1,
			ReplicationFactor: 3,
		},
		"test.4": {
			NumPartitions:     4,
			ReplicationFactor: 3,
		},
		"test.64": {
			NumPartitions:     64,
			ReplicationFactor: 3,
		},
		"uncommitted-topic-test-4": {
			NumPartitions:     1,
			ReplicationFactor: 3,
		},
	}

	FunctionalTestEnv *testEnvironment
)

func TestMain(m *testing.M) {
	// Functional tests for Sarama
	//
	// You can either set TOXIPROXY_ADDR, which points at a toxiproxy address
	// already set up with 21801-21805 bound to zookeeper and 29091-29095
	// bound to kafka. Alternatively, if TOXIPROXY_ADDR is not set, we'll try
	// and use Docker to bring up a 5-node zookeeper cluster & 5-node kafka
	// cluster, with toxiproxy configured as above.
	//
	// In either case, the following topics will be deleted (if they exist) and
	// then created/pre-seeded with data for the functional test run:
	//     * uncomitted-topic-test-4
	//     * test.1
	//     * test.4
	//     * test.64
	os.Exit(testMain(m))
}

func testMain(m *testing.M) int {
	ctx := context.Background()
	var env testEnvironment

	if os.Getenv("DEBUG") == "true" {
		Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
	}

	usingExisting, err := existingEnvironment(ctx, &env)
	if err != nil {
		panic(err)
	}
	if !usingExisting {
		err := prepareDockerTestEnvironment(ctx, &env)
		if err != nil {
			_ = tearDownDockerTestEnvironment(ctx, &env)
			panic(err)
		}
		defer tearDownDockerTestEnvironment(ctx, &env) // nolint:errcheck
	}
	if err := prepareTestTopics(ctx, &env); err != nil {
		panic(err)
	}
	FunctionalTestEnv = &env
	return m.Run()
}

type testEnvironment struct {
	ToxiproxyClient  *toxiproxy.Client
	Proxies          map[string]*toxiproxy.Proxy
	KafkaBrokerAddrs []string
	KafkaVersion     string
}

func prepareDockerTestEnvironment(ctx context.Context, env *testEnvironment) error {
	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 {
		// We have cp-5.5.0 as the default in the docker-compose file, so that's kafka 2.5.0.
		env.KafkaVersion = "2.5.0"
	}

	// the mapping of confluent platform docker image versions -> kafka versions can be
	// found here: https://docs.confluent.io/current/installation/versions-interoperability.html
	var confluentPlatformVersion string
	switch env.KafkaVersion {
	case "2.5.0":
		confluentPlatformVersion = "5.5.0"
	case "2.4.1":
		confluentPlatformVersion = "5.4.2"
	default:
		return fmt.Errorf("don't know what confluent platform version to use for kafka %s", env.KafkaVersion)
	}

	c := exec.Command("docker-compose", "up", "-d")
	c.Stdout = os.Stdout
	c.Stderr = os.Stderr
	c.Env = append(os.Environ(), fmt.Sprintf("CONFLUENT_PLATFORM_VERSION=%s", confluentPlatformVersion))
	err := c.Run()
	if err != nil {
		return fmt.Errorf("failed to run docker-compose to start test enviroment: %w", err)
	}

	// Set up toxiproxy Proxies
	env.ToxiproxyClient = toxiproxy.NewClient("localhost:8474")
	env.Proxies = map[string]*toxiproxy.Proxy{}
	for i := 1; i <= 5; i++ {
		proxyName := fmt.Sprintf("kafka%d", i)
		proxy, err := env.ToxiproxyClient.CreateProxy(
			proxyName,
			fmt.Sprintf("0.0.0.0:%d", 29090+i),
			fmt.Sprintf("kafka-%d:%d", i, 29090+i),
		)
		if err != nil {
			return fmt.Errorf("failed to create toxiproxy: %w", err)
		}
		env.Proxies[proxyName] = proxy
		env.KafkaBrokerAddrs = append(env.KafkaBrokerAddrs, fmt.Sprintf("127.0.0.1:%d", 29090+i))
	}

	// Wait for the kafka broker to come up
	allBrokersUp := false
	for i := 0; i < 45 && !allBrokersUp; i++ {
		Logger.Println("waiting for kafka brokers to come up")
		time.Sleep(1 * time.Second)
		config := NewConfig()
		config.Version, err = ParseKafkaVersion(env.KafkaVersion)
		if err != nil {
			return err
		}
		config.Net.DialTimeout = 1 * time.Second
		config.Net.ReadTimeout = 1 * time.Second
		config.Net.WriteTimeout = 1 * time.Second
		config.ClientID = "sarama-tests"
		brokersOk := make([]bool, len(env.KafkaBrokerAddrs))
	retryLoop:
		for j, addr := range env.KafkaBrokerAddrs {
			client, err := NewClient([]string{addr}, config)
			if err != nil {
				continue
			}
{'fraction_non_alphanumeric': 0.13635452175656707, 'fraction_numerical': 0.021756567074393423, 'mean_word_length': 3.62877030162413, 'pattern_counts': {'":': 12, '<': 7, '<?xml version=': 0, '>': 6, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2051301', 'n_tokens_mistral': 4010, 'n_tokens_neox': 3627, 'n_words': 884}