Repository: gofr-dev/gofr

Issue:
How to configure Gofr for Kafka with connection and security parameters
Hello!
Can you configure extra connection and security parameters in Gofr to use apache Kafka?

...
security.protocol=SASL_SSL
sasl.mechanisms=PLAIN
sasl.username={{ CLUSTER_API_KEY }}
sasl.password={{ CLUSTER_API_SECRET }}
...

With the Confluent ecosystem, Kafka SaaS

Thanks!

Discussion:
@Umang01-hash: @wfrankoti Hello!

Thank you for reaching out and for the detailed feature request.

Currently, GoFr's Kafka integration does not directly support the configuration of security parameters such as `security.protocol`, `sasl.mechanisms`, `sasl.username`, and `sasl.password`. The existing configuration options are limited to the basic settings outlined in our documentation.

However, we recognize the importance of these features for many use cases, and we plan to add support for them in a future release.

In the meantime, if you are interested in contributing to GoFr and pick this up, we would be more than happy to accept your contributions to implement this feature. Please let us know if you'd like to discuss this further!

We appreciate your understanding and patience.@wfrankoti: @Umang01-hash Hello!
Thanks for your invitation. 

I will contribute with comments, to speed up the change.

I have studied the used library segmentio/kafka-go, and I can see that it has the expected functionality to implement it in a new version of Gofr

From the segmentio repo, use
https://github.com/segmentio/kafka-go/tree/main?tab=readme-ov-file#sasl-support

[Plain](https://godoc.org/github.com/segmentio/kafka-go/sasl/plain#Mechanism)

mechanism := plain.Mechanism{
    Username: "username",
    Password: "password",
}

[SCRAM](https://godoc.org/github.com/segmentio/kafka-go/sasl/scram#Mechanism)

mechanism, err := scram.Mechanism(scram.SHA512, "username", "password")
if err != nil {
    panic(err)
}

and,

dialer := &kafka.Dialer{
    Timeout:       10 * time.Second,
    DualStack:     true,
    SASLMechanism: mechanism,
}

And add configuration and code in Gofr

I hope to be able to contribute soon by delivering code


Solution patch:
diff --git a/docs/references/configs/page.md b/docs/references/configs/page.md
index 3e982f3b..f74e697b 100644
--- a/docs/references/configs/page.md
+++ b/docs/references/configs/page.md
@@ -297,6 +297,25 @@ This document lists all the configuration options supported by the GoFr framewor
 -  Unique identifier for this consumer
 -  gofr-consumer
 
+---
+
+
+- KAFKA_SASL_MECHANISM
+- SASL mechanism for authentication (e.g. PLAIN, SCRAM-SHA-256, SCRAM-SHA-512)
+- None
+
+---
+
+- KAFKA_SASL_USERNAME
+- Username for SASL authentication
+- None
+
+---
+
+- KAFKA_SASL_PASSWORD
+- Password for SASL authentication
+- None
+
 {% /table %}
 
 **Google**
diff --git a/pkg/gofr/container/container.go b/pkg/gofr/container/container.go
index 2e02933b..cab2dc99 100644
--- a/pkg/gofr/container/container.go
+++ b/pkg/gofr/container/container.go
@@ -138,6 +138,9 @@ func (c *Container) Create(conf config.Config) {
 				BatchSize:       batchSize,
 				BatchBytes:      batchBytes,
 				BatchTimeout:    batchTimeout,
+				SASLMechanism:   conf.Get("KAFKA_SASL_MECHANISM"),
+				SASLUser:        conf.Get("KAFKA_SASL_USERNAME"),
+				SASLPassword:    conf.Get("KAFKA_SASL_PASSWORD"),
 			}, c.Logger, c.metricsManager)
 		}
 	case "GOOGLE":
