Repository: deviceinsight/kafkactl
Pull request #113 (2022-01-15 13:19:57)

extend the documentation of the command completion of zsh
# Description

In the documentation is a little mistake about the zsh completion. The documentation of zsh recommends installing all completions to the `$ZSH/cache/completions` directory and reloading the completion plugin. The file `.zprofile` is executed in every new terminal session. So every time the completion is activated. It's more explicit.

## Type of change
- [X] Bug fix (non-breaking change which fixes an issue)

## Documentation
- [x] fix the zsh completion in`README.md`


tarys: Nice, thanks a lot!
Looking forward for merge and new release :)
Am3o: Its a pleasure and i really love the tool! 

Sorry for the force pushes, I had to change my git identity
Am3o: I thought a rebase is a good idea and than this happens... i will undo the rebase 😒
d-rk: @Am3o don't worry about it. I will get it merged somehow :)
But first I need to find the time to take a closer look at the PR.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 679d033..c66a76c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,20 +6,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-## 2.0.0 - 2022-01-17
-
-### 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`.
-
-### Fixed
-- fixed error handling for describe topic command. previously errors for requests to describe partitions had been swallowed.
-
-### Changed
-- changed naming of kafkactl pod when running in k8s. pod is now named based on clientID specified in config.yml
-
 ## 1.24.0 - 2021-12-03
 
 ### Added
@@ -168,7 +154,7 @@ of `plaintext`.
 ## 1.11.0 - 2020-08-07
 
 ### Added
