Repository: deviceinsight/kafkactl
Pull request #37 (2020-02-20 09:25:41)

Add replication-factor flag to alter topic
This is a working draft based on our changes to the sarama lib: https://github.com/Shopify/sarama/pull/1617

*Note:* Currently you have to have the sarama source code available locally to build this.

Some things we may want to add:
- [ ] List pending reassignments
- [ ] Abort pending reassignments
- [ ] Interactive reassignment (execute and wait until all partitions are re-assigned, <kbd>Ctrl</kbd>+<kbd>C</kbd> to abort re-assignment)



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ab794d6..c7cb6ec 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- Add parameter `--replication-factor` to `alter topic` command which allows changing the replication factor of a topic.
+  Note that kafka >= 2.4.0.0 is required, otherwise the relevant api calls are not available.
+- Added command `alter partition` which currently only enables to manually assign broker replicas to a partition.
+  Note that kafka >= 2.4.0.0 is required, otherwise the relevant api calls are not available.
+- Added `requestTimeout` config to control timeout of admin requests.
+
 ## 1.13.3 - 2020-11-11
 
 ### Fixed
diff --git a/README.md b/README.md
index 692dcb1..a53624d 100644
--- a/README.md
+++ b/README.md
@@ -110,6 +110,9 @@ contexts:
     # optional: kafkaVersion (defaults to 2.0.0)
     kafkaVersion: 1.1.1
 
+    # optional: timeout for admin requests (defaults to 3s)
+    requestTimeout: 10s
+
     # optional: avro schema registry
     avro:
       schemaRegistry: localhost:8081
@@ -404,18 +407,37 @@ An additional parameter `print-schema` can be provided to display the schema use
 
 ### Altering topics
 
-Using the `alter topic` command allows you to change the partition count and topic-level configurations of an existing topic.
+Using the `alter topic` command allows you to change the partition count, replication factor and topic-level
+configurations of an existing topic.
 
 The partition count can be increased with:
 ```bash
 kafkactl alter topic my-topic --partitions 32
 ```
 
+The replication factor can be altered with:
+```bash
+kafkactl alter topic my-topic --replication-factor 2
+```
+
+> :information_source: when altering replication factor, kafkactl tries to keep the number of replicas assigned to each
+> broker balanced. If you need more control over the assigned replicas use `alter partition` directly.
+
 The topic configs can be edited by supplying key value pairs as follows:
 ```bash
 kafkactl alter topic my-topic --config retention.ms=3600 --config cleanup.policy=compact
 ```
 
+> :bulb: use the flag `--validate-only` to perform a dry-run without actually modifying the topic 
+
+### Altering partitions
+
+The assigned replicas of a partition can directly be altered with:
+```bash
+# set brokers 102,103 as replicas for partition 3 of topic my-topic
+kafkactl alter topic my-topic 3 -r 102,103
+```
+
 ### Consumer groups
 
 In order to get a list of consumer groups the `get consumer-groups` command can be used:
diff --git a/cmd/alter/alter-partition.go b/cmd/alter/alter-partition.go
new file mode 100644
index 0000000..2b79ca9
--- /dev/null
+++ b/cmd/alter/alter-partition.go
@@ -0,0 +1,59 @@
+package alter
+
+import (
+	"github.com/deviceinsight/kafkactl/cmd/validation"
+	"github.com/deviceinsight/kafkactl/operations"
+	"github.com/deviceinsight/kafkactl/operations/k8s"
+	"github.com/deviceinsight/kafkactl/operations/partitions"
+	"github.com/deviceinsight/kafkactl/output"
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+	"strconv"
+)
+
+func newAlterPartitionCmd() *cobra.Command {
+
+	var flags partitions.AlterPartitionFlags
+
+	var cmdAlterPartition = &cobra.Command{
+		Use:   "partition TOPIC PARTITION",
+		Short: "alter a partition",
+		Args:  cobra.ExactArgs(2),
+		Run: func(cmd *cobra.Command, args []string) {
+			if !(&k8s.K8sOperation{}).TryRun(cmd, args) {
+
+				var partition int32
+
+				if i, err := strconv.ParseInt(args[1], 10, 64); err != nil {
+					output.Fail(errors.Errorf("argument 2 needs to be a partition %s", args[1]))
+				} else {
+					partition = int32(i)
+				}
+
+				if err := (&partitions.PartitionOperation{}).AlterPartition(args[0], partition, flags); err != nil {
+					output.Fail(err)
+				}
+			}
+		},
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			return validation.ValidateAtLeastOneRequiredFlag(cmd)
+		},
+		ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+			if len(args) == 0 {
+				return operations.CompleteTopicNames(cmd, args, toComplete)
+			} else if len(args) == 1 {
+				return partitions.CompletePartitionIds(cmd, args, toComplete)
+			} else {
+				return nil, cobra.ShellCompDirectiveNoFileComp
+			}
+		},
+	}
+
+	cmdAlterPartition.Flags().Int32SliceVarP(&flags.Replicas, "replicas", "r", nil, "set replicas for a partition")
+	cmdAlterPartition.Flags().BoolVarP(&flags.ValidateOnly, "validate-only", "v", false, "validate only")
+
+	if err := validation.MarkFlagAtLeastOneRequired(cmdAlterPartition.Flags(), "replicas"); err != nil {
+		panic(err)
+	}
+	return cmdAlterPartition
+}
diff --git a/cmd/alter/alter-topic.go b/cmd/alter/alter-topic.go
index bfb7f10..a48783f 100644
--- a/cmd/alter/alter-topic.go
+++ b/cmd/alter/alter-topic.go
@@ -30,6 +30,7 @@ func newAlterTopicCmd() *cobra.Command {
 	}
 
 	cmdAlterTopic.Flags().Int32VarP(&flags.Partitions, "partitions", "p", flags.Partitions, "number of partitions")
+	cmdAlterTopic.Flags().Int16VarP(&flags.ReplicationFactor, "replication-factor", "r", flags.ReplicationFactor, "replication factor")
 	cmdAlterTopic.Flags().StringArrayVarP(&flags.Configs, "config", "c", flags.Configs, "configs in format `key=value`")
 	cmdAlterTopic.Flags().BoolVarP(&flags.ValidateOnly, "validate-only", "v", false, "validate only")
 
@@ -39,6 +40,9 @@ func newAlterTopicCmd() *cobra.Command {
 	if err := validation.MarkFlagAtLeastOneRequired(cmdAlterTopic.Flags(), "config"); err != nil {
 		panic(err)
 	}
+	if err := validation.MarkFlagAtLeastOneRequired(cmdAlterTopic.Flags(), "replication-factor"); err != nil {
+		panic(err)
+	}
 
 	return cmdAlterTopic
 }
