|
|
@@ -1,10 +1,18 @@
|
|
|
+//+build functional
|
|
|
+
|
|
|
package sarama
|
|
|
|
|
|
import (
|
|
|
+ "context"
|
|
|
+ "fmt"
|
|
|
+ "io"
|
|
|
"log"
|
|
|
- "math/rand"
|
|
|
"net"
|
|
|
+ "net/http"
|
|
|
+ "net/url"
|
|
|
"os"
|
|
|
+ "os/exec"
|
|
|
+ "path/filepath"
|
|
|
"strconv"
|
|
|
"strings"
|
|
|
"testing"
|
|
|
@@ -14,66 +22,373 @@ import (
|
|
|
)
|
|
|
|
|
|
const (
|
|
|
- VagrantToxiproxy = "http://192.168.100.67:8474"
|
|
|
- VagrantKafkaPeers = "192.168.100.67:9091,192.168.100.67:9092,192.168.100.67:9093,192.168.100.67:9094,192.168.100.67:9095"
|
|
|
- VagrantZookeeperPeers = "192.168.100.67:2181,192.168.100.67:2182,192.168.100.67:2183,192.168.100.67:2184,192.168.100.67:2185"
|
|
|
+ uncomittedMsgJar = "https://github.com/FrancoisPoinsot/simplest-uncommitted-msg/releases/download/0.1/simplest-uncommitted-msg-0.1-jar-with-dependencies.jar"
|
|
|
)
|
|
|
|
|
|
var (
|
|
|
- kafkaAvailable, kafkaRequired bool
|
|
|
- kafkaBrokers []string
|
|
|
+ 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,
|
|
|
+ },
|
|
|
+ }
|
|
|
|
|
|
- proxyClient *toxiproxy.Client
|
|
|
- Proxies map[string]*toxiproxy.Proxy
|
|
|
+ FunctionalTestEnv *testEnvironment
|
|
|
)
|
|
|
|
|
|
-func init() {
|
|
|
+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)
|
|
|
}
|
|
|
|
|
|
- seed := time.Now().UTC().UnixNano()
|
|
|
- if tmp := os.Getenv("TEST_SEED"); tmp != "" {
|
|
|
- seed, _ = strconv.ParseInt(tmp, 0, 64)
|
|
|
+ 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)
|
|
|
}
|
|
|
- Logger.Println("Using random seed:", seed)
|
|
|
- rand.Seed(seed)
|
|
|
|
|
|
- proxyAddr := os.Getenv("TOXIPROXY_ADDR")
|
|
|
- if proxyAddr == "" {
|
|
|
- proxyAddr = VagrantToxiproxy
|
|
|
+ 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"
|
|
|
}
|
|
|
- proxyClient = toxiproxy.NewClient(proxyAddr)
|
|
|
|
|
|
- kafkaPeers := os.Getenv("KAFKA_PEERS")
|
|
|
- if kafkaPeers == "" {
|
|
|
- kafkaPeers = VagrantKafkaPeers
|
|
|
+ // 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))
|
|
|
}
|
|
|
- kafkaBrokers = strings.Split(kafkaPeers, ",")
|
|
|
|
|
|
- if c, err := net.DialTimeout("tcp", kafkaBrokers[0], 5*time.Second); err == nil {
|
|
|
- if err = c.Close(); err == nil {
|
|
|
- kafkaAvailable = true
|
|
|
+ // 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
|
|
|
+ }
|
|
|
+ err = client.RefreshMetadata()
|
|
|
+ if err != nil {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ brokers := client.Brokers()
|
|
|
+ if len(brokers) < 5 {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ for _, broker := range brokers {
|
|
|
+ err := broker.Open(client.Config())
|
|
|
+ if err != nil {
|
|
|
+ continue retryLoop
|
|
|
+ }
|
|
|
+ connected, err := broker.Connected()
|
|
|
+ if err != nil || !connected {
|
|
|
+ continue retryLoop
|
|
|
+ }
|
|
|
+ }
|
|
|
+ brokersOk[j] = true
|
|
|
+ }
|
|
|
+ allBrokersUp = true
|
|
|
+ for _, u := range brokersOk {
|
|
|
+ allBrokersUp = allBrokersUp && u
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if !allBrokersUp {
|
|
|
+ return fmt.Errorf("timed out waiting for broker to come up")
|
|
|
}
|
|
|
|
|
|
- kafkaRequired = os.Getenv("CI") != ""
|
|
|
+ return nil
|
|
|
}
|
|
|
|
|
|
-func checkKafkaAvailability(t testing.TB) {
|
|
|
- if !kafkaAvailable {
|
|
|
- if kafkaRequired {
|
|
|
- t.Fatalf("Kafka broker is not available on %s. Set KAFKA_PEERS to connect to Kafka on a different location.", kafkaBrokers[0])
|
|
|
- } else {
|
|
|
- t.Skipf("Kafka broker is not available on %s. Set KAFKA_PEERS to connect to Kafka on a different location.", kafkaBrokers[0])
|
|
|
+func existingEnvironment(ctx context.Context, env *testEnvironment) (bool, error) {
|
|
|
+ toxiproxyAddr, ok := os.LookupEnv("TOXIPROXY_ADDR")
|
|
|
+ if !ok {
|
|
|
+ return false, nil
|
|
|
+ }
|
|
|
+ toxiproxyURL, err := url.Parse(toxiproxyAddr)
|
|
|
+ if err != nil {
|
|
|
+ return false, fmt.Errorf("$TOXIPROXY_ADDR not parseable as url")
|
|
|
+ }
|
|
|
+ toxiproxyHost := toxiproxyURL.Hostname()
|
|
|
+
|
|
|
+ env.ToxiproxyClient = toxiproxy.NewClient(toxiproxyAddr)
|
|
|
+ for i := 1; i <= 5; i++ {
|
|
|
+ proxyName := fmt.Sprintf("kafka%d", i)
|
|
|
+ proxy, err := env.ToxiproxyClient.Proxy(proxyName)
|
|
|
+ if err != nil {
|
|
|
+ return false, fmt.Errorf("no proxy kafka%d on toxiproxy: %w", i, err)
|
|
|
+ }
|
|
|
+ env.Proxies[proxyName] = proxy
|
|
|
+ // get the host:port from the proxy & toxiproxy addr, so we can do "$toxiproxy_addr:$proxy_port"
|
|
|
+ _, proxyPort, err := net.SplitHostPort(proxy.Listen)
|
|
|
+ if err != nil {
|
|
|
+ return false, fmt.Errorf("proxy.Listen not a host:port combo: %w", err)
|
|
|
}
|
|
|
+ env.KafkaBrokerAddrs = append(env.KafkaBrokerAddrs, fmt.Sprintf("%s:%s", toxiproxyHost, proxyPort))
|
|
|
+ }
|
|
|
+
|
|
|
+ env.KafkaVersion, ok = os.LookupEnv("KAFKA_VERSION")
|
|
|
+ if !ok {
|
|
|
+ return false, fmt.Errorf("KAFKA_VERSION needs to be provided with TOXIPROXY_ADDR")
|
|
|
}
|
|
|
+ return true, nil
|
|
|
+}
|
|
|
+
|
|
|
+func tearDownDockerTestEnvironment(ctx context.Context, env *testEnvironment) error {
|
|
|
+ c := exec.Command("docker-compose", "down", "--volumes")
|
|
|
+ c.Stdout = os.Stdout
|
|
|
+ c.Stderr = os.Stderr
|
|
|
+ downErr := c.Run()
|
|
|
+
|
|
|
+ c = exec.Command("docker-compose", "rm", "-v", "--force", "--stop")
|
|
|
+ c.Stdout = os.Stdout
|
|
|
+ c.Stderr = os.Stderr
|
|
|
+ rmErr := c.Run()
|
|
|
+ if downErr != nil {
|
|
|
+ return fmt.Errorf("failed to run docker-compose to stop test enviroment: %w", downErr)
|
|
|
+ }
|
|
|
+ if rmErr != nil {
|
|
|
+ return fmt.Errorf("failed to run docker-compose to rm test enviroment: %w", rmErr)
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+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 := NewConfig()
|
|
|
+ config.Metadata.Retry.Max = 5
|
|
|
+ config.Metadata.Retry.Backoff = 10 * time.Second
|
|
|
+ config.ClientID = "sarama-tests"
|
|
|
+ var err error
|
|
|
+ config.Version, err = ParseKafkaVersion(env.KafkaVersion)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed to parse kafka version %s: %w", env.KafkaVersion, err)
|
|
|
+ }
|
|
|
+
|
|
|
+ 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)
|
|
|
+ deleteRes, err := controller.DeleteTopics(&DeleteTopicsRequest{
|
|
|
+ Topics: testTopicNames,
|
|
|
+ Timeout: 30 * time.Second,
|
|
|
+ })
|
|
|
+ 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
|
|
|
+ for i := 0; i < 20 && !topicsOk; i++ {
|
|
|
+ time.Sleep(1 * time.Second)
|
|
|
+ md, err := controller.GetMetadata(&MetadataRequest{
|
|
|
+ Topics: testTopicNames,
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed to get metadata for test topics: %w", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ topicsOk = true
|
|
|
+ for _, topicsMd := range md.Topics {
|
|
|
+ if !isTopicNotExistsErrorOrOk(topicsMd.Err) {
|
|
|
+ topicsOk = false
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if !topicsOk {
|
|
|
+ return fmt.Errorf("timed out waiting for test topics to be gone")
|
|
|
+ }
|
|
|
+
|
|
|
+ // now create the topics empty
|
|
|
+ createRes, err := controller.CreateTopics(&CreateTopicsRequest{
|
|
|
+ TopicDetails: testTopicDetails,
|
|
|
+ Timeout: 30 * time.Second,
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed to create test topics: %w", err)
|
|
|
+ }
|
|
|
+ for topic, topicErr := range createRes.TopicErrors {
|
|
|
+ if !isTopicExistsErrorOrOk(topicErr.Err) {
|
|
|
+ return fmt.Errorf("failed to create test topic %s: %w", topic, topicErr)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // This is kind of gross, but we don't actually have support for doing transactional publishing
|
|
|
+ // with sarama, so we need to use a java-based tool to publish uncomitted messages to
|
|
|
+ // the uncommitted-topic-test-4 topic
|
|
|
+ jarName := filepath.Base(uncomittedMsgJar)
|
|
|
+ if _, err := os.Stat(jarName); err != nil {
|
|
|
+ Logger.Printf("Downloading %s\n", uncomittedMsgJar)
|
|
|
+ req, err := http.NewRequest("GET", uncomittedMsgJar, nil)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed creating requst for uncomitted msg jar: %w", err)
|
|
|
+ }
|
|
|
+ res, err := http.DefaultClient.Do(req)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed fetching the uncommitted msg jar: %w", err)
|
|
|
+ }
|
|
|
+ defer res.Body.Close()
|
|
|
+ jarFile, err := os.OpenFile(jarName, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed opening the uncomitted msg jar: %w", err)
|
|
|
+ }
|
|
|
+ defer jarFile.Close()
|
|
|
+
|
|
|
+ _, err = io.Copy(jarFile, res.Body)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed writing the uncomitted msg jar: %w", err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ c := exec.Command("java", "-jar", jarName, "-b", env.KafkaBrokerAddrs[0], "-c", "4")
|
|
|
+ c.Stdout = os.Stdout
|
|
|
+ c.Stderr = os.Stderr
|
|
|
+ err = c.Run()
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("failed running uncomitted msg jar: %w", err)
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+func isTopicNotExistsErrorOrOk(err KError) bool {
|
|
|
+ return err == ErrUnknownTopicOrPartition || err == ErrInvalidTopic || err == ErrNoError
|
|
|
+}
|
|
|
+
|
|
|
+func isTopicExistsErrorOrOk(err KError) bool {
|
|
|
+ return err == ErrTopicAlreadyExists || err == ErrNoError
|
|
|
}
|
|
|
|
|
|
func checkKafkaVersion(t testing.TB, requiredVersion string) {
|
|
|
- kafkaVersion := os.Getenv("KAFKA_VERSION")
|
|
|
+ kafkaVersion := FunctionalTestEnv.KafkaVersion
|
|
|
if kafkaVersion == "" {
|
|
|
- t.Logf("No KAFKA_VERSION set. This test requires Kafka version %s or higher. Continuing...", requiredVersion)
|
|
|
+ t.Skipf("No KAFKA_VERSION set. This test requires Kafka version %s or higher. Continuing...", requiredVersion)
|
|
|
} else {
|
|
|
available := parseKafkaVersion(kafkaVersion)
|
|
|
required := parseKafkaVersion(requiredVersion)
|
|
|
@@ -84,30 +399,19 @@ func checkKafkaVersion(t testing.TB, requiredVersion string) {
|
|
|
}
|
|
|
|
|
|
func resetProxies(t testing.TB) {
|
|
|
- if err := proxyClient.ResetState(); err != nil {
|
|
|
+ if err := FunctionalTestEnv.ToxiproxyClient.ResetState(); err != nil {
|
|
|
t.Error(err)
|
|
|
}
|
|
|
- Proxies = nil
|
|
|
-}
|
|
|
-
|
|
|
-func fetchProxies(t testing.TB) {
|
|
|
- var err error
|
|
|
- Proxies, err = proxyClient.Proxies()
|
|
|
- if err != nil {
|
|
|
- t.Fatal(err)
|
|
|
- }
|
|
|
}
|
|
|
|
|
|
func SaveProxy(t *testing.T, px string) {
|
|
|
- if err := Proxies[px].Save(); err != nil {
|
|
|
+ if err := FunctionalTestEnv.Proxies[px].Save(); err != nil {
|
|
|
t.Fatal(err)
|
|
|
}
|
|
|
}
|
|
|
|
|
|
func setupFunctionalTest(t testing.TB) {
|
|
|
- checkKafkaAvailability(t)
|
|
|
resetProxies(t)
|
|
|
- fetchProxies(t)
|
|
|
}
|
|
|
|
|
|
func teardownFunctionalTest(t testing.TB) {
|
Vlad Gorodetsky