Repository: deviceinsight/kafkactl (go)

Issue:
Producer configuration
It would be great if we could add a `producer` property to the `config.yml` file so we can configure sarama client. The object could be retrieved in [CreateClientConfig](https://github.com/deviceinsight/kafkactl/blob/0e7910b1e416dab11ed454c0b510ac2a9ca78017/internal/common-operation.go#L136) for generating [sarama.Config](https://github.com/Shopify/sarama/blob/main/config.go).

I can provide a PR.


Pull request description:
Feature/refactor producer configs
# Description

make `maxMessageBytes` configurable in produce command

Fixes #112

## Type of change

Please delete options that are not relevant.
- [x] Breaking change (fix or feature that would cause existing functionality to not work as expected)

## Documentation

- [x] the change is mentioned in the `## [Unreleased]` section of `CHANGELOG.md`
- [x] the configuration yaml was changed and the example config in `README.md` was updated
- [ ] a usage example was added to `README.md`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c66a76c..e7cc237 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- [#112](https://github.com/deviceinsight/kafkactl/issues/112) make `maxMessageBytes` configurable in produce command.
+  :warning: this is a breaking change since the format of the `config.yml` has been changed in order to group producer
+  related configurations under `context.producer`.
+
 ## 1.24.0 - 2021-12-03
 
 ### Added
diff --git a/README.md b/README.md
index e531b9b..9184791 100644
--- a/README.md
+++ b/README.md
@@ -130,13 +130,16 @@ contexts:
        protosetFiles:
           - "/usr/include/protoset/other.protoset"
     
-    # optional: changes the default partitioner
-    defaultPartitioner: "hash"
+    producer:
+      # optional: changes the default partitioner
+      partitioner: "hash"
 
-    # optional: changes default required acks in produce request
-    # see: https://pkg.go.dev/github.com/Shopify/sarama?utm_source=godoc#RequiredAcks
-    requiredAcks: "WaitForAll"
+      # optional: changes default required acks in produce request
+      # see: https://pkg.go.dev/github.com/Shopify/sarama?utm_source=godoc#RequiredAcks
+      requiredAcks: "WaitForAll"
 
+      # optional: maximum permitted size of a message (defaults to 1000000)
+      maxMessageBytes: 1000000
 
 current-context: default
 ```
diff --git a/cmd/produce/produce.go b/cmd/produce/produce.go
index 202aa36..f214b62 100644
--- a/cmd/produce/produce.go
+++ b/cmd/produce/produce.go
@@ -29,6 +29,7 @@ func NewProduceCmd() *cobra.Command {
 	cmdProduce.Flags().Int32VarP(&flags.Partition, "partition", "p", -1, "partition to produce to")
 	cmdProduce.Flags().StringVarP(&flags.Partitioner, "partitioner", "P", "", "the partitioning scheme to use. Can be `murmur2`, `hash`, `hash-ref` `manual`, or `random`. (default is murmur2)")
 	cmdProduce.Flags().StringVarP(&flags.RequiredAcks, "required-acks", "", "", "required acks. One of `NoResponse`, `WaitForLocal`, `WaitForAll`. (default is WaitForLocal)")
+	cmdProduce.Flags().IntVarP(&flags.MaxMessageBytes, "max-message-bytes", "", 0, "the maximum permitted size of a message (defaults to 1000000)")
 	cmdProduce.Flags().StringVarP(&flags.Key, "key", "k", "", "key to use for all messages")
 	cmdProduce.Flags().StringVarP(&flags.Value, "value", "v", "", "value to produce")
 	cmdProduce.Flags().BoolVarP(&flags.NullValue, "null-value", "", false, "produce a null value (can be used instead of providing a value with --value)")
diff --git a/cmd/root.go b/cmd/root.go
index cf1f632..39f770b 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -6,6 +6,8 @@ import (
 	"path/filepath"
 	"strings"
 
+	"github.com/deviceinsight/kafkactl/internal/env"
+
 	"github.com/deviceinsight/kafkactl/cmd/alter"
 	"github.com/deviceinsight/kafkactl/cmd/attach"
 	"github.com/deviceinsight/kafkactl/cmd/config"
@@ -26,27 +28,7 @@ import (
 var cfgFile string
 var Verbose bool
 
-var envMapping = map[string]string{
-	"REQUESTTIMEOUT":         "CONTEXTS_DEFAULT_REQUESTTIMEOUT",
-	"BROKERS":                "CONTEXTS_DEFAULT_BROKERS",
-	"TLS_ENABLED":            "CONTEXTS_DEFAULT_TLS_ENABLED",
-	"TLS_CA":                 "CONTEXTS_DEFAULT_TLS_CA",
-	"TLS_CERT":               "CONTEXTS_DEFAULT_TLS_CERT",
-	"TLS_CERTKEY":            "CONTEXTS_DEFAULT_TLS_CERTKEY",
-	"TLS_INSECURE":           "CONTEXTS_DEFAULT_TLS_INSECURE",
-	"SASL_ENABLED":           "CONTEXTS_DEFAULT_SASL_ENABLED",
-	"SASL_USERNAME":          "CONTEXTS_DEFAULT_SASL_USERNAME",
-	"SASL_PASSWORD":          "CONTEXTS_DEFAULT_SASL_PASSWORD",
-	"SASL_MECHANISM":         "CONTEXTS_DEFAULT_SASL_MECHANISM",
-	"CLIENTID":               "CONTEXTS_DEFAULT_CLIENTID",
-	"KAFKAVERSION":           "CONTEXTS_DEFAULT_KAFKAVERSION",
-	"AVRO_SCHEMAREGISTRY":    "CONTEXTS_DEFAULT_AVRO_SCHEMAREGISTRY",
-	"PROTOBUF_PROTOSETFILES": "CONTEXTS_DEFAULT_PROTOBUF_PROTOSETFILES",
-	"PROTOBUF_PROTOSETPATHS": "CONTEXTS_DEFAULT_PROTOBUF_PROTOSETPATHS",
-	"PROTOBUF_IMPORTPATHS":   "CONTEXTS_DEFAULT_PROTOBUF_IMPORTPATHS",
-	"PROTOBUF_PROTOFILES":    "CONTEXTS_DEFAULT_PROTOBUF_PROTOFILES",
-	"DEFAULTPARTITIONER":     "CONTEXTS_DEFAULT_DEFAULTPARTITIONER",
-}
+const defaultContextPrefix = "CONTEXTS_DEFAULT_"
 
 var configPaths = []string{"$HOME/.config/kafkactl", "$HOME/.kafkactl", "$SNAP_REAL_HOME/.config/kafkactl", "$SNAP_DATA/kafkactl", "/etc/kafkactl"}
 
@@ -128,7 +110,8 @@ func initConfig() {
 }
 
 func mapEnvVariables() {
-	for short, long := range envMapping {
+	for _, short := range env.Variables {
+		long := defaultContextPrefix + short
 		if os.Getenv(short) != "" && os.Getenv(long) == "" {
 			_ = os.Setenv(long, os.Getenv(short))
 		}
diff --git a/internal/common-operation.go b/internal/common-operation.go
index 01bb849..08af4cd 100644
--- a/internal/common-operation.go
+++ b/internal/common-operation.go
@@ -45,6 +45,12 @@ type K8sConfig struct {
 	Namespace   string
 }
 
+type ProducerConfig struct {
+	Partitioner     string
+	RequiredAcks    string
+	MaxMessageBytes int
+}
+
 type ClientContext struct {
 	Name               string
 	Brokers            []string
@@ -56,8 +62,7 @@ type ClientContext struct {
 	KafkaVersion       sarama.KafkaVersion
 	AvroSchemaRegistry string
 	Protobuf           protobuf.SearchContext
-	DefaultPartitioner string
-	RequiredAcks       string
+	Producer           ProducerConfig
 }
 
 type Config struct {
@@ -100,8 +105,9 @@ func CreateClientContext() (ClientContext, error) {
 	context.Protobuf.ProtosetFiles = viper.GetStringSlice("contexts." + context.Name + ".protobuf.protosetFiles")
 	context.Protobuf.ProtoImportPaths = viper.GetStringSlice("contexts." + context.Name + ".protobuf.importPaths")
 	context.Protobuf.ProtoFiles = viper.GetStringSlice("contexts." + context.Name + ".protobuf.protoFiles")
-	context.DefaultPartitioner = viper.GetString("contexts." + context.Name + ".defaultPartitioner")
-	context.RequiredAcks = viper.GetString("contexts." + context.Name + ".requiredAcks")
+	context.Producer.Partitioner = viper.GetString("contexts." + context.Name + ".producer.partitioner")
+	context.Producer.RequiredAcks = viper.GetString("contexts." + context.Name + ".producer.requiredAcks")
+	context.Producer.MaxMessageBytes = viper.GetInt("contexts." + context.Name + ".producer.maxMessageBytes")
 	context.Sasl.Enabled = viper.GetBool("contexts." + context.Name + ".sasl.enabled")
 	context.Sasl.Username = viper.GetString("contexts." + context.Name + ".sasl.username")
 	context.Sasl.Password = viper.GetString("contexts." + context.Name + ".sasl.password")
diff --git a/internal/env/variables.go b/internal/env/variables.go
new file mode 100644
index 0000000..967ef3a
--- /dev/null
+++ b/internal/env/variables.go
@@ -0,0 +1,47 @@
+package env
+
+const (
+	RequestTimeout          = "REQUESTTIMEOUT"
+	Brokers                 = "BROKERS"
+	TLSEnabled              = "TLS_ENABLED"
+	TLSCa                   = "TLS_CA"
+	TLSCert                 = "TLS_CERT"
+	TLSCertKey              = "TLS_CERTKEY"
+	TLSInsecure             = "TLS_INSECURE"
+	SaslEnabled             = "SASL_ENABLED"
+	SaslUsername            = "SASL_USERNAME"
+	SaslPassword            = "SASL_PASSWORD"
+	SaslMechanism           = "SASL_MECHANISM"
+	ClientID                = "CLIENTID"
+	KafkaVersion            = "KAFKAVERSION"
+	AvroSchemaRegistry      = "AVRO_SCHEMAREGISTRY"
+	ProtobufProtoSetFiles   = "PROTOBUF_PROTOSETFILES"
+	ProtobufImportPaths     = "PROTOBUF_IMPORTPATHS"
+	ProtobufProtoFiles      = "PROTOBUF_PROTOFILES"
+	ProducerPartitioner     = "PRODUCER_PARTITIONER"
+	ProducerRequiredAcks    = "PRODUCER_REQUIREDACKS"
+	ProducerMaxMessageBytes = "PRODUCER_MAXMESSAGEBYTES"
+)
+
+var Variables = []string{
+	RequestTimeout,
+	Brokers,
+	TLSEnabled,
+	TLSCa,
+	TLSCert,
+	TLSCertKey,
+	TLSInsecure,
+	SaslEnabled,
+	SaslUsername,
+	SaslPassword,
+	SaslMechanism,
+	ClientID,
+	KafkaVersion,
+	AvroSchemaRegistry,
+	ProtobufProtoSetFiles,
+	ProtobufImportPaths,
+	ProtobufProtoFiles,
+	ProducerPartitioner,
+	ProducerRequiredAcks,
+	ProducerMaxMessageBytes,
+}
diff --git a/internal/k8s/k8s-operation.go b/internal/k8s/k8s-operation.go
index 32e22e4..8259af6 100644
--- a/internal/k8s/k8s-operation.go
+++ b/internal/k8s/k8s-operation.go
@@ -6,6 +6,8 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/deviceinsight/kafkactl/internal/env"
+
 	"github.com/deviceinsight/kafkactl/internal"
 	"github.com/deviceinsight/kafkactl/output"
 	"github.com/pkg/errors"
@@ -134,25 +136,30 @@ func parseCompleteCommand(cmd *cobra.Command, found []string) []string {
 
 func parsePodEnvironment(context internal.ClientContext) []string {
 
-	var env []string
-
-	env = appendStrings(env, "BROKERS", context.Brokers)
-	env = appendBool(env, "TLS_ENABLED", context.TLS.Enabled)
-	env = appendStringIfDefined(env, "TLS_CA", context.TLS.CA)
-	env = appendStringIfDefined(env, "TLS_CERT", context.TLS.Cert)
-	env = appendStringIfDefined(env, "TLS_CERTKEY", context.TLS.CertKey)
-	env = appendBool(env, "TLS_INSECURE", context.TLS.Insecure)
-	env = appendBool(env, "SASL_ENABLED", context.Sasl.Enabled)
-	env = appendStringIfDefined(env, "SASL_USERNAME", context.Sasl.Username)
-	env = appendStringIfDefined(env, "SASL_PASSWORD", context.Sasl.Password)
-	env = appendStringIfDefined(env, "SASL_MECHANISM", context.Sasl.Mechanism)
-	env = appendStringIfDefined(env, "REQUESTTIMEOUT", context.RequestTimeout.String())
-	env = appendStringIfDefined(env, "CLIENTID", context.ClientID)
-	env = appendStringIfDefined(env, "KAFKAVERSION", context.KafkaVersion.String())
-	env = appendStringIfDefined(env, "AVRO_SCHEMAREGISTRY", context.AvroSchemaRegistry)
-	env = appendStringIfDefined(env, "DEFAULTPARTITIONER", context.DefaultPartitioner)
-
-	return env
+	var envVariables []string
+
+	envVariables = appendStrings(envVariables, env.Brokers, context.Brokers)
+	envVariables = appendBool(envVariables, env.TLSEnabled, context.TLS.Enabled)
+	envVariables = appendStringIfDefined(envVariables, env.TLSCa, context.TLS.CA)
+	envVariables = appendStringIfDefined(envVariables, env.TLSCert, context.TLS.Cert)
+	envVariables = appendStringIfDefined(envVariables, env.TLSCertKey, context.TLS.CertKey)
+	envVariables = appendBool(envVariables, env.TLSInsecure, context.TLS.Insecure)
+	envVariables = appendBool(envVariables, env.SaslEnabled, context.Sasl.Enabled)
+	envVariables = appendStringIfDefined(envVariables, env.SaslUsername, context.Sasl.Username)
+	envVariables = appendStringIfDefined(envVariables, env.SaslPassword, context.Sasl.Password)
+	envVariables = appendStringIfDefined(envVariables, env.SaslMechanism, context.Sasl.Mechanism)
+	envVariables = appendStringIfDefined(envVariables, env.RequestTimeout, context.RequestTimeout.String())
+	envVariables = appendStringIfDefined(envVariables, env.ClientID, context.ClientID)
+	envVariables = appendStringIfDefined(envVariables, env.KafkaVersion, context.KafkaVersion.String())
+	envVariables = appendStringIfDefined(envVariables, env.AvroSchemaRegistry, context.AvroSchemaRegistry)
+	envVariables = appendStrings(envVariables, env.ProtobufProtoSetFiles, context.Protobuf.ProtosetFiles)
+	envVariables = appendStrings(envVariables, env.ProtobufImportPaths, context.Protobuf.ProtoImportPaths)
+	envVariables = appendStrings(envVariables, env.ProtobufProtoFiles, context.Protobuf.ProtoFiles)
+	envVariables = appendStringIfDefined(envVariables, env.ProducerPartitioner, context.Producer.Partitioner)
+	envVariables = appendStringIfDefined(envVariables, env.ProducerRequiredAcks, context.Producer.RequiredAcks)
+	envVariables = appendIntIfGreaterZero(envVariables, env.ProducerMaxMessageBytes, context.Producer.MaxMessageBytes)
+
+	return envVariables
 }
 
 func appendStrings(env []string, key string, value []string) []string {
@@ -172,3 +179,10 @@ func appendStringIfDefined(env []string, key string, value string) []string {
 	}
 	return env
 }
+
+func appendIntIfGreaterZero(env []string, key string, value int) []string {
+	if value > 0 {
+		return append(env, fmt.Sprintf("%s=%d", key, value))
+	}
+	return env
+}
diff --git a/internal/producer/producer-operation.go b/internal/producer/producer-operation.go
index 8d1152c..01382bf 100644
--- a/internal/producer/producer-operation.go
+++ b/internal/producer/producer-operation.go
@@ -19,6 +19,7 @@ import (
 type Flags struct {
 	Partitioner        string
 	RequiredAcks       string
+	MaxMessageBytes    int
 	Partition          int32
 	Separator          string
 	LineSeparator      string
@@ -64,23 +65,7 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 	config.Producer.Return.Errors = true
 	config.Producer.Return.Successes = true
 
-	partitioner := clientContext.DefaultPartitioner
-	if flags.Partitioner != "" {
-		partitioner = flags.Partitioner
-	}
-
-	config.Producer.Partitioner, err = parsePartitioner(partitioner, flags)
-	if err != nil {
-		return err
-	}
-
-	requiredAcks := clientContext.RequiredAcks
-	if flags.RequiredAcks != "" {
-		requiredAcks = flags.RequiredAcks
-	}
-
-	config.Producer.RequiredAcks, err = parseRequiredAcks(requiredAcks)
-	if err != nil {
+	if err = applyProducerConfigs(config, clientContext, flags); err != nil {
 		return err
 	}
 
@@ -260,6 +245,40 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 	return nil
 }
 
+func applyProducerConfigs(config *sarama.Config, clientContext internal.ClientContext, flags Flags) error {
+
+	var err error
+
+	partitioner := clientContext.Producer.Partitioner
+	if flags.Partitioner != "" {
+		partitioner = flags.Partitioner
+	}
+
+	if config.Producer.Partitioner, err = parsePartitioner(partitioner, flags); err != nil {
+		return err
+	}
+
+	requiredAcks := clientContext.Producer.RequiredAcks
+	if flags.RequiredAcks != "" {
+		requiredAcks = flags.RequiredAcks
+	}
+
+	if config.Producer.RequiredAcks, err = parseRequiredAcks(requiredAcks); err != nil {
+		return err
+	}
+
+	maxMessageBytes := clientContext.Producer.MaxMessageBytes
+	if flags.MaxMessageBytes > 0 {
+		maxMessageBytes = flags.MaxMessageBytes
+	}
+
+	if maxMessageBytes > 0 {
+		config.Producer.MaxMessageBytes = maxMessageBytes
+	}
+
+	return nil
+}
+
 func parseRequiredAcks(requiredAcks string) (sarama.RequiredAcks, error) {
 	switch requiredAcks {
 	case "NoResponse":


Test patch:
diff --git a/cmd/root_test.go b/cmd/root_test.go
index bb70ac8..3873434 100644
--- a/cmd/root_test.go
+++ b/cmd/root_test.go
@@ -4,6 +4,8 @@ import (
 	"os"
 	"testing"
 
+	"github.com/deviceinsight/kafkactl/internal/env"
+
 	"github.com/deviceinsight/kafkactl/testutil"
 	"github.com/spf13/viper"
 )
@@ -40,25 +42,32 @@ func TestEnvironmentVariableLoadingAliases(t *testing.T) {
 
 	testutil.StartUnitTest(t)
 
-	_ = os.Setenv("REQUESTTIMEOUT", "30")
-	_ = os.Setenv("BROKERS", "broker1:9092 broker2:9092")
-	_ = os.Setenv("TLS_ENABLED", "true")
-	_ = os.Setenv("TLS_CA", "my-ca")
-	_ = os.Setenv("TLS_CERT", "my-cert")
-	_ = os.Setenv("TLS_CERTKEY", "my-cert-key")
-	_ = os.Setenv("TLS_INSECURE", "true")
-	_ = os.Setenv("SASL_ENABLED", "true")
-	_ = os.Setenv("SASL_USERNAME", "user")
-	_ = os.Setenv("SASL_PASSWORD", "pass")
-	_ = os.Setenv("SASL_MECHANISM", "scram-sha512")
-	_ = os.Setenv("CLIENTID", "my-client")
-	_ = os.Setenv("KAFKAVERSION", "2.0.1")
-	_ = os.Setenv("AVRO_SCHEMAREGISTRY", "registry:8888")
-	_ = os.Setenv("PROTOBUF_PROTOSETFILES", "/usr/include/protosets/ps1.protoset /usr/lib/ps2.protoset")
-	_ = os.Setenv("PROTOBUF_PROTOSETPATHS", "/usr/include/protosets /usr/lib/protosets")
-	_ = os.Setenv("PROTOBUF_IMPORTPATHS", "/usr/include/protobuf /usr/lib/protobuf")
-	_ = os.Setenv("PROTOBUF_PROTOFILES", "message.proto other.proto")
-	_ = os.Setenv("DEFAULTPARTITIONER", "hash")
+	_ = os.Setenv(env.RequestTimeout, "30")
+	_ = os.Setenv(env.Brokers, "broker1:9092 broker2:9092")
+	_ = os.Setenv(env.TLSEnabled, "true")
+	_ = os.Setenv(env.TLSCa, "my-ca")
+	_ = os.Setenv(env.TLSCert, "my-cert")
+	_ = os.Setenv(env.TLSCertKey, "my-cert-key")
+	_ = os.Setenv(env.TLSInsecure, "true")
+	_ = os.Setenv(env.SaslEnabled, "true")
+	_ = os.Setenv(env.SaslUsername, "user")
+	_ = os.Setenv(env.SaslPassword, "pass")
+	_ = os.Setenv(env.SaslMechanism, "scram-sha512")
+	_ = os.Setenv(env.ClientID, "my-client")
+	_ = os.Setenv(env.KafkaVersion, "2.0.1")
+	_ = os.Setenv(env.AvroSchemaRegistry, "registry:8888")
+	_ = os.Setenv(env.ProtobufProtoSetFiles, "/usr/include/protosets/ps1.protoset /usr/lib/ps2.protoset")
+	_ = os.Setenv(env.ProtobufImportPaths, "/usr/include/protobuf /usr/lib/protobuf")
+	_ = os.Setenv(env.ProtobufProtoFiles, "message.proto other.proto")
+	_ = os.Setenv(env.ProducerPartitioner, "hash")
+	_ = os.Setenv(env.ProducerRequiredAcks, "WaitForAll")
+	_ = os.Setenv(env.ProducerMaxMessageBytes, "1234")
+
+	for _, key := range env.Variables {
+		if os.Getenv(key) == "" {
+			t.Fatalf("missing test case for env variable: %s", key)
+		}
+	}
 
 	kafkaCtl := testutil.CreateKafkaCtlCommand()
 
@@ -86,8 +95,9 @@ func TestEnvironmentVariableLoadingAliases(t *testing.T) {
 	testutil.AssertEquals(t, "2.0.1", viper.GetString("contexts.default.kafkaVersion"))
 	testutil.AssertEquals(t, "registry:8888", viper.GetString("contexts.default.avro.schemaRegistry"))
 	testutil.AssertEquals(t, "/usr/include/protosets/ps1.protoset", viper.GetStringSlice("contexts.default.protobuf.protosetFiles")[0])
-	testutil.AssertEquals(t, "/usr/include/protosets", viper.GetStringSlice("contexts.default.protobuf.protosetPaths")[0])
 	testutil.AssertEquals(t, "/usr/include/protobuf", viper.GetStringSlice("contexts.default.protobuf.importPaths")[0])
 	testutil.AssertEquals(t, "message.proto", viper.GetStringSlice("contexts.default.protobuf.protoFiles")[0])
-	testutil.AssertEquals(t, "hash", viper.GetString("contexts.default.defaultPartitioner"))
+	testutil.AssertEquals(t, "hash", viper.GetString("contexts.default.producer.partitioner"))
+	testutil.AssertEquals(t, "WaitForAll", viper.GetString("contexts.default.producer.requiredAcks"))
+	testutil.AssertEquals(t, "1234", viper.GetString("contexts.default.producer.maxMessageBytes"))
 }
diff --git a/internal/k8s/export_test.go b/internal/k8s/export_test.go
new file mode 100644
index 0000000..9bd53f0
--- /dev/null
+++ b/internal/k8s/export_test.go
@@ -0,0 +1,3 @@
+package k8s
+
+var ParsePodEnvironment = parsePodEnvironment
diff --git a/internal/k8s/k8s-operation_test.go b/internal/k8s/k8s-operation_test.go
new file mode 100644
index 0000000..4f76038
--- /dev/null
+++ b/internal/k8s/k8s-operation_test.go
@@ -0,0 +1,74 @@
+package k8s_test
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Shopify/sarama"
+	"github.com/deviceinsight/kafkactl/internal"
+	"github.com/deviceinsight/kafkactl/internal/env"
+	"github.com/deviceinsight/kafkactl/internal/k8s"
+	"github.com/deviceinsight/kafkactl/testutil"
+)
+
+func TestAllAvailableEnvironmentVariablesAreParsed(t *testing.T) {
+
+	var context internal.ClientContext
+	context.RequestTimeout = 30 * time.Second
+	context.Brokers = []string{"broker1:9092", "broker2:9092"}
+	context.TLS.Enabled = true
+	context.TLS.CA = "my-ca"
+	context.TLS.Cert = "my-cert"
+	context.TLS.CertKey = "my-cert-key"
+	context.TLS.Insecure = true
+	context.Sasl.Enabled = true
+	context.Sasl.Username = "user"
+	context.Sasl.Password = "pass"
+	context.Sasl.Mechanism = "scram-sha512"
+	context.ClientID = "my-client"
+	context.KafkaVersion = sarama.V2_0_1_0
+	context.AvroSchemaRegistry = "registry:8888"
+	context.Protobuf.ProtosetFiles = []string{"/usr/include/protosets/ps1.protoset", "/usr/lib/ps2.protoset"}
+	context.Protobuf.ProtoImportPaths = []string{"/usr/include/protobuf", "/usr/lib/protobuf"}
+	context.Protobuf.ProtoFiles = []string{"message.proto", "other.proto"}
+	context.Producer.Partitioner = "hash"
+	context.Producer.RequiredAcks = "WaitForAll"
+	context.Producer.MaxMessageBytes = 1234
+
+	environment := k8s.ParsePodEnvironment(context)
+
+	envMap := make(map[string]string)
+
+	for _, envVar := range environment {
+		parts := strings.SplitAfterN(envVar, "=", 2)
+		envMap[strings.TrimSuffix(parts[0], "=")] = parts[1]
+	}
+
+	for _, key := range env.Variables {
+		if _, found := envMap[key]; !found {
+			t.Fatalf("env variable not found in parsed environment: %s", key)
+		}
+	}
+
+	testutil.AssertEquals(t, "broker1:9092 broker2:9092", envMap[env.Brokers])
+	testutil.AssertEquals(t, "30s", envMap[env.RequestTimeout])
+	testutil.AssertEquals(t, "true", envMap[env.TLSEnabled])
+	testutil.AssertEquals(t, "my-ca", envMap[env.TLSCa])
+	testutil.AssertEquals(t, "my-cert", envMap[env.TLSCert])
+	testutil.AssertEquals(t, "my-cert-key", envMap[env.TLSCertKey])
+	testutil.AssertEquals(t, "true", envMap[env.TLSInsecure])
+	testutil.AssertEquals(t, "true", envMap[env.SaslEnabled])
+	testutil.AssertEquals(t, "user", envMap[env.SaslUsername])
+	testutil.AssertEquals(t, "pass", envMap[env.SaslPassword])
+	testutil.AssertEquals(t, "scram-sha512", envMap[env.SaslMechanism])
+	testutil.AssertEquals(t, "my-client", envMap[env.ClientID])
+	testutil.AssertEquals(t, "2.0.1", envMap[env.KafkaVersion])
+	testutil.AssertEquals(t, "registry:8888", envMap[env.AvroSchemaRegistry])
+	testutil.AssertEquals(t, "/usr/include/protosets/ps1.protoset /usr/lib/ps2.protoset", envMap[env.ProtobufProtoSetFiles])
+	testutil.AssertEquals(t, "/usr/include/protobuf /usr/lib/protobuf", envMap[env.ProtobufImportPaths])
+	testutil.AssertEquals(t, "message.proto other.proto", envMap[env.ProtobufProtoFiles])
+	testutil.AssertEquals(t, "hash", envMap[env.ProducerPartitioner])
+	testutil.AssertEquals(t, "WaitForAll", envMap[env.ProducerRequiredAcks])
+	testutil.AssertEquals(t, "1234", envMap[env.ProducerMaxMessageBytes])
+}