diff --git a/cmd/alter/alter.go b/cmd/alter/alter.go
index 8098bdd..0139c1f 100644
--- a/cmd/alter/alter.go
+++ b/cmd/alter/alter.go
@@ -7,11 +7,12 @@ import (
 func NewAlterCmd() *cobra.Command {
 
 	var cmdAlter = &cobra.Command{
-		Use:   "alter",
-		Short: "alter topics",
+		Use:     "alter",
+		Aliases: []string{"edit"},
+		Short:   "alter topics, partitions",
 	}
 
 	cmdAlter.AddCommand(newAlterTopicCmd())
-
+	cmdAlter.AddCommand(newAlterPartitionCmd())
 	return cmdAlter
 }
diff --git a/cmd/root.go b/cmd/root.go
index 8b20ff0..b16b723 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -26,6 +26,7 @@ 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",
diff --git a/docker/.env b/docker/.env
index df37c59..884f758 100644
--- a/docker/.env
+++ b/docker/.env
@@ -1,1 +1,2 @@
-CP_VERSION=5.5.0
+CP_VERSION=6.0.1
+WAIT_FOR_KAFKA_TIMEOUT=480
\ No newline at end of file
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index efdab30..da1f662 100644
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -88,8 +88,8 @@ services:
       - kafka2
       - kafka3
     command: >
-      bash -c 'cub kafka-ready -b kafka1:9092,kafka2:9092,kafka3:9092 3 60
-      && cub sr-ready schema-registry 8081 60'
+      bash -c "cub kafka-ready -b kafka1:9092,kafka2:9092,kafka3:9092 3 ${WAIT_FOR_KAFKA_TIMEOUT}
+      && cub sr-ready schema-registry 8081 ${WAIT_FOR_KAFKA_TIMEOUT}"
     environment:
       KAFKA_BROKER_ID: ignored
       KAFKA_ZOOKEEPER_CONNECT: ignored
diff --git a/docker/wait-for-kafka.sh b/docker/wait-for-kafka.sh
index 787d682..c0c893c 100755
--- a/docker/wait-for-kafka.sh
+++ b/docker/wait-for-kafka.sh
@@ -1,6 +1,10 @@
 #!/bin/bash
 
-timeout=60
+SCRIPT_DIR="$(dirname "$(readlink -f "$0")")"
+
+source ${SCRIPT_DIR}/.env
+
+timeout=${WAIT_FOR_KAFKA_TIMEOUT}
 time=0
 
 docker inspect wait-for-kafka 2>/dev/null | jq -c -e '.[0].State | select (.Status == "exited" and .ExitCode == 0)' > /dev/null
diff --git a/generate_completion.sh b/generate_completion.sh
index 5b2c9e9..09d8ac3 100755
--- a/generate_completion.sh
+++ b/generate_completion.sh
@@ -1,12 +1,20 @@
 #!/bin/sh
 
-SCRIPT_PATH="$( cd "$(dirname "$0")" >/dev/null 2>&1 ; pwd -P )"
+SCRIPT_PATH=`readlink -m $0/..`
 
 TARGET=$1
 BIN_PATH=$2
 
 ARCH=linux_amd64
 
+if [ "${BIN_PATH}" == "" ]; then
+  BIN_PATH=${SCRIPT_PATH}/kafkactl
+fi
+
+if [ "$TARGET" == "" ]; then
+  TARGET=$ARCH
+fi
+
 if [ "$(uname)" == "Darwin" ]; then
   ARCH=darwin_amd64
 fi
@@ -16,4 +24,6 @@ if [ "$ARCH" == "$TARGET" ]; then
 
   echo "" > /tmp/empty.yaml
   ${BIN_PATH} completion bash > ${SCRIPT_PATH}/kafkactl-completion.bash --config-file=/tmp/empty.yaml
+else
+  echo "not generating: $ARCH $TARGET"
 fi
diff --git a/go.mod b/go.mod
index 5e164ad..08a92f1 100644
--- a/go.mod
+++ b/go.mod
@@ -22,9 +22,10 @@ require (
 	go.uber.org/ratelimit v0.1.0
 	golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a // indirect
 	golang.org/x/net v0.0.0-20200923182212-328152dc79b1 // indirect
-	golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d // indirect
+	gopkg.in/errgo.v2 v2.1.0
 	gopkg.in/ini.v1 v1.61.0 // indirect
 	gopkg.in/yaml.v2 v2.3.0
+	gotest.tools/gotestsum v0.6.0 // indirect
 )
 
 go 1.14
diff --git a/go.sum b/go.sum
index a9edfe7..6ba15ce 100644
--- a/go.sum
+++ b/go.sum
@@ -55,6 +55,8 @@ github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
+github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
+github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 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.10.0 h1:Gfh+GAJZOAoKZsIZeZbdn2JF10kN1XHNvjsvQK8gVkE=
@@ -92,10 +94,13 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.4.1 h1:/exdXoGamhu5ONeUJH0deniYLWYvQwW66yvlfiiKTu0=
 github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 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=
@@ -169,7 +174,14 @@ 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.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-colorable v0.1.6 h1:6Su7aK7lXmJ/U79bYtBjLNaha4Fs1Rg9plHpcH+vvnE=
+github.com/mattn/go-colorable v0.1.6/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.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/mattn/go-isatty v0.0.11/go.mod h1:PhnuNfih5lzO57/f3n+odYbM4JtupLOxQOAqxQCu2WE=
+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=
@@ -344,6 +356,7 @@ golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5h
 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=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -351,9 +364,14 @@ golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 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-20200923182605-d9f96fdee20d h1:L/IKR6COd7ubZrs2oTnTi73IhgqJ71c9s80WsQnh0Es=
 golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201009025420-dfb3f7c4e634 h1:bNEHhJCnrwMKNMmOx3yAynp5vs5/gRy+XWFtZFu7NBM=
+golang.org/x/sys v0.0.0-20201009025420-dfb3f7c4e634/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 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=
@@ -375,6 +393,7 @@ golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBn
 golang.org/x/tools v0.0.0-20190506145303-2d16b83fe98c/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=
@@ -414,6 +433,7 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/errgo.v2 v2.1.0 h1:0vLT13EuvQ0hNvakwLuFZ/jYrLp5F3kcWHXdRggjCE8=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
@@ -442,6 +462,10 @@ 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-20200601152816-913338de1bd2 h1:VEmvx0P+GVTgkNu2EdTN988YCZPcD3lo9AoczZpucwc=
 gopkg.in/yaml.v3 v3.0.0-20200601152816-913338de1bd2/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools v1.4.0 h1:BjtEgfuw8Qyd+jPvQz8CfoxiO/UjFEidWinwEXZiWv0=
+gotest.tools/gotestsum v0.6.0 h1:0zIxynXq9gkAcRpboAi3qOQIkZkCt/stfQzd7ab7Czs=
+gotest.tools/gotestsum v0.6.0/go.mod h1:LEX+ioCVdeWhZc8GYfiBRag360eBhwixWJ62R9eDQtI=
+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/it-config.yml b/it-config.yml
index 7736079..8ef8b63 100644
--- a/it-config.yml
+++ b/it-config.yml
@@ -6,4 +6,5 @@ contexts:
       - localhost:39092
     avro:
       schemaRegistry: localhost:18081
+    kafkaversion: 2.6.0
 current-context: default
diff --git a/kafkactl-completion.bash b/kafkactl-completion.bash
index b444327..b0ded9f 100644
--- a/kafkactl-completion.bash
+++ b/kafkactl-completion.bash
@@ -353,6 +353,39 @@ __kafkactl_handle_word()
     __kafkactl_handle_word
 }
 