diff --git a/pkg/gofr/datasource/pubsub/kafka/kafka.go b/pkg/gofr/datasource/pubsub/kafka/kafka.go
index 069f38e6..f49daeb8 100644
--- a/pkg/gofr/datasource/pubsub/kafka/kafka.go
+++ b/pkg/gofr/datasource/pubsub/kafka/kafka.go
@@ -22,6 +22,8 @@ var (
 	errBatchBytes               = errors.New("KAFKA_BATCH_BYTES must be greater than 0")
 	errBatchTimeout             = errors.New("KAFKA_BATCH_TIMEOUT must be greater than 0")
 	errClientNotConnected       = errors.New("kafka client not connected")
+	errUnsupportedSASLMechanism = errors.New("unsupported SASL mechanism")
+	errSASLCredentialsMissing   = errors.New("SASL credentials missing")
 )
 
 const (
@@ -40,6 +42,9 @@ type Config struct {
 	BatchBytes      int
 	BatchTimeout    int
 	RetryTimeout    time.Duration
+	SASLMechanism   string
+	SASLUser        string
+	SASLPassword    string
 }
 
 type kafkaClient struct {
@@ -112,6 +117,12 @@ func validateConfigs(conf *Config) error {
 		return errBatchTimeout
 	}
 
+	if conf.SASLMechanism != "" {
+		if conf.SASLUser == "" || conf.SASLPassword == "" {
+			return errSASLCredentialsMissing
+		}
+	}
+
 	return nil
 }
 
@@ -245,7 +256,16 @@ func initializeKafkaClient(conf *Config, logger pubsub.Logger) (*kafka.Dialer, C
 		DualStack: true,
 	}
 
-	conn, err := kafka.Dial("tcp", conf.Broker)
+	if conf.SASLMechanism != "" {
+		mechanism, err := getSASLMechanism(conf.SASLMechanism, conf.SASLUser, conf.SASLPassword)
+		if err != nil {
+			return nil, nil, nil, nil, err
+		}
+
+		dialer.SASLMechanism = mechanism
+	}
+
+	conn, err := dialer.DialContext(context.Background(), "tcp", conf.Broker)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}
diff --git a/pkg/gofr/datasource/pubsub/kafka/kafka_sasl.go b/pkg/gofr/datasource/pubsub/kafka/kafka_sasl.go
new file mode 100644
index 00000000..ee011129
--- /dev/null
+++ b/pkg/gofr/datasource/pubsub/kafka/kafka_sasl.go
@@ -0,0 +1,30 @@
+package kafka
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/segmentio/kafka-go/sasl"
+	"github.com/segmentio/kafka-go/sasl/plain"
+	"github.com/segmentio/kafka-go/sasl/scram"
+)
+
+func getSASLMechanism(mechanism, username, password string) (sasl.Mechanism, error) {
+	switch strings.ToUpper(mechanism) {
+	case "PLAIN":
+		return plain.Mechanism{
+			Username: username,
+			Password: password,
+		}, nil
+	case "SCRAM-SHA-256":
+		mechanism, _ := scram.Mechanism(scram.SHA256, username, password)
+
+		return mechanism, nil
+	case "SCRAM-SHA-512":
+		mechanism, _ := scram.Mechanism(scram.SHA512, username, password)
+
+		return mechanism, nil
+	default:
+		return nil, fmt.Errorf("%w: %s", errUnsupportedSASLMechanism, mechanism)
+	}
+}



Test patch:
diff --git a/pkg/gofr/datasource/pubsub/kafka/kafka_sasl_test.go b/pkg/gofr/datasource/pubsub/kafka/kafka_sasl_test.go
new file mode 100644
index 00000000..bd1c6476
--- /dev/null
+++ b/pkg/gofr/datasource/pubsub/kafka/kafka_sasl_test.go
@@ -0,0 +1,63 @@
+package kafka
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetSASLMechanism_Success(t *testing.T) {
+	tests := []struct {
+		name         string
+		mechanism    string
+		username     string
+		password     string
+		expectedName string
+	}{
+		{
+			name:         "PLAIN uppercase",
+			mechanism:    "PLAIN",
+			username:     "user",
+			password:     "pass",
+			expectedName: "PLAIN",
+		},
+		{
+			name:         "plain lowercase",
+			mechanism:    "plain",
+			username:     "user",
+			password:     "pass",
+			expectedName: "PLAIN",
+		},
+		{
+			name:         "SCRAM-SHA-256",
+			mechanism:    "SCRAM-SHA-256",
+			username:     "user",
+			password:     "pass",
+			expectedName: "SCRAM-SHA-256",
+		},
+		{
+			name:         "SCRAM-SHA-512",
+			mechanism:    "SCRAM-SHA-512",
+			username:     "user",
+			password:     "pass",
+			expectedName: "SCRAM-SHA-512",
+		},
+	}
+
+	for _, tc := range tests {
+		mechanism, err := getSASLMechanism(tc.mechanism, tc.username, tc.password)
+
+		require.NoError(t, err, "unexpected error: %v", err)
+		assert.Equal(t, tc.expectedName, mechanism.Name(),
+			"expected mechanism name %q, got %q", tc.expectedName, mechanism.Name())
+	}
+}
+
+func TestGetSASLMechanism_Failure(t *testing.T) {
+	_, err := getSASLMechanism("FOO", "user", "pass")
+
+	require.Error(t, err, "expected an error for unsupported mechanism but got none")
+	assert.Contains(t, err.Error(), "unsupported SASL mechanism",
+		"unexpected error message: %v", err)
+}
diff --git a/pkg/gofr/datasource/pubsub/kafka/kafka_test.go b/pkg/gofr/datasource/pubsub/kafka/kafka_test.go
index 28934eff..a6c8cd77 100644
--- a/pkg/gofr/datasource/pubsub/kafka/kafka_test.go
+++ b/pkg/gofr/datasource/pubsub/kafka/kafka_test.go
@@ -23,8 +23,9 @@ func TestValidateConfigs(t *testing.T) {
 		expected error
 	}{
 		{
-			name:     "Valid Config",
-			config:   Config{Broker: "kafkabroker", BatchSize: 1, BatchBytes: 1, BatchTimeout: 1},
+			name: "Valid Config",
+			config: Config{Broker: "kafkabroker", BatchSize: 1, BatchBytes: 1, BatchTimeout: 1,
+				SASLMechanism: "PLAIN", SASLUser: "user", SASLPassword: "password"},
 			expected: nil,
 		},
 		{
@@ -47,6 +48,12 @@ func TestValidateConfigs(t *testing.T) {
 			config:   Config{Broker: "kafkabroker", BatchSize: 1, BatchBytes: 1, BatchTimeout: 0},
 			expected: errBatchTimeout,
 		},
+		{
+			name: "Empty SASL Password",
+			config: Config{Broker: "kafkabroker", BatchSize: 1, BatchBytes: 1, BatchTimeout: 1,
+				SASLMechanism: "PLAIN", SASLUser: "user"},
+			expected: errSASLCredentialsMissing,
+		},
 	}
 
 	for _, tc := range testCases {