-- direct support for kafka clusters [running in kubernetes](https://github.com/deviceinsight/kafkactl/blob/main/README.md#running-in-kubernetes)
+- direct support for kafka clusters [running in kubernetes](https://github.com/deviceinsight/kafkactl/blob/master/README.md#running-in-kubernetes)
 - `attach` command to get a bash into kafkactl pod when running in kubernetes
 
 ## 1.10.0 - 2020-08-03
diff --git a/README.md b/README.md
index 87222fe..27ad95c 100644
--- a/README.md
+++ b/README.md
@@ -3,7 +3,7 @@
 
 A command-line interface for interaction with Apache Kafka
 
-[![Build Status](https://github.com/deviceinsight/kafkactl/workflows/Lint%20%2F%20Test%20%2F%20IT/badge.svg?branch=main)](https://github.com/deviceinsight/kafkactl/actions)
+[![Build Status](https://github.com/deviceinsight/kafkactl/workflows/Lint%20%2F%20Test%20%2F%20IT/badge.svg?branch=master)](https://github.com/deviceinsight/kafkactl/actions)
 | [![command docs](https://img.shields.io/badge/command-docs-blue.svg)](https://deviceinsight.github.io/kafkactl/)  
 
 ## Features
@@ -130,16 +130,13 @@ contexts:
        protosetFiles:
           - "/usr/include/protoset/other.protoset"
     
-    producer:
-      # optional: changes the default partitioner
-      partitioner: "hash"
+    # optional: changes the default partitioner
+    defaultPartitioner: "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
 ```
@@ -177,15 +174,21 @@ kafkactl completion bash > /usr/local/etc/bash_completion.d/kafkactl
 
 #### zsh
 
+If shell completion is not already enabled in your environment,
+you will need to enable it.  You can execute the following once:
+
 ```
-source <(kafkactl completion zsh)
+echo "autoload -U compinit; compinit" >> ~/.zshrc
 ```
 
 To load completions for each session, execute once:
+
 ```
 kafkactl completion zsh > "${fpath[1]}/_kafkactl"
 ```
 
+You will need to start a new shell for this setup to take effect.
+
 #### Fish
 
 ```
diff --git a/cmd/produce/produce.go b/cmd/produce/produce.go
index f214b62..202aa36 100644
--- a/cmd/produce/produce.go
+++ b/cmd/produce/produce.go
@@ -29,7 +29,6 @@ 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 39f770b..cf1f632 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -6,8 +6,6 @@ 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"
@@ -28,7 +26,27 @@ import (
 var cfgFile string
 var Verbose bool
 
-const defaultContextPrefix = "CONTEXTS_DEFAULT_"
+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",
+}
 
 var configPaths = []string{"$HOME/.config/kafkactl", "$HOME/.kafkactl", "$SNAP_REAL_HOME/.config/kafkactl", "$SNAP_DATA/kafkactl", "/etc/kafkactl"}
 
@@ -110,8 +128,7 @@ func initConfig() {
 }
 
 func mapEnvVariables() {
-	for _, short := range env.Variables {
-		long := defaultContextPrefix + short
+	for short, long := range envMapping {
 		if os.Getenv(short) != "" && os.Getenv(long) == "" {
 			_ = os.Setenv(long, os.Getenv(short))
 		}
diff --git a/go.mod b/go.mod
index 5c7f1c7..2dfc197 100644
--- a/go.mod
+++ b/go.mod
@@ -11,8 +11,6 @@ require (
 	github.com/linkedin/goavro/v2 v2.10.0
 	github.com/magiconair/properties v1.8.4 // indirect
 	github.com/mitchellh/mapstructure v1.3.3 // indirect
-	github.com/onsi/ginkgo v1.8.0 // indirect
-	github.com/onsi/gomega v1.4.3 // indirect
 	github.com/pelletier/go-toml v1.8.1 // indirect
 	github.com/pkg/errors v0.9.1
 	github.com/spf13/afero v1.4.0 // indirect
@@ -27,12 +25,9 @@ require (
 	go.uber.org/ratelimit v0.1.0
 	golang.org/x/tools v0.1.0 // indirect
 	google.golang.org/protobuf v1.25.1-0.20200805231151-a709e31e5d12
-	gopkg.in/airbrake/gobrake.v2 v2.0.9 // indirect
 	gopkg.in/errgo.v2 v2.1.0
-	gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 // indirect
 	gopkg.in/ini.v1 v1.61.0 // indirect
 	gopkg.in/yaml.v2 v2.3.0
-	gotest.tools/gotestsum v1.7.0 // indirect
 )
 
 go 1.16
diff --git a/go.sum b/go.sum
index 01f8c18..7c16fa2 100644
--- a/go.sum
+++ b/go.sum
@@ -49,8 +49,6 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
-github.com/dnephin/pflag v1.0.7 h1:oxONGlWxhmUct0YzKTgrpQv9AUA1wtPBn7zuSjJqptk=
-github.com/dnephin/pflag v1.0.7/go.mod h1:uxE91IoWURlOiTUIA8Mq5ZZkAv3dPUfZNaT80Zm7OQE=
 github.com/eapache/go-resiliency v1.2.0 h1:v7g92e/KSN71Rq7vSThKaWIq68fL4YHvWyiUKorFR1Q=
 github.com/eapache/go-resiliency v1.2.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw=
@@ -60,8 +58,6 @@ github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFP
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
-github.com/fatih/color v1.10.0 h1:s36xzo75JdqLaaWoiEHk767eHiwo0598uUxyfiPkDsg=
-github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/frankban/quicktest v1.11.3 h1:8sXhOn0uLys67V8EsXLc6eszDs8VXWxL3iRvebPhedY=
@@ -110,8 +106,6 @@ github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXi
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
-github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
-github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
@@ -148,7 +142,6 @@ github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO
 github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=
 github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
 github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
-github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
@@ -166,8 +159,6 @@ github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJk
 github.com/jhump/protoreflect v1.10.1 h1:iH+UZfsbRE6vpyZH7asAjTPWJf7RJbpZ9j/N3lDlKs0=
 github.com/jhump/protoreflect v1.10.1/go.mod h1:7GcYQDdMU/O/BBrl/cX6PNHpXh6cenjd8pneu5yW7Tg=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
-github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
-github.com/jonboulle/clockwork v0.2.2/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
@@ -196,11 +187,7 @@ github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czP
 github.com/magiconair/properties v1.8.4 h1:8KGKTcQQGm0Kv7vEbKFErAoAOFyyacLStRtQSeYtvkY=
 github.com/magiconair/properties v1.8.4/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
-github.com/mattn/go-colorable v0.1.8 h1:c1ghPdyEDarC70ftn0y+A/Ee++9zz8ljHG1b13eJ0s8=
-github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
-github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
-github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
@@ -218,9 +205,6 @@ github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3Rllmb
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/nishanths/predeclared v0.0.0-20200524104333-86fad755b4d3/go.mod h1:nt3d53pc1VYcphSCIaYAJtnPYnr3Zyn8fMq2wvPGPso=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
-github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pelletier/go-toml v1.8.1 h1:1Nf83orprkJyknT6h7zbuEGUEjcyVlCxSUGTENmNCRM=
@@ -346,11 +330,9 @@ golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028/go.mod h1:E/iHnbuqvinMTCc
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.3.0 h1:RM4zey1++hCTbCVQfnWeKs9/IEsaBLA8vTkd0WVtmH4=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -381,12 +363,10 @@ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -399,8 +379,6 @@ golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -408,7 +386,6 @@ golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1 h1:SrN+KX8Art/Sf4HNj6Zcz06G7VEz+7w9tdXTPOZ7+l4=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -433,7 +410,6 @@ golang.org/x/tools v0.0.0-20190506145303-2d16b83fe98c/go.mod h1:RgjU9mgBXZiqYHBn
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190606124116-d0a3d012864b/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
-golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190628153133-6cdbf07be9d0/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190816200558-6889da9d5479/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -488,7 +464,6 @@ google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.25.1-0.20200805231151-a709e31e5d12 h1:OwhZOOMuf7leLaSCuxtQ9FW7ui2L2L6UKOtKAUqovUQ=
 google.golang.org/protobuf v1.25.1-0.20200805231151-a709e31e5d12/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
-gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
@@ -496,8 +471,6 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntN
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/errgo.v2 v2.1.0 h1:0vLT13EuvQ0hNvakwLuFZ/jYrLp5F3kcWHXdRggjCE8=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
-gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
-gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/ini.v1 v1.61.0 h1:LBCdW4FmFYL4s/vDZD1RQYX7oAR6IjujCYgMdbHBR10=
 gopkg.in/ini.v1 v1.61.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
@@ -513,9 +486,6 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gotest.tools/gotestsum v1.7.0 h1:RwpqwwFKBAa2h+F6pMEGpE707Edld0etUD3GhqqhDNc=
-gotest.tools/gotestsum v1.7.0/go.mod h1:V1m4Jw3eBerhI/A6qCxUE07RnCg7ACkKj9BYcAm09V8=
-gotest.tools/v3 v3.0.3/go.mod h1:Z7Lb0S5l+klDB31fvDQX8ss/FlKDxtlFlw3Oa8Ymbl8=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
diff --git a/internal/common-operation.go b/internal/common-operation.go
index 5dc0642..01bb849 100644
--- a/internal/common-operation.go
+++ b/internal/common-operation.go
@@ -45,12 +45,6 @@ type K8sConfig struct {
 	Namespace   string
 }
 
-type ProducerConfig struct {
-	Partitioner     string
-	RequiredAcks    string
-	MaxMessageBytes int
-}
-
 type ClientContext struct {
 	Name               string
 	Brokers            []string
@@ -62,7 +56,8 @@ type ClientContext struct {
 	KafkaVersion       sarama.KafkaVersion
 	AvroSchemaRegistry string
 	Protobuf           protobuf.SearchContext
-	Producer           ProducerConfig
+	DefaultPartitioner string
+	RequiredAcks       string
 }
 
 type Config struct {
@@ -105,9 +100,8 @@ 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.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.DefaultPartitioner = viper.GetString("contexts." + context.Name + ".defaultPartitioner")
+	context.RequiredAcks = viper.GetString("contexts." + context.Name + ".requiredAcks")
 	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")
@@ -143,7 +137,7 @@ func CreateClientConfig(context *ClientContext) (*sarama.Config, error) {
 
 	var config = sarama.NewConfig()
 	config.Version = context.KafkaVersion
-	config.ClientID = GetClientID(context, "kafkactl-")
+	config.ClientID = getClientID(context)
 
 	if context.RequestTimeout > 0 {
 		output.Debugf("using admin request timeout: %s", context.RequestTimeout.String())
@@ -191,7 +185,7 @@ func CreateClientConfig(context *ClientContext) (*sarama.Config, error) {
 	return config, nil
 }
 
-func GetClientID(context *ClientContext, defaultPrefix string) string {
+func getClientID(context *ClientContext) string {
 
 	var (
 		err error
@@ -202,9 +196,9 @@ func GetClientID(context *ClientContext, defaultPrefix string) string {
 		return context.ClientID
 	} else if usr, err = user.Current(); err != nil {
 		output.Warnf("Failed to read current user: %v", err)
-		return strings.TrimSuffix(defaultPrefix, "-")
+		return "kafkactl"
 	} else {
-		return defaultPrefix + sanitizeUsername(usr.Username)
+		return "kafkactl-" + sanitizeUsername(usr.Username)
 	}
 }
 
diff --git a/internal/consumergroups/consumer-group-operation.go b/internal/consumergroups/consumer-group-operation.go
index fe9e527..c861bcc 100644
--- a/internal/consumergroups/consumer-group-operation.go
+++ b/internal/consumergroups/consumer-group-operation.go
@@ -441,12 +441,6 @@ func findAssignedTopics(admin sarama.ClusterAdmin, groupNames []string) (map[str
 				continue
 			}
 
-			if description.State != "Stable" {
-				// only take stable groups into account to circumvent https://github.com/deviceinsight/kafkactl/issues/109
-				output.Debugf("do not filter on group %s, because state is: %s", description.GroupId, description.State)
-				continue
-			}
-
 			assignment, err := member.GetMemberAssignment()
 
 			if err != nil {
diff --git a/internal/env/variables.go b/internal/env/variables.go
deleted file mode 100644
index 967ef3a..0000000
--- a/internal/env/variables.go
+++ /dev/null
@@ -1,47 +0,0 @@
-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/executor.go b/internal/k8s/executor.go
index e54e1d9..41c231b 100644
--- a/internal/k8s/executor.go
+++ b/internal/k8s/executor.go
@@ -6,7 +6,6 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/deviceinsight/kafkactl/internal"
 	"github.com/deviceinsight/kafkactl/output"
@@ -22,7 +21,6 @@ type executor struct {
 	kubectlBinary string
 	version       Version
 	runner        Runner
-	clientID      string
 	kubeConfig    string
 	kubeContext   string
 	namespace     string
@@ -32,9 +30,6 @@ type executor struct {
 const letterBytes = "abcdefghijklmnpqrstuvwxyz123456789"
 
 func randomString(n int) string {
-
-	rand.Seed(time.Now().UnixNano())
-
 	b := make([]byte, n)
 	for i := range b {
 		b[i] = letterBytes[rand.Intn(len(letterBytes))]
@@ -86,7 +81,6 @@ func newExecutor(context internal.ClientContext, runner Runner) *executor {
 	return &executor{
 		kubectlBinary: context.Kubernetes.Binary,
 		version:       getKubectlVersion(context.Kubernetes.Binary, runner),
-		clientID:      internal.GetClientID(&context, ""),
 		kubeConfig:    context.Kubernetes.KubeConfig,
 		kubeContext:   context.Kubernetes.KubeContext,
 		namespace:     context.Kubernetes.Namespace,
@@ -110,10 +104,6 @@ func (kubectl *executor) Run(dockerImageType, entryPoint string, kafkactlArgs []
 
 	podName := "kafkactl-" + randomString(10)
 
-	if kubectl.clientID != "" {
-		podName = "kafkactl-" + kubectl.clientID
-	}
-
 	kubectlArgs := []string{
 		"run", "--rm", "-i", "--tty", "--restart=Never", podName,
 		"--image", dockerImage,
diff --git a/internal/k8s/k8s-operation.go b/internal/k8s/k8s-operation.go
index 8259af6..665b17a 100644
--- a/internal/k8s/k8s-operation.go
+++ b/internal/k8s/k8s-operation.go
@@ -6,8 +6,6 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/deviceinsight/kafkactl/internal/env"
-
 	"github.com/deviceinsight/kafkactl/internal"
 	"github.com/deviceinsight/kafkactl/output"
 	"github.com/pkg/errors"
@@ -136,30 +134,24 @@ func parseCompleteCommand(cmd *cobra.Command, found []string) []string {
 
 func parsePodEnvironment(context internal.ClientContext) []string {
 
-	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
+	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, "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
 }
 
 func appendStrings(env []string, key string, value []string) []string {
@@ -179,10 +171,3 @@ 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 01382bf..8d1152c 100644
--- a/internal/producer/producer-operation.go
+++ b/internal/producer/producer-operation.go
@@ -19,7 +19,6 @@ import (
 type Flags struct {
 	Partitioner        string
 	RequiredAcks       string
-	MaxMessageBytes    int
 	Partition          int32
 	Separator          string
 	LineSeparator      string
@@ -65,7 +64,23 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 	config.Producer.Return.Errors = true
 	config.Producer.Return.Successes = true
 
-	if err = applyProducerConfigs(config, clientContext, flags); err != nil {
+	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 {
 		return err
 	}
 
@@ -245,40 +260,6 @@ 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":
diff --git a/internal/topic/topic-operation.go b/internal/topic/topic-operation.go
index a118068..28e5312 100644
--- a/internal/topic/topic-operation.go
+++ b/internal/topic/topic-operation.go
@@ -138,7 +138,6 @@ func (operation *Operation) DescribeTopic(topic string, flags DescribeTopicFlags
 		admin   sarama.ClusterAdmin
 		err     error
 		exists  bool
-		t       Topic
 	)
 
 	if context, err = internal.CreateClientContext(); err != nil {
@@ -161,9 +160,7 @@ func (operation *Operation) DescribeTopic(topic string, flags DescribeTopicFlags
 		return errors.Wrap(err, "failed to create cluster admin")
 	}
 
-	if t, err = readTopic(&client, &admin, topic, allFields); err != nil {
-		return errors.Wrap(err, "failed to read topic")
-	}
+	var t, _ = readTopic(&client, &admin, topic, allFields)
 
 	return operation.printTopic(t, flags)
 }
@@ -242,7 +239,6 @@ func (operation *Operation) AlterTopic(topic string, flags AlterTopicFlags) erro
 		admin   sarama.ClusterAdmin
 		err     error
 		exists  bool
-		t       Topic
 	)
 
 	if context, err = internal.CreateClientContext(); err != nil {
@@ -265,9 +261,7 @@ func (operation *Operation) AlterTopic(topic string, flags AlterTopicFlags) erro
 		return errors.Wrap(err, "failed to create cluster admin")
 	}
 
-	if t, err = readTopic(&client, &admin, topic, allFields); err != nil {
-		return errors.Wrap(err, "failed to read topic")
-	}
+	var t, _ = readTopic(&client, &admin, topic, allFields)
 
 	if flags.Partitions != 0 {
 		if len(t.Partitions) > int(flags.Partitions) {
@@ -601,6 +595,7 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 	}
 
 	partitionChannel := make(chan Partition)
+	errChannel := make(chan error)
 
 	// read partitions in parallel
 	for _, p := range ps {
@@ -611,37 +606,38 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 
 			if requestedFields.partitionOffset {
 				if np.OldestOffset, err = (*client).GetOffset(name, partitionId, sarama.OffsetOldest); err != nil {
-					output.Warnf("unable to read oldest offset for topic %s partition %d", name, partitionId)
+					errChannel <- errors.Errorf("unable to read oldest offset for topic %s partition %d", name, partitionId)
+					return
 				}
 
 				if np.NewestOffset, err = (*client).GetOffset(name, partitionId, sarama.OffsetNewest); err != nil {
-					output.Warnf("unable to read newest offset for topic %s partition %d", name, partitionId)
+					errChannel <- errors.Errorf("unable to read newest offset for topic %s partition %d", name, partitionId)
+					return
 				}
 			}
 
 			if requestedFields.partitionLeader {
 				if led, err = (*client).Leader(name, partitionId); err != nil {
-					output.Warnf("unable to read leader for topic %s partition %d", name, partitionId)
-					np.Leader = "none"
-				} else {
-					np.Leader = led.Addr()
+					errChannel <- errors.Errorf("unable to read leader for topic %s partition %d", name, partitionId)
+					return
 				}
+				np.Leader = led.Addr()
 			}
 
 			if requestedFields.partitionReplicas {
 				if np.Replicas, err = (*client).Replicas(name, partitionId); err != nil {
-					output.Warnf("unable to read replicas for topic %s partition %d", name, partitionId)
-				} else {
-					sort.Slice(np.Replicas, func(i, j int) bool { return np.Replicas[i] < np.Replicas[j] })
+					errChannel <- errors.Errorf("unable to read replicas for topic %s partition %d", name, partitionId)
+					return
 				}
+				sort.Slice(np.Replicas, func(i, j int) bool { return np.Replicas[i] < np.Replicas[j] })
 			}
 
 			if requestedFields.partitionISRs {
 				if np.ISRs, err = (*client).InSyncReplicas(name, partitionId); err != nil {
-					output.Warnf("unable to read inSyncReplicas for topic %s partition %d", name, partitionId)
-				} else {
-					sort.Slice(np.ISRs, func(i, j int) bool { return np.ISRs[i] < np.ISRs[j] })
+					errChannel <- errors.Errorf("unable to read inSyncReplicas for topic %s partition %d", name, partitionId)
+					return
 				}
+				sort.Slice(np.ISRs, func(i, j int) bool { return np.ISRs[i] < np.ISRs[j] })
 			}
 
 			partitionChannel <- np
@@ -649,8 +645,12 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 	}
 
 	for range ps {
-		partition := <-partitionChannel
-		top.Partitions = append(top.Partitions, partition)
+		select {
+		case partition := <-partitionChannel:
+			top.Partitions = append(top.Partitions, partition)
+		case err := <-errChannel:
+			return top, err
+		}
 	}
 
 	sort.Slice(top.Partitions, func(i, j int) bool {


Test patch:
diff --git a/.github/workflows/lint_test.yml b/.github/workflows/lint_test.yml
index 8fc2e97..796c629 100644
--- a/.github/workflows/lint_test.yml
+++ b/.github/workflows/lint_test.yml
@@ -3,7 +3,7 @@ name: Lint / Test / IT
 on:
   push:
     branches:
-      - main
+      - master
     tags:
       - v*
   pull_request:
diff --git a/cmd/root_test.go b/cmd/root_test.go
index 3873434..bb70ac8 100644
--- a/cmd/root_test.go
+++ b/cmd/root_test.go
@@ -4,8 +4,6 @@ import (
 	"os"
 	"testing"
 
-	"github.com/deviceinsight/kafkactl/internal/env"
-
 	"github.com/deviceinsight/kafkactl/testutil"
 	"github.com/spf13/viper"
 )
@@ -42,32 +40,25 @@ func TestEnvironmentVariableLoadingAliases(t *testing.T) {
 
 	testutil.StartUnitTest(t)
 
-	_ = 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)
-		}
-	}
+	_ = 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")
 
 	kafkaCtl := testutil.CreateKafkaCtlCommand()
 
@@ -95,9 +86,8 @@ 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.producer.partitioner"))
-	testutil.AssertEquals(t, "WaitForAll", viper.GetString("contexts.default.producer.requiredAcks"))
-	testutil.AssertEquals(t, "1234", viper.GetString("contexts.default.producer.maxMessageBytes"))
+	testutil.AssertEquals(t, "hash", viper.GetString("contexts.default.defaultPartitioner"))
 }
diff --git a/internal/k8s/export_test.go b/internal/k8s/export_test.go
deleted file mode 100644
index 9bd53f0..0000000
--- a/internal/k8s/export_test.go
+++ /dev/null
@@ -1,3 +0,0 @@
-package k8s
-
-var ParsePodEnvironment = parsePodEnvironment
diff --git a/internal/k8s/k8s-operation_test.go b/internal/k8s/k8s-operation_test.go
deleted file mode 100644
index 4f76038..0000000
--- a/internal/k8s/k8s-operation_test.go
+++ /dev/null
@@ -1,74 +0,0 @@
-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])
-}