+_kafkactl_alter_partition()
+{
+    last_command="kafkactl_alter_partition"
+
+    command_aliases=()
+
+    commands=()
+
+    flags=()
+    two_word_flags=()
+    local_nonpersistent_flags=()
+    flags_with_completion=()
+    flags_completion=()
+
+    flags+=("--replicas=")
+    two_word_flags+=("--replicas")
+    two_word_flags+=("-r")
+    local_nonpersistent_flags+=("--replicas=")
+    flags+=("--validate-only")
+    flags+=("-v")
+    local_nonpersistent_flags+=("--validate-only")
+    flags+=("--config-file=")
+    two_word_flags+=("--config-file")
+    two_word_flags+=("-C")
+    flags+=("--verbose")
+    flags+=("-V")
+
+    must_have_one_flag=()
+    must_have_one_noun=()
+    has_completion_function=1
+    noun_aliases=()
+}
+
 _kafkactl_alter_topic()
 {
     last_command="kafkactl_alter_topic"
@@ -375,6 +408,10 @@ _kafkactl_alter_topic()
     two_word_flags+=("--partitions")
     two_word_flags+=("-p")
     local_nonpersistent_flags+=("--partitions=")
+    flags+=("--replication-factor=")
+    two_word_flags+=("--replication-factor")
+    two_word_flags+=("-r")
+    local_nonpersistent_flags+=("--replication-factor=")
     flags+=("--validate-only")
     flags+=("-v")
     local_nonpersistent_flags+=("--validate-only")
@@ -397,6 +434,7 @@ _kafkactl_alter()
     command_aliases=()
 
     commands=()
+    commands+=("partition")
     commands+=("topic")
 
     flags=()
@@ -1275,6 +1313,10 @@ _kafkactl_root_command()
 
     commands=()
     commands+=("alter")
+    if [[ -z "${BASH_VERSION}" || "${BASH_VERSINFO[0]}" -gt 3 ]]; then
+        command_aliases+=("edit")
+        aliashash["edit"]="alter"
+    fi
     commands+=("attach")
     commands+=("completion")
     commands+=("config")
diff --git a/operations/common-operation.go b/operations/common-operation.go
index 48d2049..b6f9dc9 100644
--- a/operations/common-operation.go
+++ b/operations/common-operation.go
@@ -11,6 +11,7 @@ import (
 	"os/user"
 	"regexp"
 	"strings"
+	"time"
 )
 
 var (
@@ -45,6 +46,7 @@ type ClientContext struct {
 	Tls                TlsConfig
 	Sasl               SaslConfig
 	Kubernetes         K8sConfig
+	RequestTimeout     time.Duration
 	ClientID           string
 	KafkaVersion       sarama.KafkaVersion
 	AvroSchemaRegistry string
@@ -75,6 +77,8 @@ func CreateClientContext() (ClientContext, error) {
 	context.Tls.Insecure = viper.GetBool("contexts." + context.Name + ".tls.insecure")
 	context.ClientID = viper.GetString("contexts." + context.Name + ".clientID")
 
+	context.RequestTimeout = viper.GetDuration("contexts." + context.Name + ".requestTimeout")
+
 	if version, err := kafkaVersion(viper.GetString("contexts." + context.Name + ".kafkaVersion")); err == nil {
 		context.KafkaVersion = version
 	} else {
@@ -138,6 +142,13 @@ func CreateClusterAdmin(context *ClientContext) (sarama.ClusterAdmin, error) {
 	config.Version = context.KafkaVersion
 	config.ClientID = getClientID(context)
 
+	if context.RequestTimeout > 0 {
+		output.Debugf("using admin request timeout: %s", context.RequestTimeout.String())
+		config.Admin.Timeout = context.RequestTimeout
+	} else {
+		output.Debugf("using default admin request timeout: 3s")
+	}
+
 	if context.Tls.Enabled {
 		output.Debugf("TLS is enabled.")
 		config.Net.TLS.Enable = true
diff --git a/operations/partitions/partition-operation.go b/operations/partitions/partition-operation.go
new file mode 100644
index 0000000..b1db1ac
--- /dev/null
+++ b/operations/partitions/partition-operation.go
@@ -0,0 +1,278 @@
+package partitions
+
+import (
+	"fmt"
+	"github.com/Shopify/sarama"
+	"github.com/deviceinsight/kafkactl/operations"
+	"github.com/deviceinsight/kafkactl/output"
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type AlterPartitionFlags struct {
+	Replicas     []int32
+	ValidateOnly bool
+}
+type partition struct {
+	Id           int32
+	OldestOffset int64   `json:"oldestOffset" yaml:"oldestOffset"`
+	NewestOffset int64   `json:"newestOffset" yaml:"newestOffset"`
+	Leader       string  `json:",omitempty" yaml:",omitempty"`
+	Replicas     []int32 `json:",omitempty" yaml:",omitempty,flow"`
+	ISRs         []int32 `json:"inSyncReplicas,omitempty" yaml:"inSyncReplicas,omitempty,flow"`
+}
+
+type PartitionOperation struct {
+}
+
+func (operation *PartitionOperation) AlterPartition(topic string, partitionId int32, flags AlterPartitionFlags) error {
+
+	var (
+		context   operations.ClientContext
+		client    sarama.Client
+		admin     sarama.ClusterAdmin
+		err       error
+		exists    bool
+		partition partition
+	)
+
+	if context, err = operations.CreateClientContext(); err != nil {
+		return err
+	}
+
+	if client, err = operations.CreateClient(&context); err != nil {
+		return err
+	}
+
+	if exists, err = operations.TopicExists(&client, topic); err != nil {
+		return err
+	}
+
+	if !exists {
+		return errors.Errorf("topic '%s' does not exist", topic)
+	}
+
+	if admin, err = operations.CreateClusterAdmin(&context); err != nil {
+		return err
+	}
+
+	if partition, err = readPartition(&client, topic, partitionId); err != nil {
+		return err
+	}
+
+	if len(flags.Replicas) > 0 {
+
+		for _, replica := range flags.Replicas {
+			brokerIdFound := false
+			for _, broker := range client.Brokers() {
+				if replica == broker.ID() {
+					brokerIdFound = true
+					break
+				}
+			}
+
+			if !brokerIdFound {
+				return errors.Errorf("unknown broker id to be used as replica: %d", replica)
+			}
+		}
+
+		var replicaAssignment, err = readCurrentReplicas(&client, topic)
+		if err != nil {
+			return errors.Errorf("Unable to read current replicas for topic '%s': %v", topic, err)
+		}
+
+		replicaAssignment[partitionId] = flags.Replicas
+
+		if flags.ValidateOnly {
+			partition.Replicas = flags.Replicas
+		} else {
+			err = admin.AlterPartitionReassignments(topic, replicaAssignment)
+
+			if err != nil {
+				return errors.Errorf("Could not reassign partition replicas for topic '%s': %v", topic, err)
+			}
+
+			partitions := make([]int32, 1)
+			partitions[0] = partitionId
+
+			assignmentRunning := true
+
+			for assignmentRunning {
+				status, err := admin.ListPartitionReassignments(topic, partitions)
+				if err != nil {
+					return errors.Errorf("Could query reassignment status for topic '%s:%d': %v", topic, partitionId, err)
+				}
+
+				assignmentRunning = false
+
+				if statusTopic, ok := status[topic]; ok {
+					if statusPartition, ok := statusTopic[partitionId]; ok {
+						output.Infof("reassignment running for topic=%s partition=%d: replicas:%v addingReplicas:%v removingReplicas:%v",
+							topic, partitionId, statusPartition.Replicas, statusPartition.AddingReplicas, statusPartition.RemovingReplicas)
+						time.Sleep(5 * time.Second)
+						assignmentRunning = true
+					}
+				}
+			}
+			output.Infof("partition replicas have been reassigned")
+		}
+	}
+
+	if flags.ValidateOnly {
+		return printPartition(partition)
+	} else {
+		return nil
+	}
+}
+
+func printPartition(p partition) error {
+
+	partitionTableWriter := output.CreateTableWriter()
+
+	if err := partitionTableWriter.WriteHeader("PARTITION", "OLDEST_OFFSET", "NEWEST_OFFSET",
+		"LEADER", "REPLICAS", "IN_SYNC_REPLICAS"); err != nil {
+		return err
+	}
+
+	replicas := strings.Trim(strings.Join(strings.Fields(fmt.Sprint(p.Replicas)), ","), "[]")
+	inSyncReplicas := strings.Trim(strings.Join(strings.Fields(fmt.Sprint(p.ISRs)), ","), "[]")
+	if err := partitionTableWriter.Write(strconv.Itoa(int(p.Id)), strconv.Itoa(int(p.OldestOffset)),
+		strconv.Itoa(int(p.NewestOffset)), p.Leader, replicas, inSyncReplicas); err != nil {
+		return err
+	}
+
+	if err := partitionTableWriter.Flush(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func readPartition(client *sarama.Client, topic string, partitionId int32) (partition, error) {
+
+	var (
+		err error
+		led *sarama.Broker
+	)
+	p := partition{Id: partitionId}
+
+	if p.OldestOffset, err = (*client).GetOffset(topic, partitionId, sarama.OffsetOldest); err != nil {
+		return p, errors.Errorf("unable to read oldest offset for topic %s partition %d", topic, partitionId)
+	}
+
+	if p.NewestOffset, err = (*client).GetOffset(topic, partitionId, sarama.OffsetNewest); err != nil {
+		return p, errors.Errorf("unable to read newest offset for topic %s partition %d", topic, partitionId)
+	}
+
+	if led, err = (*client).Leader(topic, partitionId); err != nil {
+		return p, errors.Errorf("unable to read leader for topic %s partition %d", topic, partitionId)
+	} else {
+		p.Leader = led.Addr()
+	}
+
+	if p.Replicas, err = (*client).Replicas(topic, partitionId); err != nil {
+		return p, errors.Errorf("unable to read replicas for topic %s partition %d", topic, partitionId)
+	}
+	sort.Slice(p.Replicas, func(i, j int) bool { return p.Replicas[i] < p.Replicas[j] })
+
+	if p.ISRs, err = (*client).InSyncReplicas(topic, partitionId); err != nil {
+		return p, errors.Errorf("unable to read inSyncReplicas for topic %s partition %d", topic, partitionId)
+	}
+	sort.Slice(p.ISRs, func(i, j int) bool { return p.ISRs[i] < p.ISRs[j] })
+
+	return p, nil
+}
+
+func readCurrentReplicas(client *sarama.Client, topic string) ([][]int32, error) {
+	var (
+		err error
+		ps  []int32
+	)
+
+	if ps, err = (*client).Partitions(topic); err != nil {
+		return nil, err
+	}
+
+	partitionChannel := make(chan partition)
+	errChannel := make(chan error)
+
+	// read partitions in parallel
+	for _, p := range ps {
+
+		go func(partitionId int32) {
+
+			np := partition{Id: partitionId}
+
+			if np.Replicas, err = (*client).Replicas(topic, partitionId); err != nil {
+				errChannel <- errors.Errorf("unable to read replicas for topic %s partition %d", topic, partitionId)
+				return
+			}
+			sort.Slice(np.Replicas, func(i, j int) bool { return np.Replicas[i] < np.Replicas[j] })
+
+			partitionChannel <- np
+		}(p)
+	}
+
+	partitions := make([]partition, 0, len(ps))
+
+	for range ps {
+		select {
+		case partition := <-partitionChannel:
+			partitions = append(partitions, partition)
+		case err := <-errChannel:
+			return nil, err
+		}
+	}
+
+	sort.Slice(partitions, func(i, j int) bool {
+		return partitions[i].Id < partitions[j].Id
+	})
+
+	replicaAssignment := make([][]int32, 0, len(ps))
+
+	for _, p := range partitions {
+		replicaAssignment = append(replicaAssignment, p.Replicas)
+	}
+
+	return replicaAssignment, nil
+}
+
+func CompletePartitionIds(_ *cobra.Command, args []string, _ string) ([]string, cobra.ShellCompDirective) {
+
+	if len(args) != 1 {
+		return nil, cobra.ShellCompDirectiveNoFileComp
+	}
+
+	var (
+		context operations.ClientContext
+		client  sarama.Client
+		err     error
+	)
+
+	if context, err = operations.CreateClientContext(); err != nil {
+		return nil, cobra.ShellCompDirectiveError
+	}
+
+	if client, err = operations.CreateClient(&context); err != nil {
+		return nil, cobra.ShellCompDirectiveError
+	}
+
+	topicName := args[0]
+	var partitions []int32
+
+	if partitions, err = client.Partitions(topicName); err != nil {
+		return nil, cobra.ShellCompDirectiveError
+	}
+
+	partitionsString := make([]string, len(partitions))
+
+	for i, p := range partitions {
+		partitionsString[i] = strconv.Itoa(int(p))
+	}
+
+	return partitionsString, cobra.ShellCompDirectiveNoFileComp
+}
diff --git a/operations/topic-operation.go b/operations/topic-operation.go
index 376dc46..f923d2c 100644
--- a/operations/topic-operation.go
+++ b/operations/topic-operation.go
@@ -4,20 +4,23 @@ import (
 	"fmt"
 	"github.com/Shopify/sarama"
 	"github.com/deviceinsight/kafkactl/output"
+	"github.com/deviceinsight/kafkactl/util"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
+	"gopkg.in/yaml.v2"
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 )
 
-type topic struct {
+type Topic struct {
 	Name       string
-	Partitions []partition `json:",omitempty" yaml:",omitempty"`
-	Configs    []config    `json:",omitempty" yaml:",omitempty"`
+	Partitions []Partition `json:",omitempty" yaml:",omitempty"`
+	Configs    []Config    `json:",omitempty" yaml:",omitempty"`
 }
 
-type partition struct {
+type Partition struct {
 	Id           int32
 	OldestOffset int64   `json:"oldestOffset" yaml:"oldestOffset"`
 	NewestOffset int64   `json:"newestOffset" yaml:"newestOffset"`
@@ -37,7 +40,7 @@ type requestedTopicFields struct {
 
 var allFields = requestedTopicFields{partitionId: true, partitionOffset: true, partitionLeader: true, partitionReplicas: true, partitionISRs: true, config: true}
 
-type config struct {
+type Config struct {
 	Name  string
 	Value string
 }
@@ -54,9 +57,10 @@ type CreateTopicFlags struct {
 }
 
 type AlterTopicFlags struct {
-	Partitions   int32
-	ValidateOnly bool
-	Configs      []string
+	Partitions        int32
+	ReplicationFactor int16
+	ValidateOnly      bool
+	Configs           []string
 }
 
 type DescribeTopicFlags struct {
@@ -162,16 +166,21 @@ func (operation *TopicOperation) DescribeTopic(topic string, flags DescribeTopic
 
 	var t, _ = readTopic(&client, &admin, topic, allFields)
 
+	return operation.printTopic(t, flags)
+}
+
+func (operation *TopicOperation) printTopic(topic Topic, flags DescribeTopicFlags) error {
+
 	if flags.PrintConfigs {
 		if flags.OutputFormat == "json" || flags.OutputFormat == "yaml" {
-			t.Configs = nil
+			topic.Configs = nil
 		} else {
 			configTableWriter := output.CreateTableWriter()
 			if err := configTableWriter.WriteHeader("CONFIG", "VALUE"); err != nil {
 				return err
 			}
 
-			for _, c := range t.Configs {
+			for _, c := range topic.Configs {
 				if err := configTableWriter.Write(c.Name, c.Value); err != nil {
 					return err
 				}
@@ -196,9 +205,9 @@ func (operation *TopicOperation) DescribeTopic(topic string, flags DescribeTopic
 	}
 
 	if flags.OutputFormat == "json" || flags.OutputFormat == "yaml" {
-		return output.PrintObject(t, flags.OutputFormat)
+		return output.PrintObject(topic, flags.OutputFormat)
 	} else if flags.OutputFormat == "wide" || flags.OutputFormat == "" {
-		for _, p := range t.Partitions {
+		for _, p := range topic.Partitions {
 			replicas := strings.Trim(strings.Join(strings.Fields(fmt.Sprint(p.Replicas)), ","), "[]")
 			inSyncReplicas := strings.Trim(strings.Join(strings.Fields(fmt.Sprint(p.ISRs)), ","), "[]")
 			if err := partitionTableWriter.Write(strconv.Itoa(int(p.Id)), strconv.Itoa(int(p.OldestOffset)),
@@ -246,48 +255,145 @@ func (operation *TopicOperation) AlterTopic(topic string, flags AlterTopicFlags)
 		return errors.Wrap(err, "failed to create cluster admin")
 	}
 
-	var t, _ = readTopic(&client, &admin, topic, requestedTopicFields{partitionId: true, config: true})
+	var t, _ = readTopic(&client, &admin, topic, allFields)
 
 	if flags.Partitions != 0 {
 		if len(t.Partitions) > int(flags.Partitions) {
 			return errors.New("Decreasing the number of partitions is not supported")
 		}
 
-		var emptyAssignment = make([][]int32, 0)
+		if flags.ValidateOnly {
+			for len(t.Partitions) < int(flags.Partitions) {
+				t.Partitions = append(t.Partitions, Partition{Id: int32(len(t.Partitions)), NewestOffset: 0, OldestOffset: 0})
+			}
+		} else {
+			var emptyAssignment = make([][]int32, 0)
 
-		err = admin.CreatePartitions(topic, flags.Partitions, emptyAssignment, flags.ValidateOnly)
-		if err != nil {
-			return errors.Errorf("Could not create partitions for topic '%s': %v", topic, err)
+			err = admin.CreatePartitions(topic, flags.Partitions, emptyAssignment, flags.ValidateOnly)
+			if err != nil {
+				return errors.Errorf("Could not create partitions for topic '%s': %v", topic, err)
+			} else {
+				output.Infof("partitions have been created")
+			}
 		}
 	}
 
-	if len(flags.Configs) == 0 {
-		return operation.DescribeTopic(topic, DescribeTopicFlags{})
-	}
+	if flags.ReplicationFactor > 0 {
+
+		var brokers = client.Brokers()
+
+		if int(flags.ReplicationFactor) > len(brokers) {
+			return errors.Errorf("Replication factor for topic '%s' must not exceed the number of available brokers", topic)
+		}
+
+		brokerReplicaCount := make(map[int32]int)
+		for _, broker := range brokers {
+			brokerReplicaCount[broker.ID()] = 0
+		}
+
+		for _, partition := range t.Partitions {
+			for _, brokerId := range partition.Replicas {
+				brokerReplicaCount[brokerId] += 1
+			}
+		}
+
+		var replicaAssignment = make([][]int32, 0, int16(len(t.Partitions)))
+
+		for _, partition := range t.Partitions {
+
+			var replicas, err = getTargetReplicas(partition.Replicas, brokerReplicaCount, flags.ReplicationFactor)
+			if err != nil {
+				return errors.Wrap(err, "unable to determine target replicas")
+			}
+			replicaAssignment = append(replicaAssignment, replicas)
+		}
+
+		for brokerId, replicaCount := range brokerReplicaCount {
+			output.Debugf("broker %d now has %d replicas", brokerId, replicaCount)
+		}
+
+		if flags.ValidateOnly {
+			for i := range t.Partitions {
+				t.Partitions[i].Replicas = replicaAssignment[i]
+			}
+		} else {
+			err = admin.AlterPartitionReassignments(topic, replicaAssignment)
+			if err != nil {
+				return errors.Errorf("Could not reassign partition replicas for topic '%s': %v", topic, err)
+			}
+
+			partitions := make([]int32, len(t.Partitions))
+
+			for _, p := range t.Partitions {
+				partitions[0] = p.Id
+			}
+
+			assignmentRunning := true
+
+			for assignmentRunning {
+				status, err := admin.ListPartitionReassignments(topic, partitions)
+				if err != nil {
+					return errors.Errorf("Could query reassignment status for topic '%s': %v", topic, err)
+				}
 
-	mergedConfigEntries := make(map[string]*string)
+				assignmentRunning = false
+
+				if statusTopic, ok := status[topic]; ok {
+					for partitionId, statusPartition := range statusTopic {
+						output.Infof("reassignment running for topic=%s partition=%d: replicas:%v addingReplicas:%v removingReplicas:%v",
+							topic, partitionId, statusPartition.Replicas, statusPartition.AddingReplicas, statusPartition.RemovingReplicas)
+						time.Sleep(5 * time.Second)
+						assignmentRunning = true
+					}
+				} else {
+					output.Debugf("Emtpy list partition reassignment result returned (len status: %d)", len(status))
+				}
+			}
 
-	for i, config := range t.Configs {
-		mergedConfigEntries[config.Name] = &(t.Configs[i].Value)
+			output.Infof("partition replicas have been reassigned")
+		}
 	}
 
-	for _, config := range flags.Configs {
-		configParts := strings.Split(config, "=")
+	if len(flags.Configs) > 0 {
+		mergedConfigEntries := make(map[string]*string)
+
+		for i, config := range t.Configs {
+			mergedConfigEntries[config.Name] = &(t.Configs[i].Value)
+		}
+
+		for _, config := range flags.Configs {
+			configParts := strings.Split(config, "=")
+
+			if len(configParts) == 2 {
+				if len(configParts[1]) == 0 {
+					delete(mergedConfigEntries, configParts[0])
+				} else {
+					mergedConfigEntries[configParts[0]] = &configParts[1]
+				}
+			}
+		}
 
-		if len(configParts) == 2 {
-			if len(configParts[1]) == 0 {
-				delete(mergedConfigEntries, configParts[0])
+		if flags.ValidateOnly {
+			// validate only - directly alter the response object
+			t.Configs = make([]Config, 0, len(mergedConfigEntries))
+			for key, value := range mergedConfigEntries {
+				t.Configs = append(t.Configs, Config{Name: key, Value: *value})
+			}
+		} else {
+			if err = admin.AlterConfig(sarama.TopicResource, topic, mergedConfigEntries, flags.ValidateOnly); err != nil {
+				return errors.Errorf("Could not alter topic config '%s': %v", topic, err)
 			} else {
-				mergedConfigEntries[configParts[0]] = &configParts[1]
+				output.Infof("config has been altered")
 			}
 		}
 	}
 
-	if err = admin.AlterConfig(sarama.TopicResource, topic, mergedConfigEntries, flags.ValidateOnly); err != nil {
-		return errors.Errorf("Could not alter topic config '%s': %v", topic, err)
+	if flags.ValidateOnly {
+		describeFlags := DescribeTopicFlags{PrintConfigs: len(flags.Configs) > 0}
+		return operation.printTopic(t, describeFlags)
+	} else {
+		return nil
 	}
-
-	return operation.DescribeTopic(topic, DescribeTopicFlags{})
 }
 
 func (operation *TopicOperation) ListTopicsNames() ([]string, error) {
@@ -314,6 +420,52 @@ func (operation *TopicOperation) ListTopicsNames() ([]string, error) {
 	}
 }
 
+func getTargetReplicas(currentReplicas []int32, brokerReplicaCount map[int32]int, targetReplicationFactor int16) ([]int32, error) {
+
+	replicas := currentReplicas
+
+	for len(replicas) > int(targetReplicationFactor) {
+
+		sort.Slice(replicas, func(i, j int) bool {
+			brokerI := replicas[i]
+			brokerJ := replicas[j]
+			return brokerReplicaCount[brokerI] < brokerReplicaCount[brokerJ] || (brokerReplicaCount[brokerI] == brokerReplicaCount[brokerJ] && brokerI < brokerJ)
+		})
+
+		lastReplica := replicas[len(replicas)-1]
+		replicas = replicas[:len(replicas)-1]
+		brokerReplicaCount[lastReplica] -= 1
+	}
+
+	var unusedBrokerIds []int32
+
+	if len(replicas) < int(targetReplicationFactor) {
+		for brokerId := range brokerReplicaCount {
+			if !util.ContainsInt32(replicas, brokerId) {
+				unusedBrokerIds = append(unusedBrokerIds, brokerId)
+			}
+		}
+		if len(unusedBrokerIds) < (int(targetReplicationFactor) - len(replicas)) {
+			return nil, errors.New("not enough brokers")
+		}
+	}
+
+	for len(replicas) < int(targetReplicationFactor) {
+
+		sort.Slice(unusedBrokerIds, func(i, j int) bool {
+			brokerI := unusedBrokerIds[i]
+			brokerJ := unusedBrokerIds[j]
+			return brokerReplicaCount[brokerI] < brokerReplicaCount[brokerJ] || (brokerReplicaCount[brokerI] == brokerReplicaCount[brokerJ] && brokerI > brokerJ)
+		})
+
+		replicas = append(replicas, unusedBrokerIds[0])
+		brokerReplicaCount[unusedBrokerIds[0]] += 1
+		unusedBrokerIds = unusedBrokerIds[1:]
+	}
+
+	return replicas, nil
+}
+
 func (operation *TopicOperation) GetTopics(flags GetTopicsFlags) error {
 
 	var (
@@ -363,7 +515,7 @@ func (operation *TopicOperation) GetTopics(flags GetTopicsFlags) error {
 		return errors.Errorf("unknown outputFormat: %s", flags.OutputFormat)
 	}
 
-	topicChannel := make(chan topic)
+	topicChannel := make(chan Topic)
 	errChannel := make(chan error)
 
 	// read topics in parallel
@@ -378,7 +530,7 @@ func (operation *TopicOperation) GetTopics(flags GetTopicsFlags) error {
 		}(topic)
 	}
 
-	topicList := make([]topic, 0, len(topics))
+	topicList := make([]Topic, 0, len(topics))
 	for range topics {
 		select {
 		case topic := <-topicChannel:
@@ -422,13 +574,13 @@ func (operation *TopicOperation) GetTopics(flags GetTopicsFlags) error {
 	return nil
 }
 
-func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, requestedFields requestedTopicFields) (topic, error) {
+func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, requestedFields requestedTopicFields) (Topic, error) {
 	var (
 		err           error
 		ps            []int32
 		led           *sarama.Broker
 		configEntries []sarama.ConfigEntry
-		top           = topic{Name: name}
+		top           = Topic{Name: name}
 	)
 
 	if !requestedFields.partitionId {
@@ -439,7 +591,7 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 		return top, err
 	}
 
-	partitionChannel := make(chan partition)
+	partitionChannel := make(chan Partition)
 	errChannel := make(chan error)
 
 	// read partitions in parallel
@@ -447,7 +599,7 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 
 		go func(partitionId int32) {
 
-			np := partition{Id: partitionId}
+			np := Partition{Id: partitionId}
 
 			if requestedFields.partitionOffset {
 				if np.OldestOffset, err = (*client).GetOffset(name, partitionId, sarama.OffsetOldest); err != nil {
@@ -465,8 +617,9 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 				if led, err = (*client).Leader(name, partitionId); err != nil {
 					errChannel <- errors.Errorf("unable to read leader for topic %s partition %d", name, partitionId)
 					return
+				} else {
+					np.Leader = led.Addr()
 				}
-				np.Leader = led.Addr()
 			}
 
 			if requestedFields.partitionReplicas {
@@ -490,7 +643,12 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 	}
 
 	for range ps {
-		top.Partitions = append(top.Partitions, <-partitionChannel)
+		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 {
@@ -511,7 +669,7 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 		for _, configEntry := range configEntries {
 
 			if !configEntry.Default && configEntry.Source != sarama.SourceDefault {
-				entry := config{Name: configEntry.Name, Value: configEntry.Value}
+				entry := Config{Name: configEntry.Name, Value: configEntry.Value}
 				top.Configs = append(top.Configs, entry)
 			}
 		}
@@ -520,7 +678,7 @@ func readTopic(client *sarama.Client, admin *sarama.ClusterAdmin, name string, r
 	return top, nil
 }
 
-func getConfigString(configs []config) string {
+func getConfigString(configs []Config) string {
 
 	var configStrings []string
 
@@ -531,7 +689,7 @@ func getConfigString(configs []config) string {
 	return strings.Trim(strings.Join(configStrings, ","), "[]")
 }
 
-func CompleteTopicNames(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+func CompleteTopicNames(_ *cobra.Command, args []string, _ string) ([]string, cobra.ShellCompDirective) {
 
 	if len(args) != 0 {
 		return nil, cobra.ShellCompDirectiveNoFileComp
@@ -545,3 +703,9 @@ func CompleteTopicNames(cmd *cobra.Command, args []string, toComplete string) ([
 
 	return topics, cobra.ShellCompDirectiveNoFileComp
 }
+
+func TopicFromYaml(yamlString string) (Topic, error) {
+	var t Topic
+	err := yaml.Unmarshal([]byte(yamlString), &t)
+	return t, err
+}


Test patch:
diff --git a/.github/workflows/lint_test.yml b/.github/workflows/lint_test.yml
index 5ce97a5..9529109 100644
--- a/.github/workflows/lint_test.yml
+++ b/.github/workflows/lint_test.yml
@@ -38,10 +38,21 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        # cp-version 5.5.x corresponds to kafka version 2.5.x
-        cp-version: [5.5.0, 5.4.2, 5.3.3]
+        # https://docs.confluent.io/platform/current/installation/versions-interoperability.html
+        kafka_version: [2.6.0, 2.5.0, 2.4.0, 2.3.0]
+        include:
+          - kafka_version: 2.6.0
+            cp_version: 6.0.1
+          - kafka_version: 2.5.0
+            cp_version: 5.5.2
+          - kafka_version: 2.4.0
+            cp_version: 5.4.3
+          - kafka_version: 2.3.0
+            cp_version: 5.3.3
     env:
-      CP_VERSION: ${{ matrix.cp-version }}
+      CP_VERSION: ${{ matrix.cp_version }}
+      KAFKAVERSION: ${{ matrix.kafka_version }}
+      REQUESTTIMEOUT: 15s
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-go@v2
diff --git a/cmd/alter/alter-partition_test.go b/cmd/alter/alter-partition_test.go
new file mode 100644
index 0000000..f30a56c
--- /dev/null
+++ b/cmd/alter/alter-partition_test.go
@@ -0,0 +1,99 @@
+package alter_test
+
+import (
+	"github.com/Rican7/retry"
+	"github.com/Rican7/retry/backoff"
+	"github.com/Rican7/retry/strategy"
+	"github.com/deviceinsight/kafkactl/operations"
+	"github.com/deviceinsight/kafkactl/test_util"
+	"gopkg.in/errgo.v2/fmt/errors"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestAlterPartitionAutoCompletionIntegration(t *testing.T) {
+
+	test_util.StartIntegrationTest(t)
+
+	prefix := "alter-p-complete-"
+
+	topicName1 := test_util.CreateTopic(t, prefix+"a", "--partitions", "2")
+	topicName2 := test_util.CreateTopic(t, prefix+"b")
+	topicName3 := test_util.CreateTopic(t, prefix+"c")
+
+	kafkaCtl := test_util.CreateKafkaCtlCommand()
+	kafkaCtl.Verbose = false
+
+	if _, err := kafkaCtl.Execute("__complete", "alter", "partition", ""); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	outputLines := strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
+
+	test_util.AssertContains(t, topicName1, outputLines)
+	test_util.AssertContains(t, topicName2, outputLines)
+	test_util.AssertContains(t, topicName3, outputLines)
+
+	if _, err := kafkaCtl.Execute("__complete", "alter", "partition", topicName1, ""); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	outputLines = strings.Split(strings.TrimSpace(kafkaCtl.GetStdOut()), "\n")
+
+	test_util.AssertContains(t, "0", outputLines)
+	test_util.AssertContains(t, "1", outputLines)
+}
+
+func TestAlterPartitionReplicationFactorIntegration(t *testing.T) {
+
+	test_util.StartIntegrationTest(t)
+
+	prefix := "alter-p-replicas-"
+
+	topicName := test_util.CreateTopic(t, prefix, "--partitions", "2", "--replication-factor", "3")
+
+	kafkaCtl := test_util.CreateKafkaCtlCommand()
+
+	if _, err := kafkaCtl.Execute("alter", "partition", topicName, "0", "--replicas", "101,102"); err != nil {
+		test_util.AssertErrorContains(t, "version of API is not supported", err)
+		return
+	}
+
+	checkReplicas := func(attempt uint) error {
+		_, err := kafkaCtl.Execute("describe", "topic", topicName, "-o", "yaml")
+
+		if err != nil {
+			return err
+		} else {
+			topic, err := operations.TopicFromYaml(kafkaCtl.GetStdOut())
+			if err != nil {
+				return err
+			}
+
+			if len(topic.Partitions) != 2 {
+				return errors.Newf("expected 2 partitions, but was %d", len(topic.Partitions))
+			}
+
+			if len(topic.Partitions[0].Replicas) == 2 && len(topic.Partitions[1].Replicas) == 3 {
+				if topic.Partitions[0].Replicas[0] == 101 && topic.Partitions[0].Replicas[1] == 102 {
+					return nil
+				} else {
+					return errors.Newf("different brokers expected %v", topic.Partitions[0].Replicas)
+				}
+			} else {
+				return errors.Newf("replica count incorrect %v", topic.Partitions)
+			}
+		}
+	}
+
+	err := retry.Retry(
+		checkReplicas,
+		strategy.Limit(5),
+		strategy.Backoff(backoff.Linear(10*time.Millisecond)),
+	)
+
+	if err != nil {
+		t.Fatalf("could not check Replicas for topic %s: %v", topicName, err)
+	}
+}
diff --git a/cmd/alter/alter-topic_test.go b/cmd/alter/alter-topic_test.go
index 8996a89..b967927 100644
--- a/cmd/alter/alter-topic_test.go
+++ b/cmd/alter/alter-topic_test.go
@@ -1,9 +1,15 @@
 package alter_test
 
 import (
+	"github.com/Rican7/retry"
+	"github.com/Rican7/retry/backoff"
+	"github.com/Rican7/retry/strategy"
+	"github.com/deviceinsight/kafkactl/operations"
 	"github.com/deviceinsight/kafkactl/test_util"
+	"gopkg.in/errgo.v2/fmt/errors"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestAlterTopicAutoCompletionIntegration(t *testing.T) {
@@ -29,3 +35,141 @@ func TestAlterTopicAutoCompletionIntegration(t *testing.T) {
 	test_util.AssertContains(t, topicName2, outputLines)
 	test_util.AssertContains(t, topicName3, outputLines)
 }
+
+func TestAlterTopicPartitionsIntegration(t *testing.T) {
+
+	test_util.StartIntegrationTest(t)
+
+	prefix := "alter-t-partition-"
+
+	topicName := test_util.CreateTopic(t, prefix)
+
+	kafkaCtl := test_util.CreateKafkaCtlCommand()
+
+	if _, err := kafkaCtl.Execute("alter", "topic", topicName, "--partitions", "32"); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	getPartitions := func(attempt uint) error {
+		_, err := kafkaCtl.Execute("describe", "topic", topicName, "-o", "yaml")
+
+		if err != nil {
+			return err
+		} else {
+			topic, err := operations.TopicFromYaml(kafkaCtl.GetStdOut())
+			if err != nil {
+				return err
+			}
+
+			if len(topic.Partitions) == 32 {
+				return nil
+			} else {
+				return errors.Newf("only the following partitions present: %v", topic.Partitions)
+			}
+		}
+	}
+
+	err := retry.Retry(
+		getPartitions,
+		strategy.Limit(5),
+		strategy.Backoff(backoff.Linear(10*time.Millisecond)),
+	)
+
+	if err != nil {
+		t.Fatalf("could not get partitions %s: %v", topicName, err)
+	}
+}
+
+func TestAlterTopicIncreaseReplicationFactorIntegration(t *testing.T) {
+
+	test_util.StartIntegrationTest(t)
+
+	prefix := "alter-t-ireplicas-"
+
+	topicName := test_util.CreateTopic(t, prefix, "--partitions", "32")
+
+	kafkaCtl := test_util.CreateKafkaCtlCommand()
+
+	if _, err := kafkaCtl.Execute("alter", "topic", topicName, "--replication-factor", "3"); err != nil {
+		test_util.AssertErrorContains(t, "version of API is not supported", err)
+		return
+	}
+
+	checkReplicas := func(attempt uint) error {
+		_, err := kafkaCtl.Execute("describe", "topic", topicName, "-o", "yaml")
+
+		if err != nil {
+			return err
+		} else {
+			topic, err := operations.TopicFromYaml(kafkaCtl.GetStdOut())
+			if err != nil {
+				return err
+			}
+
+			for _, p := range topic.Partitions {
+				if len(p.Replicas) != 3 {
+					return errors.Newf("partition %d has %d replicas != 3", p.Id, len(p.Replicas))
+				}
+			}
+
+			return nil
+		}
+	}
+
+	err := retry.Retry(
+		checkReplicas,
+		strategy.Limit(5),
+		strategy.Backoff(backoff.Linear(10*time.Millisecond)),
+	)
+
+	if err != nil {
+		t.Fatalf("could not check Replicas for topic %s: %v", topicName, err)
+	}
+}
+
+func TestAlterTopicDecreaseReplicationFactorIntegration(t *testing.T) {
+
+	test_util.StartIntegrationTest(t)
+
+	prefix := "alter-t-dreplicas-"
+
+	topicName := test_util.CreateTopic(t, prefix, "--partitions", "32", "--replication-factor", "3")
+
+	kafkaCtl := test_util.CreateKafkaCtlCommand()
+
+	if _, err := kafkaCtl.Execute("alter", "topic", topicName, "--replication-factor", "1"); err != nil {
+		test_util.AssertErrorContains(t, "version of API is not supported", err)
+		return
+	}
+
+	checkReplicas := func(attempt uint) error {
+		_, err := kafkaCtl.Execute("describe", "topic", topicName, "-o", "yaml")
+
+		if err != nil {
+			return err
+		} else {
+			topic, err := operations.TopicFromYaml(kafkaCtl.GetStdOut())
+			if err != nil {
+				return err
+			}
+
+			for _, p := range topic.Partitions {
+				if len(p.Replicas) != 1 {
+					return errors.Newf("partition %d has %d replicas != 1", p.Id, len(p.Replicas))
+				}
+			}
+
+			return nil
+		}
+	}
+
+	err := retry.Retry(
+		checkReplicas,
+		strategy.Limit(5),
+		strategy.Backoff(backoff.Linear(10*time.Millisecond)),
+	)
+
+	if err != nil {
+		t.Fatalf("could not check Replicas for topic %s: %v", topicName, err)
+	}
+}
diff --git a/cmd/root_test.go b/cmd/root_test.go
index 9379ebf..7e3730d 100644
--- a/cmd/root_test.go
+++ b/cmd/root_test.go
@@ -39,6 +39,7 @@ func TestEnvironmentVariableLoadingAliases(t *testing.T) {
 
 	test_util.StartUnitTest(t)
 
+	_ = os.Setenv("REQUESTTIMEOUT", "30")
 	_ = os.Setenv("BROKERS", "broker1:9092 broker2:9092")
 	_ = os.Setenv("TLS_ENABLED", "true")
 	_ = os.Setenv("TLS_CA", "my-ca")
@@ -65,6 +66,7 @@ func TestEnvironmentVariableLoadingAliases(t *testing.T) {
 
 	test_util.AssertEquals(t, "broker1:9092", viper.GetStringSlice("contexts.default.brokers")[0])
 	test_util.AssertEquals(t, "broker2:9092", viper.GetStringSlice("contexts.default.brokers")[1])
+	test_util.AssertEquals(t, "30", viper.GetString("contexts.default.requestTimeout"))
 	test_util.AssertEquals(t, "true", viper.GetString("contexts.default.tls.enabled"))
 	test_util.AssertEquals(t, "my-ca", viper.GetString("contexts.default.tls.ca"))
 	test_util.AssertEquals(t, "my-cert", viper.GetString("contexts.default.tls.cert"))
diff --git a/docker/run-integration-test-matrix.sh b/docker/run-integration-test-matrix.sh
new file mode 100755
index 0000000..c42e950
--- /dev/null
+++ b/docker/run-integration-test-matrix.sh
@@ -0,0 +1,24 @@
+#!/bin/bash
+
+set -e
+set -o pipefail
+
+SCRIPT_DIR="$(dirname "$(readlink -f "$0")")"
+ROOT_DIR="$(dirname "${SCRIPT_DIR}")"
+
+kafkaVersions=`cat ${ROOT_DIR}/.github/workflows/lint_test.yml | yq -r '.jobs.integration_test.strategy.matrix.kafka_version | @csv'`
+
+for kafkaVersionQuoted in $(echo $kafkaVersions | sed "s/,/ /g")
+do
+    kafkaVersion=`echo $kafkaVersionQuoted | sed 's/"//g'`
+    cpVersion=`cat ${ROOT_DIR}/.github/workflows/lint_test.yml | sed -e "1,/kafka_version: ${kafkaVersion}/d" | head -n1 | xargs | cut -f2 -d ' '`
+
+    echo "------------------------------------------------------------"
+    echo " IT for Kafka $kafkaVersion (confluent platform: $cpVersion)"
+    echo "------------------------------------------------------------"
+    export CP_VERSION=$cpVersion
+    export KAFKAVERSION=$kafkaVersion
+
+    set +e
+    $SCRIPT_DIR/run-integration-tests.sh
+done
diff --git a/docker/run-integration-tests.sh b/docker/run-integration-tests.sh
index 68e805f..5ebee59 100755
--- a/docker/run-integration-tests.sh
+++ b/docker/run-integration-tests.sh
@@ -7,6 +7,7 @@ SCRIPT_DIR="$(dirname "$(readlink -f "$0")")"
 ROOT_DIR="$(dirname "${SCRIPT_DIR}")"
 
 echo "using confluent platform version: ${CP_VERSION}"
+echo "using kafka version: ${KAFKAVERSION}"
 
 # docker compose up
 pushd ${ROOT_DIR}
@@ -14,8 +15,8 @@ docker-compose -f ${SCRIPT_DIR}/docker-compose.yml --env-file=${SCRIPT_DIR}/.env
 
 # docker compose down
 function tearDown {
-  popd
-  docker-compose -f ${SCRIPT_DIR}/docker-compose.yml down
+  popd >/dev/null 2>&1
+  docker-compose -f ${SCRIPT_DIR}/docker-compose.yml --env-file=${SCRIPT_DIR}/.env down
 }
 trap tearDown EXIT
 
diff --git a/test_util/helpers.go b/test_util/helpers.go
index d3b0cea..cf88d65 100644
--- a/test_util/helpers.go
+++ b/test_util/helpers.go
@@ -87,7 +87,7 @@ func VerifyTopicExists(t *testing.T, topic string) {
 
 	// add a sleep here, so that the new topic is known by all
 	// brokers hopefully
-	time.Sleep(200 * time.Millisecond)
+	time.Sleep(500 * time.Millisecond)
 }
 
 func CreateConsumerGroup(t *testing.T, topic string, groupPrefix string) string {
@@ -139,5 +139,5 @@ func VerifyGroupExists(t *testing.T, group string) {
 
 	// add a sleep here, so that the new group is known by all
 	// brokers hopefully
-	time.Sleep(200 * time.Millisecond)
+	time.Sleep(500 * time.Millisecond)
 }
diff --git a/test_util/test_util.go b/test_util/test_util.go
index 05af213..94b1fbc 100644
--- a/test_util/test_util.go
+++ b/test_util/test_util.go
@@ -105,6 +105,12 @@ func AssertEquals(t *testing.T, expected, actual string) {
 	}
 }
 
+func AssertErrorContains(t *testing.T, expected string, err error) {
+	if !strings.Contains(err.Error(), expected) {
+		t.Fatalf("expected error to contain: %s\n: %v", expected, err)
+	}
+}
+
 func AssertContains(t *testing.T, expected string, array []string) {
 	if !util.ContainsString(array, expected) {
 		t.Fatalf("expected array to contain: %s\narray: %v", expected, array)