Increase codecov of pkg/kafka

cmd/ingester/app/flags_test.go

@@ -43,6 +43,7 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--kafka.consumer.protocol-version=1.0.0",
 		"--ingester.parallelism=5",
 		"--ingester.deadlockInterval=2m",
+		"--kafka.consumer.authentication=tls",
 	})
 	o.InitFromViper(v)
 
@@ -56,6 +57,7 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, 5, o.Parallelism)
 	assert.Equal(t, 2*time.Minute, o.DeadlockInterval)
 	assert.Equal(t, kafka.EncodingJSON, o.Encoding)
+	assert.Equal(t, "tls", o.AuthenticationConfig.Authentication)
 }
 
 func TestTLSFlags(t *testing.T) {

internal/tracegen/worker_test.go

@@ -16,30 +16,63 @@ import (
 )
 
 func Test_SimulateTraces(t *testing.T) {
-	logger, buf := testutils.NewLogger()
-	tp := sdktrace.NewTracerProvider()
-	tracers := []trace.Tracer{tp.Tracer("stdout")}
-	wg := sync.WaitGroup{}
-	wg.Add(1)
-	var running uint32 = 1
-
-	worker := &worker{
-		logger:  logger,
-		tracers: tracers,
-		wg:      &wg,
-		id:      7,
-		running: &running,
-		Config: Config{
-			Traces:   7,
-			Duration: time.Second,
-			Pause:    time.Second,
-			Service:  "stdout",
-			Debug:    true,
-			Firehose: true,
-		},
-	}
-	expectedOutput := `{"level":"info","msg":"Worker 7 generated 7 traces"}` + "\n"
-
-	worker.simulateTraces()
-	assert.Equal(t, expectedOutput, buf.String())
+	// First test case with Pause = time.Second
+	t.Run("Pause time.Second", func(t *testing.T) {
+		logger, buf := testutils.NewLogger()
+		tp := sdktrace.NewTracerProvider()
+		tracers := []trace.Tracer{tp.Tracer("stdout")}
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		var running uint32 = 1
+
+		worker := &worker{
+			logger:  logger,
+			tracers: tracers,
+			wg:      &wg,
+			id:      7,
+			running: &running,
+			Config: Config{
+				Traces:   7,
+				Duration: time.Second,
+				Pause:    time.Second,
+				Service:  "stdout",
+				Debug:    true,
+				Firehose: true,
+			},
+		}
+		expectedOutput := `{"level":"info","msg":"Worker 7 generated 7 traces"}` + "\n"
+
+		worker.simulateTraces()
+		assert.Equal(t, expectedOutput, buf.String())
+	})
+
+	// Second test case with Pause = 3 * time.Second
+	t.Run("Pause 3 * time.Second", func(t *testing.T) {
+		logger, buf := testutils.NewLogger()
+		tp := sdktrace.NewTracerProvider()
+		tracers := []trace.Tracer{tp.Tracer("stdout")}
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		var running uint32 = 1
+
+		worker := &worker{
+			logger:  logger,
+			tracers: tracers,
+			wg:      &wg,
+			id:      7,
+			running: &running,
+			Config: Config{
+				Traces:   7,
+				Duration: time.Second,
+				Pause:    3 * time.Second,
+				Service:  "stdout",
+				Debug:    true,
+				Firehose: true,
+			},
+		}
+		expectedOutput := `{"level":"info","msg":"Worker 7 generated 7 traces"}` + "\n"
+
+		worker.simulateTraces()
+		assert.Equal(t, expectedOutput, buf.String())
+	})
 }

pkg/kafka/auth/config_test.txt

@@ -0,0 +1,87 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package auth
+
+import (
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/spf13/viper"
+	//"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	//"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
+)
+
+
+
+func TestSetConfiguration(t *testing.T) {
+	logger, _ := zap.NewDevelopment()
+	saramaConfig := sarama.NewConfig()
+	v := viper.New()
+	configPrefix := "kafka.auth."
+	v.Set(configPrefix+suffixAuthentication, "kerberos")
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosServiceName, "kafka")
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosRealm, "EXAMPLE.COM")
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosUseKeyTab, true)
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosUsername, "user")
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosPassword, "password")
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosConfig, "/path/to/krb5.conf")
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosKeyTab, "/path/to/keytab")
+	v.Set(configPrefix+kerberosPrefix+suffixKerberosDisablePAFXFAST, true)
+	v.Set(configPrefix+plainTextPrefix+suffixPlainTextUsername, "user")
+	v.Set(configPrefix+plainTextPrefix+suffixPlainTextPassword, "password")
+	v.Set(configPrefix+plainTextPrefix+suffixPlainTextMechanism, "SCRAM-SHA-256")
+
+	authConfig := &AuthenticationConfig{}
+	err := authConfig.InitFromViper(configPrefix, v)
+	require.NoError(t, err)
+	require.NoError(t, authConfig.SetConfiguration(saramaConfig, logger))
+	// plaintest test(SCRAM-SHA-256)
+	v.Set(configPrefix+suffixAuthentication, "plaintext")
+	authConfig = &AuthenticationConfig{}
+	err = authConfig.InitFromViper(configPrefix, v)
+	require.NoError(t, err)
+	require.NoError(t, authConfig.SetConfiguration(saramaConfig, logger))
+	// plaintest test(SCRAM-SHA-512)
+	v.Set(configPrefix+plainTextPrefix+suffixPlainTextMechanism, "SCRAM-SHA-512")
+	authConfig = &AuthenticationConfig{}
+	err = authConfig.InitFromViper(configPrefix, v)
+	require.NoError(t, err)
+	require.NoError(t, authConfig.SetConfiguration(saramaConfig, logger))
+	// plaintest test(PLAIN)
+	v.Set(configPrefix+plainTextPrefix+suffixPlainTextMechanism, "PLAIN")
+	authConfig = &AuthenticationConfig{}
+	err = authConfig.InitFromViper(configPrefix, v)
+	require.NoError(t, err)
+	require.NoError(t, authConfig.SetConfiguration(saramaConfig, logger))
+	// default value
+	v.Set(configPrefix+suffixAuthentication, "none")
+	authConfig = &AuthenticationConfig{}
+	err = authConfig.InitFromViper(configPrefix, v)
+	require.NoError(t, err)
+	require.Error(t, authConfig.SetConfiguration(saramaConfig, logger), nil)
+	// none value
+	v.Set(configPrefix+suffixAuthentication, " ")
+	authConfig = &AuthenticationConfig{}
+	err = authConfig.InitFromViper(configPrefix, v)
+	require.NoError(t, err)
+	require.NoError(t, authConfig.SetConfiguration(saramaConfig, logger))
+	// tls value
+	// v.Set(configPrefix + suffixAuthentication, "tls")
+	// authConfig = &AuthenticationConfig{}
+	// err = authConfig.InitFromViper(configPrefix, v)
+	// require.NoError(t, err)
+	// _,err=authConfig.TLS.Config(zap.NewNop())
+	// // require.NoError(t, err)
+	// // require.NoError(t,authConfig.SetConfiguration(saramaConfig, logger))
+	// default value
+	v.Set(configPrefix+suffixAuthentication, "error")
+	authConfig = &AuthenticationConfig{}
+	err = authConfig.InitFromViper(configPrefix, v)
+	require.NoError(t, err)
+	expected_error := "Unknown/Unsupported authentication method error to kafka cluster"
+	require.Error(t, authConfig.SetConfiguration(saramaConfig, logger), expected_error)
+}

plugin/storage/integration/kafka_test.go

@@ -82,6 +82,7 @@ func (s *KafkaIntegrationTestSuite) initialize(t *testing.T) {
 		clientID,
 		"--ingester.parallelism",
 		"1000",
+		"--kafka.consumer.authentication=plaintext",
 	})
 	require.NoError(t, err)
 	options := app.Options{