Merge pull request #2200 from Shopify/dnwe/test-timeout-param

chore: bump functional test timeout to 12m

.github/workflows/ci.yml

@@ -1,41 +1,47 @@
 ---
 name: CI
-
 on: [push, pull_request]
-
 jobs:
+  lint:
+    name: Linting with Go ${{ matrix.go-version }}
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        go-version: [1.17.x, 1.18.x]
+    steps:
+    - uses: actions/checkout@v2
+    - name: Setup Go
+      uses: actions/setup-go@v3
+      with:
+        go-version: ${{ matrix.go-version }}
+    - name: golangci-lint
+      env:
+        GOFLAGS: -tags=functional
+      uses: golangci/golangci-lint-action@v2
+      with:
+        version: v1.45.2
   test:
-    name: Go ${{ matrix.go-version }} with Kafka ${{ matrix.kafka-version }}
+    name: Unit Testing with Go ${{ matrix.go-version }}
     runs-on: ubuntu-latest
-    # We want to run on external PRs, but not on our own internal PRs as they'll be run
-    # by the push to the branch.
-    if: github.event_name == 'push' || github.event.pull_request.head.repo.full_name != github.repository
     strategy:
       fail-fast: false
       matrix:
         go-version: [1.17.x, 1.18.x]
-        kafka-version: [2.8.1, 3.0.1, 3.1.0]
-        platform: [ubuntu-latest]
-
     env:
       DEBUG: true
       GOFLAGS: -trimpath
-      KAFKA_VERSION: ${{ matrix.kafka-version }}
-
     steps:
     - uses: actions/checkout@v2
-
     - name: Setup Go
       uses: actions/setup-go@v3
       with:
         go-version: ${{ matrix.go-version }}
-
     - name: Get Go environment
       id: go-env
       run: |
         echo "::set-output name=cache::$(go env GOCACHE)"
         echo "::set-output name=modcache::$(go env GOMODCACHE)"
-
     - name: Set up cache
       uses: actions/cache@v2
       with:
@@ -45,19 +51,5 @@ jobs:
         key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
         restore-keys: |
           ${{ runner.os }}-go-
-
-    - name: Lint
-      uses: golangci/golangci-lint-action@v2
-      if: ${{ matrix.go-version == '1.18.x' }}
-      env:
-        GOFLAGS: -tags=functional
-      with:
-        version: v1.45.0
-
     - name: Test (Unit)
       run: make test
-      if: ${{ matrix.go-version != '1.18.x' }}
-
-    - name: Test (Functional)
-      run: make test_functional
-      if: ${{ matrix.go-version == '1.18.x' }}

.github/workflows/fvt.yml

@@ -0,0 +1,42 @@
+---
+name: FVT
+on: [push, pull_request]
+jobs:
+  fvt:
+    name: Test with Kafka ${{ matrix.kafka-version }}
+    runs-on: ubuntu-latest
+
+    # We want to run on external PRs, but not on our own internal PRs as they'll be run
+    # by the push to the branch.
+    if: github.event_name == 'push' || github.event.pull_request.head.repo.full_name != github.repository
+    strategy:
+      fail-fast: false
+      matrix:
+        go-version: [1.18.x]
+        kafka-version: [2.8.1, 3.0.1, 3.1.0]
+    env:
+      DEBUG: true
+      GOFLAGS: -trimpath
+      KAFKA_VERSION: ${{ matrix.kafka-version }}
+    steps:
+    - uses: actions/checkout@v2
+    - name: Setup Go
+      uses: actions/setup-go@v3
+      with:
+        go-version: ${{ matrix.go-version }}
+    - name: Get Go environment
+      id: go-env
+      run: |
+        echo "::set-output name=cache::$(go env GOCACHE)"
+        echo "::set-output name=modcache::$(go env GOMODCACHE)"
+    - name: Set up cache
+      uses: actions/cache@v2
+      with:
+        path: |
+          ${{ steps.go-env.outputs.cache }}
+          ${{ steps.go-env.outputs.modcache }}
+        key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+        restore-keys: |
+          ${{ runner.os }}-go-
+    - name: Test (Functional)
+      run: make test_functional

Makefile

@@ -2,7 +2,7 @@ default: fmt get update test lint
 
 GO       := go
 GOBUILD  := CGO_ENABLED=0 $(GO) build $(BUILD_FLAG)
-GOTEST   := $(GO) test -v -race -timeout 10m -coverprofile=profile.out -covermode=atomic
+GOTEST   := $(GO) test -v -race -coverprofile=profile.out -covermode=atomic
 
 FILES    := $(shell find . -name '*.go' -type f -not -name '*.pb.go' -not -name '*_generated.go' -not -name '*_test.go')
 TESTS    := $(shell find . -name '*.go' -type f -not -name '*.pb.go' -not -name '*_generated.go' -name '*_test.go')
