Add go leak check to Cassandra and Kafka e2e tests (jaegertracing#6336)

## Which problem is this PR solving?
Fixes a part of jaegertracing#5006 
## Description of the changes
1) Cassandra factory is closed in the test manually after the test, not
doing so will fail the leak test probably due to CleanUp function is not
closing factory properly and returning in the test leading to end it
that further fails the leak test.
2) Kafka Producer and Consumer was not closed before leading to failure
of leak test which is fixed in this PR.

## How was this change tested?
- By running integartion tests manually

## Checklist
- [x] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [x] I have signed all commits
- [x] I have added unit tests for the new functionality
- [x] I have run lint and test steps successfully
  - for `jaeger`: `make lint test`
  - for `jaeger-ui`: `yarn lint` and `yarn test`

---------

Signed-off-by: Manik2708 <[email protected]>
Signed-off-by: Manik Mehta <[email protected]>
Co-authored-by: Yuri Shkuro <[email protected]>

pkg/testutils/leakcheck.go

@@ -23,6 +23,17 @@ func IgnoreOpenCensusWorkerLeak() goleak.Option {
 	return goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start")
 }
 
+// IgnoreGoMetricsMeterLeak prevents the leak created by go-metrics which is
+// used by Sarama (Kafka Client) in Jaeger v1. This reason of this leak is
+// not Jaeger but the go-metrics used by Samara.
+// See these issues for the context
+// - https://github.com/IBM/sarama/issues/1321
+// - https://github.com/IBM/sarama/issues/1340
+// - https://github.com/IBM/sarama/issues/2832
+func IgnoreGoMetricsMeterLeak() goleak.Option {
+	return goleak.IgnoreTopFunction("github.com/rcrowley/go-metrics.(*meterArbiter).tick")
+}
+
 // VerifyGoLeaks verifies that unit tests do not leak any goroutines.
 // It should be called in TestMain.
 func VerifyGoLeaks(m *testing.M) {
@@ -36,5 +47,5 @@ func VerifyGoLeaks(m *testing.M) {
 //
 //	defer testutils.VerifyGoLeaksOnce(t)
 func VerifyGoLeaksOnce(t *testing.T) {
-	goleak.VerifyNone(t, IgnoreGlogFlushDaemonLeak(), IgnoreOpenCensusWorkerLeak())
+	goleak.VerifyNone(t, IgnoreGlogFlushDaemonLeak(), IgnoreOpenCensusWorkerLeak(), IgnoreGoMetricsMeterLeak())
 }

plugin/storage/integration/cassandra_test.go

@@ -9,12 +9,14 @@ import (
 	"os"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
+	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/plugin/storage/cassandra"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
 )
@@ -47,6 +49,9 @@ func (*CassandraStorageIntegration) initializeCassandraFactory(t *testing.T, fla
 	require.NoError(t, command.ParseFlags(flags))
 	f.InitFromViper(v, logger)
 	require.NoError(t, f.Initialize(metrics.NullFactory, logger))
+	t.Cleanup(func() {
+		assert.NoError(t, f.Close())
+	})
 	return f
 }
 
@@ -78,9 +83,6 @@ func (s *CassandraStorageIntegration) initializeCassandra(t *testing.T) {
 	s.SamplingStore, err = f.CreateSamplingStore(0)
 	require.NoError(t, err)
 	s.initializeDependencyReaderAndWriter(t, f)
-	t.Cleanup(func() {
-		require.NoError(t, f.Close())
-	})
 }
 
 func (s *CassandraStorageIntegration) initializeDependencyReaderAndWriter(t *testing.T, f *cassandra.Factory) {
@@ -99,6 +101,9 @@ func (s *CassandraStorageIntegration) initializeDependencyReaderAndWriter(t *tes
 
 func TestCassandraStorage(t *testing.T) {
 	SkipUnlessEnv(t, "cassandra")
+	t.Cleanup(func() {
+		testutils.VerifyGoLeaksOnce(t)
+	})
 	s := newCassandraStorageIntegration()
 	s.initializeCassandra(t)
 	s.RunAll(t)

plugin/storage/integration/kafka_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
@@ -20,6 +21,7 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
+	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/plugin/storage/kafka"
 	"github.com/jaegertracing/jaeger/plugin/storage/memory"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
@@ -53,10 +55,11 @@ func (s *KafkaIntegrationTestSuite) initialize(t *testing.T) {
 	f.InitFromViper(v, logger)
 	err = f.Initialize(metrics.NullFactory, logger)
 	require.NoError(t, err)
-
+	t.Cleanup(func() {
+		assert.NoError(t, f.Close())
+	})
 	spanWriter, err := f.CreateSpanWriter()
 	require.NoError(t, err)
-
 	v, command = config.Viperize(app.AddFlags)
 	err = command.ParseFlags([]string{
 		"--kafka.consumer.topic",
@@ -82,6 +85,9 @@ func (s *KafkaIntegrationTestSuite) initialize(t *testing.T) {
 	traceStore := memory.NewStore()
 	spanConsumer, err := builder.CreateConsumer(logger, metrics.NullFactory, traceStore, options)
 	require.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, spanConsumer.Close())
+	})
 	spanConsumer.Start()
 
 	s.SpanWriter = spanWriter
@@ -120,6 +126,9 @@ func (*ingester) FindTraceIDs(context.Context, *spanstore.TraceQueryParameters)
 
 func TestKafkaStorage(t *testing.T) {
 	SkipUnlessEnv(t, "kafka")
+	t.Cleanup(func() {
+		testutils.VerifyGoLeaksOnce(t)
+	})
 	s := &KafkaIntegrationTestSuite{}
 	s.initialize(t)
 	t.Run("GetTrace", s.testGetTrace)