remove kafka globals

wundergraph · Feb 11, 2025 · 5c8e500 · 5c8e500
1 parent 4f8f17e
commit 5c8e500
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 36 deletions.
diff --git a/router-tests/events/kafka_events_test.go b/router-tests/events/kafka_events_test.go
@@ -54,7 +54,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			var subscriptionOne struct {
@@ -115,7 +114,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			var subscriptionOne struct {
@@ -201,7 +199,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			var subscriptionOne struct {
@@ -268,7 +265,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			var subscriptionOne struct {
@@ -364,7 +360,6 @@ func TestKafkaEvents(t *testing.T) {
 				engineExecutionConfiguration.WebSocketClientReadTimeout = time.Millisecond * 100
 			},
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			var subscriptionOne struct {
@@ -419,7 +414,7 @@ func TestKafkaEvents(t *testing.T) {
 	t.Run("multipart", func(t *testing.T) {
 		t.Parallel()
 
-		var multipartHeartbeatInterval = time.Second * 5
+		multipartHeartbeatInterval := time.Second * 5
 
 		t.Run("subscribe sync", func(t *testing.T) {
 			t.Parallel()
@@ -433,7 +428,6 @@ func TestKafkaEvents(t *testing.T) {
 					core.WithMultipartHeartbeatInterval(multipartHeartbeatInterval),
 				},
 			}, func(t *testing.T, xEnv *testenv.Environment) {
-
 				ensureTopicExists(t, xEnv, topics...)
 
 				subscribePayload := []byte(`{"query":"subscription { employeeUpdatedMyKafka(employeeID: 1) { id details { forename surname } }}"}`)
@@ -529,7 +523,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			subscribePayload := []byte(`{"query":"subscription { employeeUpdatedMyKafka(employeeID: 1) { id details { forename surname } }}"}`)
@@ -565,7 +558,6 @@ func TestKafkaEvents(t *testing.T) {
 				line, _, gErr := reader.ReadLine()
 				require.NoError(t, gErr)
 				require.Equal(t, "", string(line))
-
 			}()
 
 			xEnv.WaitForSubscriptionCount(1, KafkaWaitTimeout)
@@ -590,7 +582,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			subscribePayload := []byte(`{"query":"subscription { employeeUpdatedMyKafka(employeeID: 1) { id details { forename surname } }}"}`)
@@ -694,7 +685,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			type subscriptionPayload struct {
@@ -808,7 +798,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			type subscriptionPayload struct {
@@ -922,7 +911,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			type subscriptionPayload struct {
@@ -990,7 +978,6 @@ func TestKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			var subscriptionOne struct {
@@ -1075,7 +1062,6 @@ func TestFlakyKafkaEvents(t *testing.T) {
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
 			EnableKafka:              true,
 		}, func(t *testing.T, xEnv *testenv.Environment) {
-
 			ensureTopicExists(t, xEnv, topics...)
 
 			type subscriptionPayload struct {

diff --git a/router-tests/testenv/pubsub.go b/router-tests/testenv/pubsub.go
@@ -22,11 +22,6 @@ import (
 	nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1"
 )
 
-var (
-	kafkaMux  sync.Mutex
-	kafkaData *KafkaData
-)
-
 type KafkaData struct {
 	Brokers []string
 }
@@ -44,14 +39,7 @@ func getHostPort(resource *dockertest.Resource, id string) string {
 }
 
 func setupKafkaServers(t testing.TB) (*KafkaData, error) {
-	kafkaMux.Lock()
-	defer kafkaMux.Unlock()
-
-	if kafkaData != nil {
-		return kafkaData, nil
-	}
-
-	kafkaData = &KafkaData{}
+	kafkaData := &KafkaData{}
 
 	dockerPool, err := dockertest.NewPool("")
 	require.NoError(t, err, "could not connect to docker")

diff --git a/router-tests/testenv/testenv.go b/router-tests/testenv/testenv.go
@@ -155,7 +155,6 @@ func Bench(b *testing.B, cfg *Config, f func(b *testing.B, xEnv *Environment)) {
 			assertCacheMetrics(b, env, v, ff)
 		}
 	}
-
 }
 
 const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
@@ -407,9 +406,7 @@ func createTestEnv(t testing.TB, cfg *Config) (*Environment, error) {
 
 	ctx, cancel := context.WithCancelCause(context.Background())
 
-	var (
-		logObserver *observer.ObservedLogs
-	)
+	var logObserver *observer.ObservedLogs
 
 	if oc := cfg.LogObservation; oc.Enabled {
 		var zCore zapcore.Core
@@ -441,9 +438,7 @@ func createTestEnv(t testing.TB, cfg *Config) (*Environment, error) {
 		Countries:    atomic.NewInt64(0),
 	}
 
-	var (
-		requiredPorts = 2
-	)
+	requiredPorts := 2
 
 	ports := freeport.GetN(t, requiredPorts)
 
@@ -1663,7 +1658,6 @@ func (e *Environment) ReadSSE(ctx context.Context, body io.ReadCloser, handler f
 				handler(data)
 			}
 		}
-
 	}
 }