@@ -24,8 +24,8 @@ lint:
 	GOFLAGS="-tags=functional" golangci-lint run
 
 test:
-	$(GOTEST) ./...
+	$(GOTEST) -timeout 2m ./...
 
 .PHONY: test_functional
 test_functional:
-	$(GOTEST) -tags=functional ./...
+	$(GOTEST) -timeout 12m -tags=functional ./...

functional_test.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"net"
 	"net/url"
 	"os"
 	"os/exec"
@@ -122,6 +123,8 @@ func setupToxiProxies(env *testEnvironment, endpoint string) error {
 }
 
 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.
@@ -148,21 +151,42 @@ func prepareDockerTestEnvironment(ctx context.Context, env *testEnvironment) err
 		return fmt.Errorf("failed to setup toxiproxies: %w", err)
 	}
 
-	// Wait for the kafka broker to come up
-	allBrokersUp := false
-	for i := 0; i < 90 && !allBrokersUp; i++ {
-		Logger.Println("waiting for kafka brokers to come up")
-		time.Sleep(1 * time.Second)
-		config := NewTestConfig()
-		config.Version, err = ParseKafkaVersion(env.KafkaVersion)
+	dialCheck := func(addr string, timeout time.Duration) error {
+		conn, err := net.DialTimeout("tcp", addr, timeout)
 		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"
+		return conn.Close()
+	}
+
+	config := NewTestConfig()
+	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"
+
+	// wait for the kafka brokers to come up
+	allBrokersUp := false
+
+mainLoop:
+	for i := 0; i < 30 && !allBrokersUp; i++ {
+		Logger.Println("waiting for kafka brokers to come up")
+		time.Sleep(3 * time.Second)
 		brokersOk := make([]bool, len(env.KafkaBrokerAddrs))
+
+		// first check that all bootstrap brokers are TCP accessible
+		for _, addr := range env.KafkaBrokerAddrs {
+			if err := dialCheck(addr, time.Second); err != nil {
+				continue mainLoop
+			}
+		}
+
+		// now check we can bootstrap metadata from the cluster and all brokers
+		// are known and accessible at their advertised address
 	retryLoop:
 		for j, addr := range env.KafkaBrokerAddrs {
 			client, err := NewClient([]string{addr}, config)
@@ -174,7 +198,7 @@ func prepareDockerTestEnvironment(ctx context.Context, env *testEnvironment) err
 				continue
 			}
 			brokers := client.Brokers()
-			if len(brokers) < 5 {
+			if len(brokers) < expectedBrokers {
 				continue
 			}
 			for _, broker := range brokers {
@@ -189,17 +213,19 @@ func prepareDockerTestEnvironment(ctx context.Context, env *testEnvironment) err
 			}
 			brokersOk[j] = true
 		}
+
 		allBrokersUp = true
 		for _, u := range brokersOk {
 			allBrokersUp = allBrokersUp && u
 		}
 	}
+
 	if !allBrokersUp {
-		c := exec.Command("docker-compose", "logs", "-t")
+		c := exec.Command("docker-compose", "logs", "-t", "kafka-1", "kafka-2", "kafka-3", "kafka-4", "kafka-5")
 		c.Stdout = os.Stdout
 		c.Stderr = os.Stderr
 		_ = c.Run()
-		return fmt.Errorf("timed out waiting for broker to come up")
+		return fmt.Errorf("timed out waiting for one or more broker to come up")
 	}
 
 	return nil

utils.go

@@ -180,6 +180,7 @@ var (
 	V2_8_0_0  = newKafkaVersion(2, 8, 0, 0)
 	V2_8_1_0  = newKafkaVersion(2, 8, 1, 0)
 	V3_0_0_0  = newKafkaVersion(3, 0, 0, 0)
+	V3_0_1_0  = newKafkaVersion(3, 0, 1, 0)
 	V3_1_0_0  = newKafkaVersion(3, 1, 0, 0)
 
 	SupportedVersions = []KafkaVersion{
@@ -224,6 +225,7 @@ var (
 		V2_8_0_0,
 		V2_8_1_0,
 		V3_0_0_0,
+		V3_0_1_0,
 		V3_1_0_0,
 	}
 	MinVersion     = V0_8_2_0

utils_test.go

@@ -71,6 +71,7 @@ func TestVersionParsing(t *testing.T) {
 		"2.8.0",
 		"2.8.1",
 		"3.0.0",
+		"3.0.1",
 		"3.1.0",
 	}
 	for _, s := range validVersions {