Repository: banzaicloud/kafka-operator
Pull request #299 (2020-02-13 12:49:15)

Add rebalance after new pvc was added, add new cruisecontrol controller
| Q               | A
| --------------- | ---
| Bug fix?        | yes
| New feature?    | yes
| API breaks?     | no
| Deprecations?   | no
| License         | Apache 2.0


### What's in this PR?

[NEW] New Cruise Control controller is added that will handle cruise control related tasks.
[NEW] Rebalance request to Cruise Control after a new PVC is mounted.
[FIX] Do not mount new PVC if the node cannot mount more. 


### Why?
Even if we add a PVC when an alert is firing because of a PVC not having enough space, without a rebalance the alert will keep firing.
The new controller will separate concerns from the existing controllers.


### Checklist
<!-- Place an '[x]' (no spaces) in all applicable fields. Please remove unrelated fields. -->

- [X] Implementation tested
- [x] Error handling code meets the [guideline](https://github.com/banzaicloud/pipeline/blob/master/docs/error-handling-guide.md)
- [x] Logging code meets the guideline



Patch:
diff --git a/api/v1beta1/common_types.go b/api/v1beta1/common_types.go
index 1eea94a2..1b0de5ae 100644
--- a/api/v1beta1/common_types.go
+++ b/api/v1beta1/common_types.go
@@ -35,10 +35,40 @@ type ConfigurationState string
 // PKIBackend represents an interface implementing the PKIManager
 type PKIBackend string
 
+// CruiseControlVolumeState holds information about the state of volume rebalance
+type CruiseControlVolumeState string
+
+func (r CruiseControlState) IsUpscale() bool {
+	return r == GracefulUpscaleRequired || r == GracefulUpscaleSucceeded || r == GracefulUpscaleRunning
+}
+
+func (r CruiseControlState) IsDownscale() bool {
+	return r == GracefulDownscaleRequired || r == GracefulDownscaleSucceeded || r == GracefulDownscaleRunning
+}
+
+func (r CruiseControlState) IsRunningState() bool {
+	return r == GracefulDownscaleRunning || r == GracefulUpscaleRunning
+}
+
+func (r CruiseControlState) IsRequiredState() bool {
+	return r == GracefulDownscaleRequired || r == GracefulUpscaleRequired
+}
+
+func (r CruiseControlState) Complete() CruiseControlState {
+	switch r {
+	case GracefulUpscaleRequired, GracefulUpscaleRunning:
+		return GracefulUpscaleSucceeded
+	case GracefulDownscaleRequired, GracefulDownscaleRunning:
+		return GracefulDownscaleSucceeded
+	default:
+		return r
+	}
+}
+
 const (
 	// PKIBackendCertManager invokes cert-manager for user certificate management
 	PKIBackendCertManager PKIBackend = "cert-manager"
-	// PKIBackendVault invokves vault PKI for user certificate management
+	// PKIBackendVault invokes vault PKI for user certificate management
 	PKIBackendVault PKIBackend = "vault"
 )
 
@@ -52,6 +82,19 @@ type GracefulActionState struct {
 	TaskStarted string `json:"TaskStarted,omitempty"`
 	// CruiseControlState holds the information about CC state
 	CruiseControlState CruiseControlState `json:"cruiseControlState"`
+	// VolumeStates holds the information about the CC disk rebalance states and tasks
+	VolumeStates map[string]VolumeState `json:"volumeStates"`
+}
+
+type VolumeState struct {
+	// ErrorMessage holds the information what happened with CC disk rebalance
+	ErrorMessage string `json:"errorMessage"`
+	// CruiseControlTaskId holds info about the task id ran by CC
+	CruiseControlTaskId string `json:"cruiseControlTaskId,omitempty"`
+	// TaskStarted hold the time when the execution started
+	TaskStarted string `json:"TaskStarted,omitempty"`
+	// CruiseControlVolumeState holds the information about the CC disk rebalance state
+	CruiseControlVolumeState CruiseControlVolumeState `json:"cruiseControlVolumeState"`
 }
 
 // BrokerState holds information about broker state
@@ -69,17 +112,32 @@ const (
 	Configured RackAwarenessState = "Configured"
 	// WaitingForRackAwareness states the broker is waiting for the rack awareness config
 	WaitingForRackAwareness RackAwarenessState = "WaitingForRackAwareness"
+
+	// Upscale cruise control states
+	// GracefulUpscaleRequired states that a broker upscale is required
+	GracefulUpscaleRequired CruiseControlState = "GracefulUpscaleRequired"
+	// GracefulUpscaleRunning states that the broker upscale task is still running in CC
+	GracefulUpscaleRunning CruiseControlState = "GracefulUpscaleRunning"
 	// GracefulUpscaleSucceeded states the broker is updated gracefully OR
 	// states that the broker is part of the initial cluster creation where CC topic is still in creating stage
 	GracefulUpscaleSucceeded CruiseControlState = "GracefulUpscaleSucceeded"
-	// GracefulUpscaleSucceeded states the broker is updated gracefully
+
+	// Downscale cruise control states
+	// GracefulDownscaleRequired states that a broker downscale is required
+	GracefulDownscaleRequired CruiseControlState = "GracefulDownscaleRequired"
+	// GracefulDownscaleRunning states that the broker downscale is still running in CC
+	GracefulDownscaleRunning CruiseControlState = "GracefulDownscaleRunning"
+	// GracefulUpscaleSucceeded states that the broker downscaled gracefully
 	GracefulDownscaleSucceeded CruiseControlState = "GracefulDownscaleSucceeded"
-	// GracefulUpdateRunning states the broker update task is still running in CC
-	GracefulUpdateRunning CruiseControlState = "GracefulUpdateRunning"
-	// GracefulUpdateFailed states the broker could not be updated gracefully
-	GracefulUpdateFailed CruiseControlState = "GracefulUpdateFailed"
-	// GracefulUpdateRequired states the broker requires an
-	GracefulUpdateRequired CruiseControlState = "GracefulUpdateRequired"
+
+	// Disk rebalance cruise control states
+	// GracefulDiskRebalanceRequired states that the broker volume needs a CC disk rebalance
+	GracefulDiskRebalanceRequired CruiseControlVolumeState = "GracefulDiskRebalanceRequired"
+	// GracefulDiskRebalanceRunning states that for the broker volume a CC disk rebalance is in progress
+	GracefulDiskRebalanceRunning CruiseControlVolumeState = "GracefulDiskRebalanceRunning"
+	// GracefulDiskRebalanceSucceeded states that the for the broker volume rebalance has succeeded
+	GracefulDiskRebalanceSucceeded CruiseControlVolumeState = "GracefulDiskRebalanceSucceeded"
+
 	// CruiseControlTopicNotReady states the CC required topic is not yet created
 	CruiseControlTopicNotReady CruiseControlTopicStatus = "CruiseControlTopicNotReady"
 	// CruiseControlTopicReady states the CC required topic is created
@@ -90,7 +148,7 @@ const (
 	CruiseControlTaskNotFound CruiseControlUserTaskState = "NotFound"
 	// CruiseControlTaskInExecution states the CC task is executing
 	CruiseControlTaskInExecution CruiseControlUserTaskState = "InExecution"
-	// CruiseControlTaskCompleted states the CC task compeleted successfully
+	// CruiseControlTaskCompleted states the CC task completed successfully
 	CruiseControlTaskCompleted CruiseControlUserTaskState = "Completed"
 	// CruiseControlTaskCompletedWithError states the CC task completed with error
 	CruiseControlTaskCompletedWithError CruiseControlUserTaskState = "CompletedWithError"
diff --git a/api/v1beta1/zz_generated.deepcopy.go b/api/v1beta1/zz_generated.deepcopy.go
index 643ea12e..a4ba8005 100644
--- a/api/v1beta1/zz_generated.deepcopy.go
+++ b/api/v1beta1/zz_generated.deepcopy.go
@@ -121,7 +121,7 @@ func (in *BrokerConfig) DeepCopy() *BrokerConfig {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *BrokerState) DeepCopyInto(out *BrokerState) {
 	*out = *in
-	out.GracefulActionState = in.GracefulActionState
+	in.GracefulActionState.DeepCopyInto(&out.GracefulActionState)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BrokerState.
@@ -263,6 +263,13 @@ func (in *ExternalListenerConfig) DeepCopy() *ExternalListenerConfig {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *GracefulActionState) DeepCopyInto(out *GracefulActionState) {
 	*out = *in
+	if in.VolumeStates != nil {
+		in, out := &in.VolumeStates, &out.VolumeStates
+		*out = make(map[string]VolumeState, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new GracefulActionState.
@@ -453,7 +460,7 @@ func (in *KafkaClusterStatus) DeepCopyInto(out *KafkaClusterStatus) {
 		in, out := &in.BrokersState, &out.BrokersState
 		*out = make(map[string]BrokerState, len(*in))
 		for key, val := range *in {
-			(*out)[key] = val
+			(*out)[key] = *val.DeepCopy()
 		}
 	}
 	out.RollingUpgrade = in.RollingUpgrade
@@ -633,3 +640,18 @@ func (in *VaultConfig) DeepCopy() *VaultConfig {
 	in.DeepCopyInto(out)
 	return out
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *VolumeState) DeepCopyInto(out *VolumeState) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new VolumeState.
+func (in *VolumeState) DeepCopy() *VolumeState {
+	if in == nil {
+		return nil
+	}
+	out := new(VolumeState)
+	in.DeepCopyInto(out)
+	return out
+}
diff --git a/charts/kafka-operator/templates/operator-kafka-crd.yaml b/charts/kafka-operator/templates/operator-kafka-crd.yaml
index e5917035..b6eecd4a 100644
--- a/charts/kafka-operator/templates/operator-kafka-crd.yaml
+++ b/charts/kafka-operator/templates/operator-kafka-crd.yaml
@@ -1259,8 +1259,6 @@ spec:
                       items:
                         type: string
                       type: array
-                  required:
-                    - httpsRedirect
                   type: object
                 nodeSelector:
                   additionalProperties:
@@ -1525,9 +1523,36 @@ spec:
                         description: ErrorMessage holds the information what happened
                           with CC
                         type: string
+                      volumeStates:
+                        additionalProperties:
+                          properties:
+                            TaskStarted:
+                              description: TaskStarted hold the time when the execution
+                                started
+                              type: string
+                            cruiseControlTaskId:
+                              description: CruiseControlTaskId holds info about the
+                                task id ran by CC
+                              type: string
+                            cruiseControlVolumeState:
+                              description: CruiseControlVolumeState holds the information
+                                about the CC disk rebalance state
+                              type: string
+                            errorMessage:
+                              description: ErrorMessage holds the information what
+                                happened with CC disk rebalance
+                              type: string
+                          required:
+                            - cruiseControlVolumeState
+                            - errorMessage
+                          type: object
+                        description: VolumeStates holds the information about the
+                          CC disk rebalance states and tasks
+                        type: object
                     required:
                       - cruiseControlState
                       - errorMessage
+                      - volumeStates
                     type: object
                   rackAwarenessState:
                     description: RackAwarenessState holds info about rack awareness
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
index 637f7125..5e92b8b5 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
@@ -1519,9 +1519,36 @@ spec:
                         description: ErrorMessage holds the information what happened
                           with CC
                         type: string
+                      volumeStates:
+                        additionalProperties:
+                          properties:
+                            TaskStarted:
+                              description: TaskStarted hold the time when the execution
+                                started
+                              type: string
+                            cruiseControlTaskId:
+                              description: CruiseControlTaskId holds info about the
+                                task id ran by CC
+                              type: string
+                            cruiseControlVolumeState:
+                              description: CruiseControlVolumeState holds the information
+                                about the CC disk rebalance state
+                              type: string
+                            errorMessage:
+                              description: ErrorMessage holds the information what
+                                happened with CC disk rebalance
+                              type: string
+                          required:
+                          - cruiseControlVolumeState
+                          - errorMessage
+                          type: object
+                        description: VolumeStates holds the information about the
+                          CC disk rebalance states and tasks
+                        type: object
                     required:
                     - cruiseControlState
                     - errorMessage
+                    - volumeStates
                     type: object
                   rackAwarenessState:
                     description: RackAwarenessState holds info about rack awareness
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml b/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml
index f8c28815..87e20454 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml
@@ -62,8 +62,8 @@ spec:
                     - write
                     type: string
                   patternType:
-                    description: 'patternType is the Resource Pattern Type of this ACL entry.
-                      More info: https://kafka.apache.org/20/javadoc/org/apache/kafka/common/resource/PatternType.html'
+                    description: KafkaPatternType hold the Resource Pattern Type of
+                      kafka ACL
                     enum:
                     - literal
                     - match
diff --git a/config/samples/banzaicloud_v1beta1_kafkacluster.yaml b/config/samples/banzaicloud_v1beta1_kafkacluster.yaml
index 327764a3..df801054 100644
--- a/config/samples/banzaicloud_v1beta1_kafkacluster.yaml
+++ b/config/samples/banzaicloud_v1beta1_kafkacluster.yaml
@@ -413,7 +413,7 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    # CapacityConfig describes Cruise Control config capacity.json
+    # CapacityConfig describes Cruise Control config capacity.json if unset the operator will calculate it for the cluster
     capacityConfig: |
       {
         "brokerCapacities":[
diff --git a/config/samples/kafkacluster-prometheus.yaml b/config/samples/kafkacluster-prometheus.yaml
index 90ed2093..e35ebd40 100644
--- a/config/samples/kafkacluster-prometheus.yaml
+++ b/config/samples/kafkacluster-prometheus.yaml
@@ -5,6 +5,7 @@ metadata:
   labels:
     app: kafka
     kafka_cr: kafka
+    release: prometheus-operator
 spec:
   selector:
     matchLabels:
@@ -45,6 +46,7 @@ metadata:
   labels:
     app: kafka
     kafka_cr: kafka
+    release: prometheus-operator
 spec:
   selector:
     matchLabels:
@@ -70,7 +72,7 @@ spec:
   - name: KafkaAlerts
     rules:
     - alert: BrokerOverLoaded
-      expr: avg(sum by(brokerId, kafka_cr, namespace) (rate(kafka_network_requestmetrics_requests_total[15m]))) > 30
+      expr: avg(sum by(brokerId, kafka_cr, namespace) (rate(kafka_network_requestmetrics_requests_total[15m]))) by (kafka_cr, namespace) > 30
       for: 5m
       labels:
         severity: alert
@@ -80,9 +82,9 @@ spec:
         summary: 'broker overloaded'
         # If brokerConfigGroup is defined it will override scaled broker config
         # brokerConfigGroup: 'default_group'
-        storageClass: 'standard'
+        storageClass: 'gp2'
         mountPath: '/kafkalog'
-        diskSize: '2G'
+        diskSize: '10G'
         image: 'wurstmeister/kafka:2.12-2.3.0'
         command: 'upScale'
     - alert: BrokerUnderReplicated
@@ -94,9 +96,9 @@ spec:
       annotations:
         description: 'broker {{ $labels.brokerId }} underreplicated'
         summary: 'broker underreplicated'
-        storageClass: 'standard'
+        storageClass: 'gp2'
         mountPath: '/kafkalog'
-        diskSize: '2G'
+        diskSize: '10G'
         image: 'wurstmeister/kafka:2.12-2.3.0'
         command: 'upScale'
     - alert: PartitionCountHigh
@@ -108,9 +110,9 @@ spec:
       annotations:
         description: 'broker {{ $labels.brokerId }} has high partition count'
         summary: 'high partition count'
-        storageClass: 'standard'
+        storageClass: 'gp2'
         mountPath: '/kafkalog'
-        diskSize: '2G'
+        diskSize: '10G'
         image: 'wurstmeister/kafka:2.12-2.3.0'
         command: 'upScale'
     - alert: PartitionCountLow
@@ -124,7 +126,7 @@ spec:
         summary: 'low partition count'
         command: 'downScale'
     - alert: RemainingDiskSpaceLow
-      expr: kubelet_volume_stats_available_bytes{persistentvolumeclaim=~"kafka-.*"} < 1 * 1000 * 1000 * 1000
+      expr: kubelet_volume_stats_available_bytes{persistentvolumeclaim=~"kafka-.*"} / kubelet_volume_stats_capacity_bytes{persistentvolumeclaim=~"kafka-.*"} * 100 < 30
       for: 2m
       labels:
         severity: alert
@@ -132,9 +134,9 @@ spec:
       annotations:
         description: 'broker {{ $labels.brokerId }} has low disk space'
         summary: 'low diskspace'
-        storageClass: 'standard'
+        storageClass: 'gp2'
         mountPathPrefix: '/kafkalog'
-        diskSize: '2G'
+        diskSize: '10G'
         command: 'addPvc'
 ---
 apiVersion: v1
@@ -179,6 +181,9 @@ kind: Prometheus
 metadata:
   name: kafka-prometheus
 spec:
+  podMonitorSelector:
+    matchLabels:
+      release: prometheus-operator
   serviceAccountName: prometheus
   alerting:
     alertmanagers:
@@ -195,4 +200,4 @@ spec:
   resources:
     requests:
       memory: 400Mi
-  enableAdminAPI: false
+  enableAdminAPI: false
\ No newline at end of file
diff --git a/config/samples/kafkacluster-with-istio.yaml b/config/samples/kafkacluster-with-istio.yaml
index 65b9dfa0..7f8bcfa4 100644
--- a/config/samples/kafkacluster-with-istio.yaml
+++ b/config/samples/kafkacluster-with-istio.yaml
@@ -237,21 +237,6 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    capacityConfig: |
-      {
-        "brokerCapacities":[
-          {
-            "brokerId": "-1",
-            "capacity": {
-              "DISK": "10000",
-              "CPU": "100",
-              "NW_IN": "10000",
-              "NW_OUT": "10000"
-            },
-            "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
-          }
-        ]
-      }
     clusterConfig: |
       {
         "min.insync.replicas": 3
diff --git a/config/samples/kafkacluster_with_external_ssl.yaml b/config/samples/kafkacluster_with_external_ssl.yaml
index efdf0385..2a1d70af 100644
--- a/config/samples/kafkacluster_with_external_ssl.yaml
+++ b/config/samples/kafkacluster_with_external_ssl.yaml
@@ -251,21 +251,6 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    capacityConfig: |
-      {
-        "brokerCapacities":[
-          {
-            "brokerId": "-1",
-            "capacity": {
-              "DISK": "10000",
-              "CPU": "100",
-              "NW_IN": "10000",
-              "NW_OUT": "10000"
-            },
-            "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
-          }
-        ]
-      }
     clusterConfig: |
       {
         "min.insync.replicas": 3
diff --git a/config/samples/kafkacluster_with_internal_ssl_vault.yaml b/config/samples/kafkacluster_with_internal_ssl_vault.yaml
index 809ee038..e8a39bc1 100644
--- a/config/samples/kafkacluster_with_internal_ssl_vault.yaml
+++ b/config/samples/kafkacluster_with_internal_ssl_vault.yaml
@@ -236,21 +236,6 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    capacityConfig: |
-      {
-        "brokerCapacities":[
-          {
-            "brokerId": "-1",
-            "capacity": {
-              "DISK": "10000",
-              "CPU": "100",
-              "NW_IN": "10000",
-              "NW_OUT": "10000"
-            },
-            "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
-          }
-        ]
-      }
     clusterConfig: |
       {
         "min.insync.replicas": 3
diff --git a/config/samples/kafkacluster_with_ssl_groups.yaml b/config/samples/kafkacluster_with_ssl_groups.yaml
index eb0342ff..0d48f4ba 100644
--- a/config/samples/kafkacluster_with_ssl_groups.yaml
+++ b/config/samples/kafkacluster_with_ssl_groups.yaml
@@ -239,21 +239,6 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    capacityConfig: |
-      {
-        "brokerCapacities":[
-          {
-            "brokerId": "-1",
-            "capacity": {
-              "DISK": "10000",
-              "CPU": "100",
-              "NW_IN": "10000",
-              "NW_OUT": "10000"
-            },
-            "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
-          }
-        ]
-      }
     clusterConfig: |
       {
         "min.insync.replicas": 3
diff --git a/config/samples/kafkacluster_without_ssl.yaml b/config/samples/kafkacluster_without_ssl.yaml
index 84fab418..a0b2a3ce 100644
--- a/config/samples/kafkacluster_without_ssl.yaml
+++ b/config/samples/kafkacluster_without_ssl.yaml
@@ -247,21 +247,6 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    capacityConfig: |
-      {
-        "brokerCapacities":[
-          {
-            "brokerId": "-1",
-            "capacity": {
-              "DISK": "10000",
-              "CPU": "100",
-              "NW_IN": "10000",
-              "NW_OUT": "10000"
-            },
-            "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
-          }
-        ]
-      }
     clusterConfig: |
       {
         "min.insync.replicas": 3
diff --git a/config/samples/kafkacluster_without_ssl_groups.yaml b/config/samples/kafkacluster_without_ssl_groups.yaml
index 394473ff..88d1878d 100644
--- a/config/samples/kafkacluster_without_ssl_groups.yaml
+++ b/config/samples/kafkacluster_without_ssl_groups.yaml
@@ -233,21 +233,6 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    capacityConfig: |
-      {
-        "brokerCapacities":[
-          {
-            "brokerId": "-1",
-            "capacity": {
-              "DISK": "10000",
-              "CPU": "100",
-              "NW_IN": "10000",
-              "NW_OUT": "10000"
-            },
-            "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
-          }
-        ]
-      }
     clusterConfig: |
       {
         "min.insync.replicas": 3
diff --git a/config/samples/simplekafkacluster.yaml b/config/samples/simplekafkacluster.yaml
index aa52f9b1..aebed068 100644
--- a/config/samples/simplekafkacluster.yaml
+++ b/config/samples/simplekafkacluster.yaml
@@ -234,21 +234,6 @@ spec:
       webserver.accesslog.path=access.log
       # HTTP Request Log retention days
       webserver.accesslog.retention.days=14
-    capacityConfig: |
-      {
-        "brokerCapacities":[
-          {
-            "brokerId": "-1",
-            "capacity": {
-              "DISK": "10000",
-              "CPU": "100",
-              "NW_IN": "10000",
-              "NW_OUT": "10000"
-            },
-            "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
-          }
-        ]
-      }
     clusterConfig: |
       {
         "min.insync.replicas": 3
diff --git a/controllers/cruisecontroltask_controller.go b/controllers/cruisecontroltask_controller.go
new file mode 100644
index 00000000..16d10a56
--- /dev/null
+++ b/controllers/cruisecontroltask_controller.go
@@ -0,0 +1,510 @@
+// Copyright © 2019 Banzai Cloud
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controllers
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+	"time"
+
+	"emperror.dev/errors"
+	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
+	"github.com/banzaicloud/kafka-operator/pkg/k8sutil"
+	"github.com/banzaicloud/kafka-operator/pkg/scale"
+	ccutils "github.com/banzaicloud/kafka-operator/pkg/util/cruisecontrol"
+
+	apiErrors "k8s.io/apimachinery/pkg/api/errors"
+
+	kafkav1beta1 "github.com/banzaicloud/kafka-operator/api/v1beta1"
+)
+
+// CruiseControlTaskReconciler reconciles a kafka cluster object
+type CruiseControlTaskReconciler struct {
+	client.Client
+	Scheme *runtime.Scheme
+
+	Log logr.Logger
+}
+
+// +kubebuilder:rbac:groups=kafka.banzaicloud.io,resources=kafkaclusters/status,verbs=get;update;patch
+
+func (r *CruiseControlTaskReconciler) Reconcile(request ctrl.Request) (ctrl.Result, error) {
+	ctx := context.Background()
+
+	log := r.Log.WithValues("clusterName", request.Name, "clusterNamespace", request.Namespace)
+
+	// Fetch the KafkaCluster instance
+	instance := &v1beta1.KafkaCluster{}
+	err := r.Get(ctx, request.NamespacedName, instance)
+	if err != nil {
+		if apiErrors.IsNotFound(err) {
+			// Object not found, return.  Created objects are automatically garbage collected.
+			// For additional cleanup logic use finalizers.
+			return reconciled()
+		}
+		// Error reading the object - requeue the request.
+		return requeueWithError(r.Log, err.Error(), err)
+	}
+
+	log.V(1).Info("Reconciling")
+
+	brokersWithRunningCCTask := make(map[string]v1beta1.BrokerState)
+	brokerVolumesWithRunningCCTask := make(map[string]map[string]v1beta1.VolumeState)
+	for brokerId, brokerStatus := range instance.Status.BrokersState {
+		if brokerStatus.GracefulActionState.CruiseControlState.IsRunningState() {
+			brokersWithRunningCCTask[brokerId] = brokerStatus
+		}
+
+		volumesState := make(map[string]v1beta1.VolumeState)
+		for mountPath, volumeState := range brokerStatus.GracefulActionState.VolumeStates {
+			if volumeState.CruiseControlVolumeState == v1beta1.GracefulDiskRebalanceRunning {
+				volumesState[mountPath] = volumeState
+			}
+		}
+		if len(volumesState) > 0 {
+			brokerVolumesWithRunningCCTask[brokerId] = volumesState
+		}
+	}
+	if len(brokersWithRunningCCTask) > 0 {
+		err = r.checkCCTaskState(instance, brokersWithRunningCCTask, log)
+	}
+
+	if err == nil && len(brokerVolumesWithRunningCCTask) > 0 {
+		err = r.checkVolumeCCTaskState(instance, brokerVolumesWithRunningCCTask, log)
+	}
+
+	if err != nil {
+		switch errors.Cause(err).(type) {
+		case errorfactory.CruiseControlNotReady, errorfactory.ResourceNotReady:
+			return ctrl.Result{
+				RequeueAfter: time.Duration(15) * time.Second,
+			}, nil
+		case errorfactory.CruiseControlTaskRunning:
+			return ctrl.Result{
+				RequeueAfter: time.Duration(20) * time.Second,
+			}, nil
+		case errorfactory.CruiseControlTaskTimeout, errorfactory.CruiseControlTaskFailure:
+			return ctrl.Result{
+				RequeueAfter: time.Duration(20) * time.Second,
+			}, nil
+		default:
+			return requeueWithError(log, err.Error(), err)
+		}
+	}
+
+	var brokersWithDownscaleRequired []string
+	var brokersWithUpscaleRequired []string
+	brokersWithDiskRebalanceRequired := make(map[string][]string)
+
+	for brokerId, brokerStatus := range instance.Status.BrokersState {
+
+		if brokerStatus.GracefulActionState.CruiseControlState == v1beta1.GracefulUpscaleRequired {
+			brokersWithUpscaleRequired = append(brokersWithUpscaleRequired, brokerId)
+		} else if brokerStatus.GracefulActionState.CruiseControlState == v1beta1.GracefulDownscaleRequired {
+			brokersWithDownscaleRequired = append(brokersWithDownscaleRequired, brokerId)
+		}
+
+		for mountPath, volumeState := range brokerStatus.GracefulActionState.VolumeStates {
+			if volumeState.CruiseControlVolumeState == v1beta1.GracefulDiskRebalanceRequired {
+				brokersWithDiskRebalanceRequired[brokerId] = append(brokersWithDiskRebalanceRequired[brokerId], mountPath)
+			}
+		}
+	}
+
+	var taskId, startTime string
+	if len(brokersWithUpscaleRequired) > 0 {
+		err = r.handlePodAddCCTask(instance, brokersWithUpscaleRequired, log)
+	} else if len(brokersWithDownscaleRequired) > 0 {
+		err = r.handlePodDeleteCCTask(instance, brokersWithDownscaleRequired, log)
+	} else if len(brokersWithDiskRebalanceRequired) > 0 {
+		// create new cc task, set status to running
+		taskId, startTime, err = scale.RebalanceDisks(brokersWithDiskRebalanceRequired, instance.Namespace, instance.Spec.CruiseControlConfig.CruiseControlEndpoint, instance.Name)
+		if err != nil {
+			log.Error(err, "executing disk rebalance cc task failed")
+		} else {
+
+			var brokerIds []string
+			brokersVolumeStates := make(map[string]map[string]v1beta1.VolumeState, len(brokersWithDiskRebalanceRequired))
+			for brokerId, mountPaths := range brokersWithDiskRebalanceRequired {
+
+				brokerVolumeState := make(map[string]v1beta1.VolumeState, len(mountPaths))
+				for _, mountPath := range mountPaths {
+					brokerVolumeState[mountPath] = kafkav1beta1.VolumeState{
+						CruiseControlTaskId:      taskId,
+						TaskStarted:              startTime,
+						CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceRunning,
+					}
+				}
+				if len(brokerVolumeState) > 0 {
+					brokersVolumeStates[brokerId] = brokerVolumeState
+					brokerIds = append(brokerIds, brokerId)
+				}
+
+			}
+			if len(brokersVolumeStates) > 0 {
+				err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, instance, brokersVolumeStates, log)
+			}
+		}
+	}
+
+	if err != nil {
+		switch errors.Cause(err).(type) {
+		case errorfactory.CruiseControlNotReady:
+			return ctrl.Result{
+				RequeueAfter: time.Duration(15) * time.Second,
+			}, nil
+		case errorfactory.CruiseControlTaskRunning:
+			return ctrl.Result{
+				RequeueAfter: time.Duration(20) * time.Second,
+			}, nil
+		case errorfactory.CruiseControlTaskTimeout, errorfactory.CruiseControlTaskFailure:
+			return ctrl.Result{
+				RequeueAfter: time.Duration(20) * time.Second,
+			}, nil
+		default:
+			return requeueWithError(log, err.Error(), err)
+		}
+	}
+
+	return reconciled()
+}
+func (r *CruiseControlTaskReconciler) handlePodAddCCTask(kafkaCluster *v1beta1.KafkaCluster, brokerIds []string, log logr.Logger) error {
+	uTaskId, taskStartTime, scaleErr := scale.UpScaleCluster(brokerIds, kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	if scaleErr != nil {
+		log.Info("cruise control communication error during upscaling broker(s)", "brokerId(s)", brokerIds)
+		return errorfactory.New(errorfactory.CruiseControlNotReady{}, scaleErr, fmt.Sprintf("broker id(s): %s", brokerIds))
+	}
+	statusErr := k8sutil.UpdateBrokerStatus(r.Client, brokerIds, kafkaCluster,
+		v1beta1.GracefulActionState{CruiseControlTaskId: uTaskId, CruiseControlState: v1beta1.GracefulUpscaleRunning,
+			TaskStarted: taskStartTime}, log)
+	if statusErr != nil {
+		return errors.WrapIfWithDetails(statusErr, "could not update status for broker", "id(s)", brokerIds)
+	}
+	return nil
+}
+func (r *CruiseControlTaskReconciler) handlePodDeleteCCTask(kafkaCluster *v1beta1.KafkaCluster, brokerIds []string, log logr.Logger) error {
+	uTaskId, taskStartTime, err := scale.DownsizeCluster(brokerIds,
+		kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	if err != nil {
+		log.Info("cruise control communication error during downscaling broker(s)", "id(s)", brokerIds)
+		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, fmt.Sprintf("broker(s) id(s): %s", brokerIds))
+	}
+	err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, kafkaCluster,
+		v1beta1.GracefulActionState{CruiseControlTaskId: uTaskId, CruiseControlState: v1beta1.GracefulDownscaleRunning,
+			TaskStarted: taskStartTime}, log)
+	if err != nil {
+		return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", brokerIds)
+	}
+
+	return nil
+}
+
+func (r *CruiseControlTaskReconciler) checkCCTaskState(kafkaCluster *v1beta1.KafkaCluster, brokersState map[string]v1beta1.BrokerState, log logr.Logger) error {
+	if len(brokersState) == 0 {
+		return nil
+	}
+
+	// a CC task may run for one or multiple brokers (e.g add_broker for multiple brokers)
+	// check that all brokers that we check the CC task status for have the same task id
+	var ccTaskId string
+	for _, brokerState := range brokersState {
+		if ccTaskId == "" {
+			ccTaskId = brokerState.GracefulActionState.CruiseControlTaskId
+		} else if ccTaskId != brokerState.GracefulActionState.CruiseControlTaskId {
+			return errors.New("multiple CC task ids found")
+		}
+	}
+
+	if ccTaskId == "" {
+		return errors.New("no CC task id provided to be checked")
+	}
+
+	// check cc task status
+	status, err := scale.GetCCTaskState(ccTaskId, kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	if err != nil {
+		log.Info("Cruise control communication error checking running task", "taskId", ccTaskId)
+		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
+	}
+
+	if status == v1beta1.CruiseControlTaskNotFound || status == v1beta1.CruiseControlTaskCompletedWithError {
+		// CC task failed or not found in CC,
+		// reschedule it by marking broker CruiseControlState= GracefulUpscaleRequired or GracefulDownscaleRequired
+		var brokerIds []string
+		requiredBrokerCCState := make(map[string]v1beta1.GracefulActionState, len(brokersState))
+		for brokerId, brokerState := range brokersState {
+			requiredCCState, err := r.getCorrectRequiredCCState(brokerState.GracefulActionState.CruiseControlState)
+			if err != nil {
+				return err
+			}
+
+			brokerIds = append(brokerIds, brokerId)
+			requiredBrokerCCState[brokerId] = v1beta1.GracefulActionState{
+				CruiseControlState:  requiredCCState,
+				ErrorMessage:        "Previous cc task status invalid",
+				CruiseControlTaskId: ccTaskId,
+			}
+		}
+
+		err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, kafkaCluster, requiredBrokerCCState, log)
+
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokerIds, ","))
+		}
+		return errorfactory.New(errorfactory.CruiseControlTaskFailure{}, err, "CC task failed", fmt.Sprintf("cc task id: %s", ccTaskId))
+	}
+
+	if status == v1beta1.CruiseControlTaskCompleted {
+		// cc task completed successfully
+		var brokerIds []string
+		completedBrokerCCState := make(map[string]v1beta1.GracefulActionState, len(brokersState))
+		for brokerId, brokerState := range brokersState {
+			brokerIds = append(brokerIds, brokerId)
+
+			completedBrokerCCState[brokerId] = v1beta1.GracefulActionState{
+				CruiseControlState:  brokerState.GracefulActionState.CruiseControlState.Complete(),
+				TaskStarted:         brokerState.GracefulActionState.TaskStarted,
+				CruiseControlTaskId: brokerState.GracefulActionState.CruiseControlTaskId,
+			}
+		}
+
+		err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, kafkaCluster, completedBrokerCCState, log)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokerIds, ","))
+		}
+		return nil
+	}
+	var brokersWithTimedOutCCTask []string
+	timedOutBrokerCCState := make(map[string]v1beta1.GracefulActionState)
+	for brokerId, brokerState := range brokersState {
+		if brokerState.GracefulActionState.CruiseControlState.IsRunningState() {
+			parsedTime, err := ccutils.ParseTimeStampToUnixTime(brokerState.GracefulActionState.TaskStarted)
+			if err != nil {
+				return errors.WrapIf(err, "could not parse timestamp")
+			}
+			if time.Now().Sub(parsedTime).Minutes() > kafkaCluster.Spec.CruiseControlConfig.CruiseControlTaskSpec.GetDurationMinutes() {
+				brokersWithTimedOutCCTask = append(brokersWithTimedOutCCTask, brokerId)
+				requiredCCState, err := r.getCorrectRequiredCCState(brokerState.GracefulActionState.CruiseControlState)
+				if err != nil {
+					return err
+				}
+
+				timedOutBrokerCCState[brokerId] = v1beta1.GracefulActionState{
+					CruiseControlState:  requiredCCState,
+					CruiseControlTaskId: brokerState.GracefulActionState.CruiseControlTaskId,
+					ErrorMessage:        "Timed out waiting for the task to complete",
+					TaskStarted:         brokerState.GracefulActionState.TaskStarted,
+				}
+			}
+		}
+	}
+
+	// task timed out
+	if len(brokersWithTimedOutCCTask) > 0 {
+		log.Info("Killing Cruise control task", "taskId", ccTaskId)
+		err = scale.KillCCTask(kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+		if err != nil {
+			return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
+		}
+
+		err = k8sutil.UpdateBrokerStatus(r.Client, brokersWithTimedOutCCTask, kafkaCluster, timedOutBrokerCCState, log)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokersWithTimedOutCCTask, ","))
+		}
+		return errorfactory.New(errorfactory.CruiseControlTaskTimeout{}, errors.New("cc task timed out"), fmt.Sprintf("cc task id: %s", ccTaskId))
+	}
+
+	// cc task still in progress
+	log.Info("Cruise control task is still running", "taskId", ccTaskId)
+	return errorfactory.New(errorfactory.CruiseControlTaskRunning{}, errors.New("cc task is still running"), fmt.Sprintf("cc task id: %s", ccTaskId))
+}
+
+// getCorrectRequiredCCState returns the correct Required CC state based on that we upscale or downscale
+func (r *CruiseControlTaskReconciler) getCorrectRequiredCCState(ccState kafkav1beta1.CruiseControlState) (kafkav1beta1.CruiseControlState, error) {
+	if ccState.IsDownscale() {
+		return kafkav1beta1.GracefulDownscaleRequired, nil
+	} else if ccState.IsUpscale() {
+		return kafkav1beta1.GracefulUpscaleRequired, nil
+	}
+
+	return ccState, errors.NewWithDetails("could not determine if cruise control state is upscale or downscale", "ccState", ccState)
+}
+
+//TODO merge with checkCCTaskState into one func (hi-im-aren)
+func (r *CruiseControlTaskReconciler) checkVolumeCCTaskState(kafkaCluster *v1beta1.KafkaCluster, brokersVolumesState map[string]map[string]v1beta1.VolumeState, log logr.Logger) error {
+	if len(brokersVolumesState) == 0 {
+		return nil
+	}
+
+	var ccTaskId string
+	for _, brokerVolumesState := range brokersVolumesState {
+		for _, volumeState := range brokerVolumesState {
+			if ccTaskId == "" {
+				ccTaskId = volumeState.CruiseControlTaskId
+			} else if ccTaskId != volumeState.CruiseControlTaskId {
+				return errors.New("multiple rebalance disk CC task ids found")
+			}
+		}
+	}
+
+	if ccTaskId == "" {
+		return errors.New("no CC task id provided to be checked")
+	}
+
+	// check cc task status
+	status, err := scale.GetCCTaskState(ccTaskId, kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	if err != nil {
+		log.Info("Cruise control communication error checking running task", "taskId", ccTaskId)
+		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
+	}
+
+	if status == v1beta1.CruiseControlTaskNotFound || status == v1beta1.CruiseControlTaskCompletedWithError {
+		// CC task failed or not found in CC,
+		// reschedule it by marking volume CruiseControlVolumeState=GracefulDiskRebalanceRequired
+		var brokerIds []string
+		requiredBrokerVolumesCCState := make(map[string]map[string]v1beta1.VolumeState, len(brokersVolumesState))
+		for brokerId, volumesState := range brokersVolumesState {
+			brokerIds = append(brokerIds, brokerId)
+
+			volumesState := make(map[string]v1beta1.VolumeState, len(volumesState))
+			for mountPath := range volumesState {
+				volumesState[mountPath] = kafkav1beta1.VolumeState{
+					CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceRequired,
+					ErrorMessage:             "Previous disk rebalance cc task status invalid",
+				}
+			}
+
+			requiredBrokerVolumesCCState[brokerId] = volumesState
+		}
+		err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, kafkaCluster, requiredBrokerVolumesCCState, log)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "could not update status for broker volume(s)", "id(s)", strings.Join(brokerIds, ","))
+		}
+		return errorfactory.New(errorfactory.CruiseControlTaskFailure{}, err, "CC task failed", fmt.Sprintf("cc task id: %s", ccTaskId))
+	}
+
+	if status == v1beta1.CruiseControlTaskCompleted {
+		// cc task completed successfully
+		var brokerIds []string
+		rebalanceCompletedVolumesState := make(map[string]map[string]v1beta1.VolumeState, len(brokersVolumesState))
+		for brokerId, volumesState := range brokersVolumesState {
+			brokerIds = append(brokerIds, brokerId)
+
+			volumesStateSucceeded := make(map[string]v1beta1.VolumeState, len(volumesState))
+			for mountPath, volumeState := range volumesState {
+				volumesStateSucceeded[mountPath] = kafkav1beta1.VolumeState{
+					CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceSucceeded,
+					TaskStarted:              volumeState.TaskStarted,
+					CruiseControlTaskId:      volumeState.CruiseControlTaskId,
+				}
+			}
+
+			rebalanceCompletedVolumesState[brokerId] = volumesStateSucceeded
+		}
+
+		err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, kafkaCluster, rebalanceCompletedVolumesState, log)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokerIds, ","))
+		}
+		return nil
+	}
+
+	var brokersWithTimedOutCCTask []string
+	brokersVolumesStateWithTimedOutDiskCCTask := make(map[string]map[string]v1beta1.VolumeState)
+
+	for brokerId, volumesState := range brokersVolumesState {
+		volumesStateWithTimedOutDiskCCTask := make(map[string]v1beta1.VolumeState)
+
+		for mountPath, volumeState := range volumesState {
+			if volumeState.CruiseControlVolumeState == v1beta1.GracefulDiskRebalanceRunning {
+				parsedTime, err := ccutils.ParseTimeStampToUnixTime(volumeState.TaskStarted)
+				if err != nil {
+					return errors.WrapIf(err, "could not parse timestamp")
+				}
+
+				if time.Now().Sub(parsedTime).Minutes() > kafkaCluster.Spec.CruiseControlConfig.CruiseControlTaskSpec.GetDurationMinutes() {
+					volumesStateWithTimedOutDiskCCTask[mountPath] = kafkav1beta1.VolumeState{
+						CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceRequired,
+						CruiseControlTaskId:      volumeState.CruiseControlTaskId,
+						ErrorMessage:             "Timed out waiting for the disk rebalance cc task to complete",
+						TaskStarted:              volumeState.TaskStarted,
+					}
+				}
+			}
+		}
+
+		if len(volumesStateWithTimedOutDiskCCTask) > 0 {
+			brokersWithTimedOutCCTask = append(brokersWithTimedOutCCTask, brokerId)
+			brokersVolumesStateWithTimedOutDiskCCTask[brokerId] = volumesStateWithTimedOutDiskCCTask
+		}
+
+	}
+
+	// task timed out
+	if len(brokersWithTimedOutCCTask) > 0 {
+		log.Info("Killing Cruise control task", "taskId", ccTaskId)
+		err = scale.KillCCTask(kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+		if err != nil {
+			return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
+		}
+		err = k8sutil.UpdateBrokerStatus(r.Client, brokersWithTimedOutCCTask, kafkaCluster, brokersVolumesStateWithTimedOutDiskCCTask, log)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokersWithTimedOutCCTask, ","))
+		}
+		return errorfactory.New(errorfactory.CruiseControlTaskTimeout{}, errors.New("cc task timed out"), fmt.Sprintf("cc task id: %s", ccTaskId))
+	}
+
+	// cc task still in progress
+	log.Info("Cruise control task is still running", "taskId", ccTaskId)
+	return errorfactory.New(errorfactory.CruiseControlTaskRunning{}, errors.New("cc task is still running"), fmt.Sprintf("cc task id: %s", ccTaskId))
+}
+
+// SetupCruiseControlWithManager registers cruise control controller to the manager
+func SetupCruiseControlWithManager(mgr ctrl.Manager) *ctrl.Builder {
+	builder := ctrl.NewControllerManagedBy(mgr).For(&kafkav1beta1.KafkaCluster{}).Named("CruiseControl")
+
+	builder.WithEventFilter(
+		predicate.Funcs{
+			UpdateFunc: func(e event.UpdateEvent) bool {
+				object, err := meta.Accessor(e.ObjectNew)
+				if err != nil {
+					return false
+				}
+				if _, ok := object.(*v1beta1.KafkaCluster); ok {
+					old := e.ObjectOld.(*v1beta1.KafkaCluster)
+					new := e.ObjectNew.(*v1beta1.KafkaCluster)
+					if !reflect.DeepEqual(old.Status.BrokersState, new.Status.BrokersState) ||
+						old.GetDeletionTimestamp() != new.GetDeletionTimestamp() ||
+						old.GetGeneration() != new.GetGeneration() {
+						return true
+					}
+					return false
+				}
+				return true
+			},
+		})
+
+	return builder
+}
diff --git a/controllers/kafkacluster_controller.go b/controllers/kafkacluster_controller.go
index 17bd2b11..fb43dac6 100644
--- a/controllers/kafkacluster_controller.go
+++ b/controllers/kafkacluster_controller.go
@@ -78,7 +78,7 @@ type KafkaClusterReconciler struct {
 
 func (r *KafkaClusterReconciler) Reconcile(request ctrl.Request) (ctrl.Result, error) {
 	ctx := context.Background()
-	log := r.Log.WithValues("kafkacluster", request.NamespacedName, "Request.Name", request.Name)
+	log := r.Log.WithValues("Request.Namespace", request.NamespacedName, "Request.Name", request.Name)
 
 	log.Info("Reconciling KafkaCluster")
 
@@ -333,7 +333,7 @@ func (r *KafkaClusterReconciler) updateAndFetchLatest(ctx context.Context, clust
 func SetupKafkaClusterWithManager(mgr ctrl.Manager, log logr.Logger) *ctrl.Builder {
 
 	builder := ctrl.NewControllerManagedBy(mgr).
-		For(&v1beta1.KafkaCluster{})
+		For(&v1beta1.KafkaCluster{}).Named("KafkaCluster")
 
 	kafkaWatches(builder)
 	envoyWatches(builder)
@@ -372,7 +372,8 @@ func SetupKafkaClusterWithManager(mgr ctrl.Manager, log logr.Logger) *ctrl.Build
 					new := e.ObjectNew.(*v1beta1.KafkaCluster)
 					if !reflect.DeepEqual(old.Spec, new.Spec) ||
 						old.GetDeletionTimestamp() != new.GetDeletionTimestamp() ||
-						old.GetGeneration() != new.GetGeneration() {
+						old.GetGeneration() != new.GetGeneration() ||
+						!reflect.DeepEqual(old.Status.BrokersState, new.Status.BrokersState) {
 						return true
 					}
 					return false
diff --git a/controllers/kafkatopic_controller.go b/controllers/kafkatopic_controller.go
index 2b54a043..9d222f37 100644
--- a/controllers/kafkatopic_controller.go
+++ b/controllers/kafkatopic_controller.go
@@ -61,7 +61,7 @@ func SetupKafkaTopicWithManager(mgr ctrl.Manager) error {
 	return nil
 }
 
-// blank assignment to verify that KafkaTopicReconciler implements reconcile.Reconciler
+// blank assignment to verify that KafkaTopicReconciler implements reconcile.KafkaTopicReconciler
 var _ reconcile.Reconciler = &KafkaTopicReconciler{}
 
 // KafkaTopicReconciler reconciles a KafkaTopic object
diff --git a/controllers/kafkauser_controller.go b/controllers/kafkauser_controller.go
index 61df94dd..baab6819 100644
--- a/controllers/kafkauser_controller.go
+++ b/controllers/kafkauser_controller.go
@@ -80,7 +80,7 @@ func SetupKafkaUserWithManager(mgr ctrl.Manager) error {
 	return nil
 }
 
-// blank assignment to verify that KafkaUserReconciler implements reconcile.Reconciler
+// blank assignment to verify that KafkaUserReconciler implements reconcile.kafkaUserReconciler
 var _ reconcile.Reconciler = &KafkaUserReconciler{}
 
 // KafkaUserReconciler reconciles a KafkaUser object
diff --git a/internal/alertmanager/currentalert/process.go b/internal/alertmanager/currentalert/process.go
index 1c443dbb..1dcce684 100644
--- a/internal/alertmanager/currentalert/process.go
+++ b/internal/alertmanager/currentalert/process.go
@@ -68,16 +68,19 @@ func (e *examiner) getKafkaCr() (*v1beta1.KafkaCluster, error) {
 			return nil, err
 		}
 	} else {
-		// If kafka_cr is not a valid alert label, try to get it from the persistenvolumeclaim
-		if pvcName, ok := e.Alert.Labels["persistenvolumeclaim"]; ok {
-			pvc, err := k8sutil.GetCr(string(pvcName), string(e.Alert.Labels["namespace"]), e.Client)
+		// If kafka_cr is not a valid alert label, try to get it from the persistentvolumeclaim
+		if pvcName, ok := e.Alert.Labels["persistentvolumeclaim"]; ok {
+			pvc, err := getPvc(string(pvcName), string(e.Alert.Labels["namespace"]), e.Client)
 			if err != nil {
 				return nil, err
 			}
-			kafkaCr = model.LabelValue(pvc.Labels["kafka_cr"])
-			cr, err = k8sutil.GetCr(string(kafkaCr), string(e.Alert.Labels["namespace"]), e.Client)
-			if err != nil {
-				return nil, err
+			if kafkaCr, ok := pvc.GetLabels()["kafka_cr"]; ok {
+				cr, err = k8sutil.GetCr(kafkaCr, string(e.Alert.Labels["namespace"]), e.Client)
+				if err != nil {
+					return nil, err
+				}
+			} else {
+				return nil, errors.New("persistentvolumeclaim doesn't contain kafka_cr label")
 			}
 		}
 	}
@@ -92,6 +95,10 @@ func (e *examiner) examineAlert(rollingUpgradeAlertCount int) (bool, error) {
 		return false, err
 	}
 
+	if cr == nil {
+		return false, errors.New("kafkaCR is nil")
+	}
+
 	if err := k8sutil.UpdateCrWithRollingUpgrade(rollingUpgradeAlertCount, cr, e.Client); err != nil {
 		return false, err
 	}
@@ -185,7 +192,7 @@ func addPvc(log logr.Logger, alertLabels model.LabelSet, alertAnnotations model.
 	}
 
 	// Check for skipping in case of pending or running CC task
-	ccTaskExists, err := pendingOrRunningCCTaskExists(pvc.Labels, client, log)
+	ccTaskExists, err := pendingOrRunningCCTaskExists(pvc.Labels, string(alertLabels["namespace"]), client, log)
 	if err != nil {
 		return err
 	}
@@ -208,7 +215,7 @@ func addPvc(log logr.Logger, alertLabels model.LabelSet, alertAnnotations model.
 	}
 
 	storageConfig := v1beta1.StorageConfig{
-		MountPath: pvc.Annotations["mountPathPrefix"] + "-" + randomIdentifier,
+		MountPath: string(alertAnnotations["mountPathPrefix"]) + "-" + randomIdentifier,
 		PvcSpec: &corev1.PersistentVolumeClaimSpec{
 			AccessModes: []corev1.PersistentVolumeAccessMode{
 				corev1.ReadWriteOnce,
@@ -221,7 +228,7 @@ func addPvc(log logr.Logger, alertLabels model.LabelSet, alertAnnotations model.
 			},
 		}}
 
-	err = k8sutil.AddPvToSpecificBroker(pvc.Labels["brokerId"], pvc.Labels["kafka_cr"], pvc.Labels["namespace"], &storageConfig, client)
+	err = k8sutil.AddPvToSpecificBroker(pvc.Labels["brokerId"], pvc.Labels["kafka_cr"], string(alertLabels["namespace"]), &storageConfig, client)
 	if err != nil {
 		return err
 	}
@@ -314,7 +321,7 @@ func upScale(log logr.Logger, labels model.LabelSet, annotations model.LabelSet,
 				Image: string(annotations["image"]),
 				StorageConfigs: []v1beta1.StorageConfig{
 					{
-						MountPath: string(annotations["mountPathPrefix"]),
+						MountPath: string(annotations["mountPath"]),
 						PvcSpec: &corev1.PersistentVolumeClaimSpec{
 							AccessModes: []corev1.PersistentVolumeAccessMode{
 								corev1.ReadWriteOnce,
@@ -341,17 +348,18 @@ func upScale(log logr.Logger, labels model.LabelSet, annotations model.LabelSet,
 
 // getPvc returns the given PVC object
 func getPvc(name, namespace string, client client.Client) (*corev1.PersistentVolumeClaim, error) {
-	cr := &corev1.PersistentVolumeClaim{}
+	pvc := &corev1.PersistentVolumeClaim{}
+
+	err := client.Get(context.TODO(), types.NamespacedName{Name: name, Namespace: namespace}, pvc)
 
-	err := client.Get(context.TODO(), types.NamespacedName{Name: name, Namespace: namespace}, cr)
 	if err != nil {
 		return nil, errors.WrapIfWithDetails(err, "could not get PVC from k8s", "PVCName", name, "namespace", namespace)
 	}
-	return cr, nil
+	return pvc, nil
 }
 
-func pendingOrRunningCCTaskExists(pvcLabels map[string]string, client client.Client, log logr.Logger) (bool, error) {
-	kafkaCluster, err := k8sutil.GetCr(pvcLabels["kafka_cr"], pvcLabels["namespace"], client)
+func pendingOrRunningCCTaskExists(pvcLabels map[string]string, namespace string, client client.Client, log logr.Logger) (bool, error) {
+	kafkaCluster, err := k8sutil.GetCr(pvcLabels["kafka_cr"], namespace, client)
 	if err != nil {
 		return false, err
 	}
@@ -388,7 +396,7 @@ func unboundPvcOnNodeExists(c client.Client, pvc *corev1.PersistentVolumeClaim,
 
 	for _, pvc := range kafkaPvcListOnNode.Items {
 		if pvc.Status.Phase == corev1.ClaimPending {
-			log.Info("addPvc is skipped because a PVC exists on the node which is unbound", "node:", nodeName)
+			log.Info("addPvc is skipped because a PVC exists on the node which is unbound", "node", nodeName)
 
 			return true, nil
 		}
diff --git a/main.go b/main.go
index eec4b065..ca14c8e3 100644
--- a/main.go
+++ b/main.go
@@ -124,6 +124,17 @@ func main() {
 		os.Exit(1)
 	}
 
+	kafkaClusterCCReconciler := &controllers.CruiseControlTaskReconciler{
+		Client: mgr.GetClient(),
+		Scheme: mgr.GetScheme(),
+		Log:    ctrl.Log.WithName("controller").WithName("CruiseControlTask"),
+	}
+
+	if err = controllers.SetupCruiseControlWithManager(mgr).Complete(kafkaClusterCCReconciler); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "CruiseControl")
+		os.Exit(1)
+	}
+
 	webhook.SetupServerHandlers(mgr, webhookCertDir)
 
 	// +kubebuilder:scaffold:builder
diff --git a/pkg/k8sutil/status.go b/pkg/k8sutil/status.go
index fdfaad6a..9e495067 100644
--- a/pkg/k8sutil/status.go
+++ b/pkg/k8sutil/status.go
@@ -46,6 +46,41 @@ func IsMarkedForDeletion(m metav1.ObjectMeta) bool {
 func UpdateBrokerStatus(c client.Client, brokerIds []string, cluster *v1beta1.KafkaCluster, state interface{}, logger logr.Logger) error {
 	typeMeta := cluster.TypeMeta
 
+	generateBrokerState(brokerIds, cluster, state)
+
+	err := c.Status().Update(context.Background(), cluster)
+	if apierrors.IsNotFound(err) {
+		err = c.Update(context.Background(), cluster)
+	}
+	if err != nil {
+		if !apierrors.IsConflict(err) {
+			return errors.WrapIff(err, "could not update Kafka broker(s) %s state", strings.Join(brokerIds, ","))
+		}
+		err := c.Get(context.TODO(), types.NamespacedName{
+			Namespace: cluster.Namespace,
+			Name:      cluster.Name,
+		}, cluster)
+		if err != nil {
+			return errors.WrapIf(err, "could not get config for updating status")
+		}
+
+		generateBrokerState(brokerIds, cluster, state)
+
+		err = c.Status().Update(context.Background(), cluster)
+		if apierrors.IsNotFound(err) {
+			err = c.Update(context.Background(), cluster)
+		}
+		if err != nil {
+			return errors.WrapIff(err, "could not update Kafka clusters broker(s) %s state", strings.Join(brokerIds, ","))
+		}
+	}
+	// update loses the typeMeta of the config that's used later when setting ownerrefs
+	cluster.TypeMeta = typeMeta
+	logger.Info("Kafka cluster state updated")
+	return nil
+}
+
+func generateBrokerState(brokerIds []string, cluster *banzaicloudv1beta1.KafkaCluster, state interface{}) {
 	for _, brokerId := range brokerIds {
 
 		if cluster.Status.BrokersState == nil {
@@ -54,8 +89,22 @@ func UpdateBrokerStatus(c client.Client, brokerIds []string, cluster *v1beta1.Ka
 				cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {RackAwarenessState: s}}
 			case banzaicloudv1beta1.GracefulActionState:
 				cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {GracefulActionState: s}}
+			case map[string]banzaicloudv1beta1.GracefulActionState:
+				state := s[brokerId]
+				cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {GracefulActionState: state}}
 			case banzaicloudv1beta1.ConfigurationState:
 				cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {ConfigurationState: s}}
+			case map[string]banzaicloudv1beta1.VolumeState:
+				gracefulActionState := banzaicloudv1beta1.GracefulActionState{
+					VolumeStates: s,
+				}
+				cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {GracefulActionState: gracefulActionState}}
+			case map[string]map[string]banzaicloudv1beta1.VolumeState:
+				state := s[brokerId]
+				gracefulActionState := banzaicloudv1beta1.GracefulActionState{
+					VolumeStates: state,
+				}
+				cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {GracefulActionState: gracefulActionState}}
 			}
 		} else if val, ok := cluster.Status.BrokersState[brokerId]; ok {
 			switch s := state.(type) {
@@ -63,8 +112,26 @@ func UpdateBrokerStatus(c client.Client, brokerIds []string, cluster *v1beta1.Ka
 				val.RackAwarenessState = s
 			case banzaicloudv1beta1.GracefulActionState:
 				val.GracefulActionState = s
+			case map[string]banzaicloudv1beta1.GracefulActionState:
+				state := s[brokerId]
+				val.GracefulActionState = state
 			case banzaicloudv1beta1.ConfigurationState:
 				val.ConfigurationState = s
+			case map[string]banzaicloudv1beta1.VolumeState:
+				if val.GracefulActionState.VolumeStates == nil {
+					val.GracefulActionState.VolumeStates = make(map[string]banzaicloudv1beta1.VolumeState)
+				}
+				for mountPath, volumeState := range s {
+					val.GracefulActionState.VolumeStates[mountPath] = volumeState
+				}
+			case map[string]map[string]banzaicloudv1beta1.VolumeState:
+				state := s[brokerId]
+				if val.GracefulActionState.VolumeStates == nil {
+					val.GracefulActionState.VolumeStates = make(map[string]banzaicloudv1beta1.VolumeState)
+				}
+				for mountPath, volumeState := range state {
+					val.GracefulActionState.VolumeStates[mountPath] = volumeState
+				}
 			}
 			cluster.Status.BrokersState[brokerId] = val
 		} else {
@@ -73,73 +140,25 @@ func UpdateBrokerStatus(c client.Client, brokerIds []string, cluster *v1beta1.Ka
 				cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{RackAwarenessState: s}
 			case banzaicloudv1beta1.GracefulActionState:
 				cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{GracefulActionState: s}
+			case map[string]banzaicloudv1beta1.GracefulActionState:
+				state := s[brokerId]
+				cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{GracefulActionState: state}
 			case banzaicloudv1beta1.ConfigurationState:
 				cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{ConfigurationState: s}
-			}
-		}
-	}
-
-	err := c.Status().Update(context.Background(), cluster)
-	if apierrors.IsNotFound(err) {
-		err = c.Update(context.Background(), cluster)
-	}
-	if err != nil {
-		if !apierrors.IsConflict(err) {
-			return errors.WrapIff(err, "could not update Kafka broker(s) %s state", strings.Join(brokerIds, ","))
-		}
-		err := c.Get(context.TODO(), types.NamespacedName{
-			Namespace: cluster.Namespace,
-			Name:      cluster.Name,
-		}, cluster)
-		if err != nil {
-			return errors.WrapIf(err, "could not get config for updating status")
-		}
-
-		for _, brokerId := range brokerIds {
-
-			if cluster.Status.BrokersState == nil {
-				switch s := state.(type) {
-				case banzaicloudv1beta1.RackAwarenessState:
-					cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {RackAwarenessState: s}}
-				case banzaicloudv1beta1.GracefulActionState:
-					cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {GracefulActionState: s}}
-				case banzaicloudv1beta1.ConfigurationState:
-					cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {ConfigurationState: s}}
+			case map[string]banzaicloudv1beta1.VolumeState:
+				gracefulActionState := banzaicloudv1beta1.GracefulActionState{
+					VolumeStates: s,
 				}
-			} else if val, ok := cluster.Status.BrokersState[brokerId]; ok {
-				switch s := state.(type) {
-				case banzaicloudv1beta1.RackAwarenessState:
-					val.RackAwarenessState = s
-				case banzaicloudv1beta1.GracefulActionState:
-					val.GracefulActionState = s
-				case banzaicloudv1beta1.ConfigurationState:
-					val.ConfigurationState = s
-				}
-				cluster.Status.BrokersState[brokerId] = val
-			} else {
-				switch s := state.(type) {
-				case banzaicloudv1beta1.RackAwarenessState:
-					cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{RackAwarenessState: s}
-				case banzaicloudv1beta1.GracefulActionState:
-					cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{GracefulActionState: s}
-				case banzaicloudv1beta1.ConfigurationState:
-					cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{ConfigurationState: s}
+				cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{GracefulActionState: gracefulActionState}
+			case map[string]map[string]banzaicloudv1beta1.VolumeState:
+				state := s[brokerId]
+				gracefulActionState := banzaicloudv1beta1.GracefulActionState{
+					VolumeStates: state,
 				}
+				cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{GracefulActionState: gracefulActionState}
 			}
 		}
-
-		err = c.Status().Update(context.Background(), cluster)
-		if apierrors.IsNotFound(err) {
-			err = c.Update(context.Background(), cluster)
-		}
-		if err != nil {
-			return errors.WrapIff(err, "could not update Kafka clusters broker(s) %s state", strings.Join(brokerIds, ","))
-		}
 	}
-	// update loses the typeMeta of the config that's used later when setting ownerrefs
-	cluster.TypeMeta = typeMeta
-	logger.Info("Kafka cluster state updated")
-	return nil
 }
 
 // DeleteStatus deletes the given broker state from the CR
diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go
index 032c11fc..d1aaa278 100644
--- a/pkg/resources/cruisecontrol/configmap.go
+++ b/pkg/resources/cruisecontrol/configmap.go
@@ -15,7 +15,9 @@
 package cruisecontrol
 
 import (
+	"encoding/json"
 	"fmt"
+	"strconv"
 
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -44,7 +46,7 @@ func (r *Reconciler) configMap(log logr.Logger, clientPass string) runtime.Objec
     zookeeper.connect=%s
 `, generateBootstrapServer(r.KafkaCluster.Spec.HeadlessServiceEnabled, r.KafkaCluster.Name), r.KafkaCluster.Spec.ListenersConfig.InternalListeners[0].ContainerPort, zookeeperutils.PrepareConnectionAddress(r.KafkaCluster.Spec.ZKAddresses, r.KafkaCluster.Spec.GetZkPath())) +
 				generateSSLConfig(&r.KafkaCluster.Spec.ListenersConfig, clientPass),
-			"capacity.json":       r.KafkaCluster.Spec.CruiseControlConfig.CapacityConfig,
+			"capacity.json":       GenerateCapacityConfig(r.KafkaCluster, log),
 			"clusterConfigs.json": r.KafkaCluster.Spec.CruiseControlConfig.ClusterConfig,
 			"log4j.properties": `
 log4j.rootLogger = INFO, FILE
@@ -92,3 +94,90 @@ func generateBootstrapServer(headlessEnabled bool, clusterName string) string {
 	}
 	return fmt.Sprintf(kafkautils.AllBrokerServiceTemplate, clusterName)
 }
+
+const (
+	storageConfigCPUDefaultValue   = "100"
+	storageConfigNWINDefaultValue  = "10000"
+	storageConfigNWOUTDefaultValue = "10000"
+)
+
+type CruiseControlCapacityConfig struct {
+	BrokerCapacities []BrokerCapacity `json:"brokerCapacities"`
+}
+type BrokerCapacity struct {
+	BrokerID string   `json:"brokerId"`
+	Capacity Capacity `json:"capacity"`
+	Doc      string   `json:"doc"`
+}
+type Capacity struct {
+	DISK  map[string]string `json:"DISK"`
+	CPU   string            `json:"CPU"`
+	NWIN  string            `json:"NW_IN"`
+	NWOUT string            `json:"NW_OUT"`
+}
+
+// generateCapacityConfig generates a CC capacity config with default values or returns the manually overridden value if it exists
+func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger) string {
+	log.Info("Generating capacity config")
+
+	// If there is already a config added manually, use that one
+	if kafkaCluster.Spec.CruiseControlConfig.CapacityConfig != "" {
+		return kafkaCluster.Spec.CruiseControlConfig.CapacityConfig
+	}
+
+	capacityConfig := CruiseControlCapacityConfig{}
+
+	for _, brokerState := range kafkaCluster.Spec.Brokers {
+		brokerCapacity := BrokerCapacity{
+			BrokerID: fmt.Sprintf("%d", brokerState.Id),
+			Capacity: Capacity{
+				DISK:  generateBrokerDisks(brokerState, kafkaCluster.Spec, log),
+				CPU:   storageConfigCPUDefaultValue,
+				NWIN:  storageConfigNWINDefaultValue,
+				NWOUT: storageConfigNWOUTDefaultValue,
+			},
+			Doc: "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB.",
+		}
+
+		log.Info("The following brokerCapacity was generated", "brokerCapacity", brokerCapacity)
+
+		capacityConfig.BrokerCapacities = append(capacityConfig.BrokerCapacities, brokerCapacity)
+	}
+
+	result, err := json.MarshalIndent(capacityConfig, "", "    ")
+	if err != nil {
+		log.Error(err, "Could not marshal cruise control capacity config")
+	}
+	log.Info("Generated capacity config was successful with values:", result)
+
+	return string(result)
+}
+
+func generateBrokerDisks(brokerState v1beta1.Broker, kafkaClusterSpec v1beta1.KafkaClusterSpec, log logr.Logger) map[string]string {
+	brokerDisks := map[string]string{}
+
+	// Get disks from the BrokerConfigGroup if it's in use
+	if brokerState.BrokerConfigGroup != "" {
+		brokerConfigGroup := kafkaClusterSpec.BrokerConfigGroups[brokerState.BrokerConfigGroup]
+		parseMountPathWithSize(brokerConfigGroup, log, brokerState, brokerDisks)
+	}
+
+	//Get disks from the BrokerConfig itself
+	if brokerState.BrokerConfig != nil {
+		parseMountPathWithSize(*brokerState.BrokerConfig, log, brokerState, brokerDisks)
+	}
+
+	return brokerDisks
+}
+
+func parseMountPathWithSize(brokerConfigGroup v1beta1.BrokerConfig, log logr.Logger, brokerState v1beta1.Broker, brokerDisks map[string]string) {
+	for _, storageConfig := range brokerConfigGroup.StorageConfigs {
+		int64Value, isConvertible := util.QuantityPointer(storageConfig.PvcSpec.Resources.Requests["storage"]).AsInt64()
+		if isConvertible == false {
+			log.Info("Could not convert 'storage' quantity to Int64 in brokerConfig for broker",
+				"brokerId", brokerState.Id)
+		}
+
+		brokerDisks[storageConfig.MountPath+"/kafka"] = strconv.FormatInt(int64Value, 10)
+	}
+}
diff --git a/pkg/resources/cruisecontrol/deployment.go b/pkg/resources/cruisecontrol/deployment.go
index cb831277..8aa094b7 100644
--- a/pkg/resources/cruisecontrol/deployment.go
+++ b/pkg/resources/cruisecontrol/deployment.go
@@ -15,6 +15,8 @@
 package cruisecontrol
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"fmt"
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
@@ -59,7 +61,7 @@ func (r *Reconciler) deployment(log logr.Logger, clientPass string) runtime.Obje
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels:      templates.ObjectMetaLabels(r.KafkaCluster, ccLabelSelector(r.KafkaCluster.Name)),
-					Annotations: util.MonitoringAnnotations(metricsPort),
+					Annotations: generatePodAnnotations(r.KafkaCluster, log),
 				},
 				Spec: corev1.PodSpec{
 					ServiceAccountName:            r.KafkaCluster.Spec.CruiseControlConfig.GetServiceAccount(),
@@ -172,6 +174,19 @@ fi`},
 	}
 }
 
+func generatePodAnnotations(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger) map[string]string {
+	hashedCruiseControlCapacityJson := sha256.Sum256([]byte(GenerateCapacityConfig(kafkaCluster, log)))
+
+	annotations := []map[string]string{
+		{
+			"cruiseControlCapacity.json": hex.EncodeToString(hashedCruiseControlCapacityJson[:]),
+		},
+		util.MonitoringAnnotations(metricsPort),
+	}
+
+	return util.MergeAnnotations(annotations...)
+}
+
 func generateVolumesForSSL(cluster *v1beta1.KafkaCluster) []corev1.Volume {
 	return []corev1.Volume{
 		{
diff --git a/pkg/resources/kafka/allBrokerService.go b/pkg/resources/kafka/allBrokerService.go
index bb0c13a3..696f5b45 100644
--- a/pkg/resources/kafka/allBrokerService.go
+++ b/pkg/resources/kafka/allBrokerService.go
@@ -47,11 +47,11 @@ func (r *Reconciler) allBrokerService() runtime.Object {
 	}
 
 	return &corev1.Service{
-		ObjectMeta: templates.ObjectMeta(fmt.Sprintf(kafkautils.AllBrokerServiceTemplate, r.KafkaCluster.Name), labelsForKafka(r.KafkaCluster.Name), r.KafkaCluster),
+		ObjectMeta: templates.ObjectMeta(fmt.Sprintf(kafkautils.AllBrokerServiceTemplate, r.KafkaCluster.Name), LabelsForKafka(r.KafkaCluster.Name), r.KafkaCluster),
 		Spec: corev1.ServiceSpec{
 			Type:            corev1.ServiceTypeClusterIP,
 			SessionAffinity: corev1.ServiceAffinityNone,
-			Selector:        labelsForKafka(r.KafkaCluster.Name),
+			Selector:        LabelsForKafka(r.KafkaCluster.Name),
 			Ports:           usedPorts,
 		},
 	}
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index 1d916d9f..f8c62ea7 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -110,7 +110,7 @@ func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loa
 		ObjectMeta: templates.ObjectMeta(
 			fmt.Sprintf(brokerConfigTemplate+"-%d", r.KafkaCluster.Name, id),
 			util.MergeLabels(
-				labelsForKafka(r.KafkaCluster.Name),
+				LabelsForKafka(r.KafkaCluster.Name),
 				map[string]string{"brokerId": fmt.Sprintf("%d", id)},
 			),
 			r.KafkaCluster,
diff --git a/pkg/resources/kafka/headlessService.go b/pkg/resources/kafka/headlessService.go
index d85bca79..e89e6da5 100644
--- a/pkg/resources/kafka/headlessService.go
+++ b/pkg/resources/kafka/headlessService.go
@@ -50,13 +50,13 @@ func (r *Reconciler) headlessService() runtime.Object {
 	return &corev1.Service{
 		ObjectMeta: templates.ObjectMeta(
 			fmt.Sprintf(kafkautils.HeadlessServiceTemplate, r.KafkaCluster.Name),
-			util.MergeLabels(labelsForKafka(r.KafkaCluster.Name), r.KafkaCluster.Labels),
+			util.MergeLabels(LabelsForKafka(r.KafkaCluster.Name), r.KafkaCluster.Labels),
 			r.KafkaCluster,
 		),
 		Spec: corev1.ServiceSpec{
 			Type:            corev1.ServiceTypeClusterIP,
 			SessionAffinity: corev1.ServiceAffinityNone,
-			Selector:        labelsForKafka(r.KafkaCluster.Name),
+			Selector:        LabelsForKafka(r.KafkaCluster.Name),
 			Ports:           usedPorts,
 			ClusterIP:       corev1.ClusterIPNone,
 		},
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index f5cce89b..cd9d88f3 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -43,7 +43,6 @@ import (
 	"github.com/banzaicloud/kafka-operator/pkg/scale"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	certutil "github.com/banzaicloud/kafka-operator/pkg/util/cert"
-	ccutils "github.com/banzaicloud/kafka-operator/pkg/util/cruisecontrol"
 	envoyutils "github.com/banzaicloud/kafka-operator/pkg/util/envoy"
 	istioingressutils "github.com/banzaicloud/kafka-operator/pkg/util/istioingress"
 	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
@@ -75,7 +74,7 @@ type Reconciler struct {
 
 // labelsForKafka returns the labels for selecting the resources
 // belonging to the given kafka CR name.
-func labelsForKafka(name string) map[string]string {
+func LabelsForKafka(name string) map[string]string {
 	return map[string]string{"app": "kafka", "kafka_cr": name}
 }
 
@@ -93,7 +92,7 @@ func getCreatedPvcForBroker(c client.Client, brokerID int32, namespace, crName s
 	foundPvcList := &corev1.PersistentVolumeClaimList{}
 	matchingLabels := client.MatchingLabels(
 		util.MergeLabels(
-			labelsForKafka(crName),
+			LabelsForKafka(crName),
 			map[string]string{"brokerId": fmt.Sprintf("%d", brokerID)},
 		),
 	)
@@ -158,110 +157,10 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		}
 	}
 	// Handle Pod delete
-	podList := &corev1.PodList{}
-	matchingLabels := client.MatchingLabels(labelsForKafka(r.KafkaCluster.Name))
-
-	err := r.Client.List(context.TODO(), podList, client.ListOption(client.InNamespace(r.KafkaCluster.Namespace)), client.ListOption(matchingLabels))
+	err := r.reconcileKafkaPodDelete(log)
 	if err != nil {
 		return errors.WrapIf(err, "failed to reconcile resource")
 	}
-	if len(podList.Items) > len(r.KafkaCluster.Spec.Brokers) {
-		deletedBrokers := make([]corev1.Pod, 0)
-	OUTERLOOP:
-		for _, pod := range podList.Items {
-			for _, broker := range r.KafkaCluster.Spec.Brokers {
-				if pod.Labels["brokerId"] == fmt.Sprintf("%d", broker.Id) {
-					continue OUTERLOOP
-				}
-			}
-			deletedBrokers = append(deletedBrokers, pod)
-		}
-
-		if !arePodsAlreadyDeleted(deletedBrokers, log) {
-			liveBrokers, err := scale.GetLiveKafkaBrokersFromCruiseControl(
-				generateBrokerIdsFromPodSlice(deletedBrokers),
-				r.KafkaCluster.Namespace, r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
-			if err != nil {
-				log.Error(err, "Could not query CC for ALIVE brokers")
-				return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, fmt.Sprintf("broker(s) id(s): %s", strings.Join(generateBrokerIdsFromPodSlice(deletedBrokers), ",")))
-			}
-			if len(liveBrokers) <= 0 {
-				log.Info("No alive broker found in CC. No need to decomission")
-			} else {
-				if r.KafkaCluster.Status.BrokersState[liveBrokers[0]].GracefulActionState.CruiseControlState != v1beta1.GracefulUpdateRunning &&
-					r.KafkaCluster.Status.BrokersState[liveBrokers[0]].GracefulActionState.CruiseControlState != v1beta1.GracefulDownscaleSucceeded {
-					uTaskId, taskStartTime, err := scale.DownsizeCluster(liveBrokers,
-						r.KafkaCluster.Namespace, r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
-					if err != nil {
-						log.Info(fmt.Sprintf("cruise control communication error during downscaling broker(s) id(s): %s", strings.Join(liveBrokers, ",")))
-						return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, fmt.Sprintf("broker(s) id(s): %s", strings.Join(liveBrokers, ",")))
-					}
-					err = k8sutil.UpdateBrokerStatus(r.Client, liveBrokers, r.KafkaCluster,
-						v1beta1.GracefulActionState{CruiseControlTaskId: uTaskId, CruiseControlState: v1beta1.GracefulUpdateRunning,
-							TaskStarted: taskStartTime}, log)
-					if err != nil {
-						return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(liveBrokers, ","))
-					}
-				}
-				if r.KafkaCluster.Status.BrokersState[liveBrokers[0]].GracefulActionState.CruiseControlState == v1beta1.GracefulUpdateRunning {
-					err = r.checkCCTaskState(liveBrokers,
-						r.KafkaCluster.Status.BrokersState[liveBrokers[0]], v1beta1.GracefulDownscaleSucceeded, log)
-					if err != nil {
-						return err
-					}
-				}
-			}
-		}
-
-		for _, broker := range deletedBrokers {
-			if broker.ObjectMeta.DeletionTimestamp != nil {
-				log.Info(fmt.Sprintf("Broker %s is already on terminating state", broker.Labels["brokerId"]))
-				continue
-			}
-			err = r.Client.Delete(context.TODO(), &broker)
-			if err != nil {
-				return errors.WrapIfWithDetails(err, "could not delete broker", "id", broker.Labels["brokerId"])
-			}
-			err = r.Client.Delete(context.TODO(), &corev1.ConfigMap{ObjectMeta: templates.ObjectMeta(fmt.Sprintf(brokerConfigTemplate+"-%s", r.KafkaCluster.Name, broker.Labels["brokerId"]), labelsForKafka(r.KafkaCluster.Name), r.KafkaCluster)})
-			if err != nil {
-				if apierrors.IsNotFound(err) {
-					// can happen when broker was not fully initialized and now is deleted
-					log.Info(fmt.Sprintf("ConfigMap for Broker %s not found. Continue", broker.Labels["brokerId"]))
-				} else {
-					return errors.WrapIfWithDetails(err, "could not delete configmap for broker", "id", broker.Labels["brokerId"])
-				}
-			}
-			if !r.KafkaCluster.Spec.HeadlessServiceEnabled {
-				err = r.Client.Delete(context.TODO(), &corev1.Service{ObjectMeta: templates.ObjectMeta(fmt.Sprintf("%s-%s", r.KafkaCluster.Name, broker.Labels["brokerId"]), labelsForKafka(r.KafkaCluster.Name), r.KafkaCluster)})
-				if err != nil {
-					if apierrors.IsNotFound(err) {
-						// can happen when broker was not fully initialized and now is deleted
-						log.Info(fmt.Sprintf("Service for Broker %s not found. Continue", broker.Labels["brokerId"]))
-					}
-					return errors.WrapIfWithDetails(err, "could not delete service for broker", "id", broker.Labels["brokerId"])
-				}
-			}
-			for _, volume := range broker.Spec.Volumes {
-				if strings.HasPrefix(volume.Name, kafkaDataVolumeMount) {
-					err = r.Client.Delete(context.TODO(), &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{
-						Name:      volume.PersistentVolumeClaim.ClaimName,
-						Namespace: r.KafkaCluster.Namespace,
-					}})
-					if err != nil {
-						if apierrors.IsNotFound(err) {
-							// can happen when broker was not fully initialized and now is deleted
-							log.Info(fmt.Sprintf("PVC for Broker %s not found. Continue", broker.Labels["brokerId"]))
-						}
-						return errors.WrapIfWithDetails(err, "could not delete pvc for broker", "id", broker.Labels["brokerId"])
-					}
-				}
-			}
-			err = k8sutil.DeleteStatus(r.Client, broker.Labels["brokerId"], r.KafkaCluster, log)
-			if err != nil {
-				return errors.WrapIfWithDetails(err, "could not delete status for broker", "id", broker.Labels["brokerId"])
-			}
-		}
-	}
 
 	lbIPs := make([]string, 0)
 
@@ -297,19 +196,35 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		return err
 	}
 
+	brokersVolumes := make(map[string][]*corev1.PersistentVolumeClaim, len(r.KafkaCluster.Spec.Brokers))
 	for _, broker := range r.KafkaCluster.Spec.Brokers {
 		brokerConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
 		if err != nil {
 			return errors.WrapIf(err, "failed to reconcile resource")
 		}
+
+		var brokerVolumes []*corev1.PersistentVolumeClaim
 		for _, storage := range brokerConfig.StorageConfigs {
 			o := r.pvc(broker.Id, storage, log)
-			err := r.reconcileKafkaPvc(log, o.(*corev1.PersistentVolumeClaim))
-			if err != nil {
-				return errors.WrapIfWithDetails(err, "failed to reconcile resource", "resource", o.GetObjectKind().GroupVersionKind())
-			}
+			brokerVolumes = append(brokerVolumes, o.(*corev1.PersistentVolumeClaim))
+		}
+		if len(brokerVolumes) > 0 {
+			brokersVolumes[strconv.Itoa(int(broker.Id))] = brokerVolumes
+		}
+	}
+	if len(brokersVolumes) > 0 {
+		err := r.reconcileKafkaPvc(log, brokersVolumes)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "failed to reconcile resource", "resources", "PersistentVolumeClaim")
+		}
+	}
 
+	for _, broker := range r.KafkaCluster.Spec.Brokers {
+		brokerConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
+		if err != nil {
+			return errors.WrapIf(err, "failed to reconcile resource")
 		}
+
 		if r.KafkaCluster.Spec.RackAwareness == nil {
 			o := r.configMap(broker.Id, brokerConfig, lbIPs, serverPass, clientPass, superUsers, log)
 			err := k8sutil.Reconcile(log, r.Client, o, r.KafkaCluster)
@@ -359,6 +274,128 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 	return nil
 }
 
+func (r *Reconciler) reconcileKafkaPodDelete(log logr.Logger) error {
+	podList := &corev1.PodList{}
+	matchingLabels := client.MatchingLabels(LabelsForKafka(r.KafkaCluster.Name))
+
+	err := r.Client.List(context.TODO(), podList, client.ListOption(client.InNamespace(r.KafkaCluster.Namespace)), client.ListOption(matchingLabels))
+	if err != nil {
+		return errors.WrapIf(err, "failed to reconcile resource")
+	}
+
+	deletedBrokers := make([]corev1.Pod, 0)
+OUTERLOOP:
+	for _, pod := range podList.Items {
+		for _, broker := range r.KafkaCluster.Spec.Brokers {
+			if pod.Labels["brokerId"] == fmt.Sprintf("%d", broker.Id) {
+				continue OUTERLOOP
+			}
+		}
+		deletedBrokers = append(deletedBrokers, pod)
+	}
+
+	if len(deletedBrokers) > 0 {
+		if !arePodsAlreadyDeleted(deletedBrokers, log) {
+			liveBrokers, err := scale.GetLiveKafkaBrokersFromCruiseControl(
+				generateBrokerIdsFromPodSlice(deletedBrokers),
+				r.KafkaCluster.Namespace, r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
+			if err != nil {
+				log.Error(err, "could not query CC for ALIVE brokers")
+				return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, fmt.Sprintf("broker(s) id(s): %s", strings.Join(liveBrokers, ",")))
+			}
+			if len(liveBrokers) <= 0 {
+				log.Info("No alive broker found in CC. No need to decommission")
+			} else {
+				var brokersPendingGracefulDownscale []string
+
+				for i := range liveBrokers {
+					if brokerState, ok := r.KafkaCluster.Status.BrokersState[liveBrokers[i]]; ok {
+						ccState := brokerState.GracefulActionState.CruiseControlState
+						if ccState != v1beta1.GracefulDownscaleRunning && (ccState == v1beta1.GracefulUpscaleSucceeded ||
+							ccState == v1beta1.GracefulUpscaleRequired) {
+							brokersPendingGracefulDownscale = append(brokersPendingGracefulDownscale, liveBrokers[i])
+						}
+					}
+				}
+
+				if len(brokersPendingGracefulDownscale) > 0 {
+					err = k8sutil.UpdateBrokerStatus(r.Client, brokersPendingGracefulDownscale, r.KafkaCluster,
+						v1beta1.GracefulActionState{
+							CruiseControlState: v1beta1.GracefulDownscaleRequired,
+						}, log)
+					if err != nil {
+						return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)",
+							strings.Join(brokersPendingGracefulDownscale, ","))
+					}
+				}
+
+			}
+		}
+
+		for _, broker := range deletedBrokers {
+			if broker.ObjectMeta.DeletionTimestamp != nil {
+				log.Info(fmt.Sprintf("Broker %s is already on terminating state", broker.Labels["brokerId"]))
+				continue
+			}
+
+			if brokerState, ok := r.KafkaCluster.Status.BrokersState[broker.Labels["brokerId"]]; ok &&
+				brokerState.GracefulActionState.CruiseControlState != v1beta1.GracefulDownscaleSucceeded {
+
+				if brokerState.GracefulActionState.CruiseControlState == v1beta1.GracefulDownscaleRunning {
+					log.Info("cc task is still running for broker", "brokerId", broker.Labels["brokerId"], "taskId", brokerState.GracefulActionState.CruiseControlTaskId)
+				}
+				continue
+			}
+
+			err = r.Client.Delete(context.TODO(), &broker)
+			if err != nil {
+				return errors.WrapIfWithDetails(err, "could not delete broker", "id", broker.Labels["brokerId"])
+			}
+			err = r.Client.Delete(context.TODO(), &corev1.ConfigMap{ObjectMeta: templates.ObjectMeta(fmt.Sprintf(brokerConfigTemplate+"-%s", r.KafkaCluster.Name, broker.Labels["brokerId"]), LabelsForKafka(r.KafkaCluster.Name), r.KafkaCluster)})
+			if err != nil {
+				if apierrors.IsNotFound(err) {
+					// can happen when broker was not fully initialized and now is deleted
+					log.Info(fmt.Sprintf("ConfigMap for Broker %s not found. Continue", broker.Labels["brokerId"]))
+				} else {
+					return errors.WrapIfWithDetails(err, "could not delete configmap for broker", "id", broker.Labels["brokerId"])
+				}
+			}
+			if !r.KafkaCluster.Spec.HeadlessServiceEnabled {
+				err = r.Client.Delete(context.TODO(), &corev1.Service{ObjectMeta: templates.ObjectMeta(fmt.Sprintf("%s-%s", r.KafkaCluster.Name, broker.Labels["brokerId"]), LabelsForKafka(r.KafkaCluster.Name), r.KafkaCluster)})
+				if err != nil {
+					if apierrors.IsNotFound(err) {
+						// can happen when broker was not fully initialized and now is deleted
+						log.Info(fmt.Sprintf("Service for Broker %s not found. Continue", broker.Labels["brokerId"]))
+					}
+					return errors.WrapIfWithDetails(err, "could not delete service for broker", "id", broker.Labels["brokerId"])
+				}
+			}
+			for _, volume := range broker.Spec.Volumes {
+				if strings.HasPrefix(volume.Name, kafkaDataVolumeMount) {
+					err = r.Client.Delete(context.TODO(), &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{
+						Name:      volume.PersistentVolumeClaim.ClaimName,
+						Namespace: r.KafkaCluster.Namespace,
+					}})
+					if err != nil {
+						if apierrors.IsNotFound(err) {
+							// can happen when broker was not fully initialized and now is deleted
+							log.Info(fmt.Sprintf("PVC for Broker %s not found. Continue", broker.Labels["brokerId"]))
+						}
+						return errors.WrapIfWithDetails(err, "could not delete pvc for broker", "id", broker.Labels["brokerId"])
+					}
+				}
+			}
+			err = k8sutil.DeleteStatus(r.Client, broker.Labels["brokerId"], r.KafkaCluster, log)
+			if err != nil {
+				return errors.WrapIfWithDetails(err, "could not delete status for broker", "id", broker.Labels["brokerId"])
+			}
+
+		}
+
+	}
+	return nil
+}
+
 func generateBrokerIdsFromPodSlice(pods []corev1.Pod) []string {
 	ids := make([]string, len(pods))
 	for i, broker := range pods {
@@ -511,7 +548,7 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod)
 
 	matchingLabels := client.MatchingLabels(
 		util.MergeLabels(
-			labelsForKafka(r.KafkaCluster.Name),
+			LabelsForKafka(r.KafkaCluster.Name),
 			map[string]string{"brokerId": desiredPod.Labels["brokerId"]},
 		),
 	)
@@ -523,26 +560,29 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod)
 		if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPod); err != nil {
 			return errors.WrapIf(err, "could not apply last state to annotation")
 		}
+
 		if err := r.Client.Create(context.TODO(), desiredPod); err != nil {
 			return errorfactory.New(errorfactory.APIFailure{}, err, "creating resource failed", "kind", desiredType)
 		}
+
 		// Update status to Config InSync because broker is configured to go
 		statusErr := k8sutil.UpdateBrokerStatus(r.Client, []string{desiredPod.Labels["brokerId"]}, r.KafkaCluster, v1beta1.ConfigInSync, log)
+
 		if statusErr != nil {
 			return errorfactory.New(errorfactory.StatusUpdateError{}, err, "updating status for resource failed", "kind", desiredType)
 		}
+
 		if val, ok := r.KafkaCluster.Status.BrokersState[desiredPod.Labels["brokerId"]]; ok && val.GracefulActionState.CruiseControlState != v1beta1.GracefulUpscaleSucceeded {
 			gracefulActionState := v1beta1.GracefulActionState{ErrorMessage: "CruiseControl not yet ready", CruiseControlState: v1beta1.GracefulUpscaleSucceeded}
 
 			if r.KafkaCluster.Status.CruiseControlTopicStatus == v1beta1.CruiseControlTopicReady {
-				gracefulActionState = v1beta1.GracefulActionState{ErrorMessage: "", CruiseControlState: v1beta1.GracefulUpdateRequired}
+				gracefulActionState = v1beta1.GracefulActionState{ErrorMessage: "", CruiseControlState: v1beta1.GracefulUpscaleRequired}
 			}
 			statusErr = k8sutil.UpdateBrokerStatus(r.Client, []string{desiredPod.Labels["brokerId"]}, r.KafkaCluster, gracefulActionState, log)
 			if statusErr != nil {
 				return errorfactory.New(errorfactory.StatusUpdateError{}, err, "could not update broker graceful action state")
 			}
 		}
-
 		log.Info("resource created")
 		return nil
 	} else if len(podList.Items) == 1 {
@@ -561,29 +601,10 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod)
 					return errorfactory.New(errorfactory.StatusUpdateError{}, err, "updating status for resource failed", "kind", desiredType)
 				}
 			}
-			if currentPod.Status.Phase == corev1.PodRunning && r.KafkaCluster.Status.BrokersState[brokerId].GracefulActionState.CruiseControlState == v1beta1.GracefulUpdateRequired {
-				//trigger add broker in CC
-				uTaskId, taskStartTime, scaleErr := scale.UpScaleCluster(desiredPod.Labels["brokerId"], desiredPod.Namespace, r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
-				if scaleErr != nil {
-					log.Info(fmt.Sprintf("cruise control communication error during upscaling broker id: %s", brokerId))
-					return errorfactory.New(errorfactory.CruiseControlNotReady{}, scaleErr, fmt.Sprintf("broker id: %s", brokerId))
-				}
-				statusErr := k8sutil.UpdateBrokerStatus(r.Client, []string{brokerId}, r.KafkaCluster,
-					v1beta1.GracefulActionState{CruiseControlTaskId: uTaskId, CruiseControlState: v1beta1.GracefulUpdateRunning,
-						TaskStarted: taskStartTime}, log)
-				if statusErr != nil {
-					return errors.WrapIfWithDetails(err, "could not update status for broker", "id", brokerId)
-				}
-			}
-			if r.KafkaCluster.Status.BrokersState[brokerId].GracefulActionState.CruiseControlState == v1beta1.GracefulUpdateRunning {
-				err = r.checkCCTaskState([]string{brokerId}, r.KafkaCluster.Status.BrokersState[brokerId], v1beta1.GracefulUpscaleSucceeded, log)
-				if err != nil {
-					return err
-				}
-			}
 		} else {
 			return errorfactory.New(errorfactory.InternalError{}, errors.New("reconcile failed"), fmt.Sprintf("could not find status for the given broker id, %s", brokerId))
 		}
+
 	} else {
 		return errorfactory.New(errorfactory.TooManyResources{}, errors.New("reconcile failed"), "more then one matching pod found", "labels", matchingLabels)
 	}
@@ -634,7 +655,7 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod)
 			if r.KafkaCluster.Status.State == v1beta1.KafkaClusterRollingUpgrading {
 				// Check if any kafka pod is in terminating or pending state
 				podList := &corev1.PodList{}
-				matchingLabels := client.MatchingLabels(labelsForKafka(r.KafkaCluster.Name))
+				matchingLabels := client.MatchingLabels(LabelsForKafka(r.KafkaCluster.Name))
 				err := r.Client.List(context.TODO(), podList, client.ListOption(client.InNamespace(r.KafkaCluster.Namespace)), client.ListOption(matchingLabels))
 				if err != nil {
 					return errors.WrapIf(err, "failed to reconcile resource")
@@ -686,138 +707,97 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod)
 
 }
 
-func (r *Reconciler) checkCCTaskState(brokerIds []string, brokerState v1beta1.BrokerState, cruiseControlState v1beta1.CruiseControlState, log logr.Logger) error {
+func (r *Reconciler) reconcileKafkaPvc(log logr.Logger, brokersDesiredPvcs map[string][]*corev1.PersistentVolumeClaim) error {
+	brokersVolumesState := make(map[string]map[string]v1beta1.VolumeState)
+	var brokerIds []string
 
-	// check cc task status
-	status, err := scale.GetCCTaskState(brokerState.GracefulActionState.CruiseControlTaskId,
-		r.KafkaCluster.Namespace, r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
-	if err != nil {
-		log.Info(fmt.Sprintf("Cruise control communication error checking running task: %s", brokerState.GracefulActionState.CruiseControlTaskId))
-		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
-	}
-	if status == v1beta1.CruiseControlTaskNotFound || status == v1beta1.CruiseControlTaskCompletedWithError {
-		// CC task failed or not found in CC,
-		// reschedule it by marking broker CruiseControlState=GracefulUpdateRequired
-		err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, r.KafkaCluster,
-			v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRequired,
-				ErrorMessage: "Previous cc task status invalid",
-			}, log)
-		if err != nil {
-			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokerIds, ","))
-		}
-		return errorfactory.New(errorfactory.CruiseControlTaskFailure{}, err, "CC task failed", fmt.Sprintf("cc task id: %s", brokerState.GracefulActionState.CruiseControlTaskId))
-	}
-	if status == v1beta1.CruiseControlTaskCompleted {
-		// cc task completed successfully
-		err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, r.KafkaCluster,
-			v1beta1.GracefulActionState{CruiseControlState: cruiseControlState,
-				TaskStarted:         brokerState.GracefulActionState.TaskStarted,
-				CruiseControlTaskId: brokerState.GracefulActionState.CruiseControlTaskId,
-			}, log)
-		if err != nil {
-			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokerIds, ","))
-		}
-		return nil
-	}
-	// cc task still in progress
-	parsedTime, err := ccutils.ParseTimeStampToUnixTime(brokerState.GracefulActionState.TaskStarted)
-	if err != nil {
-		return errors.WrapIf(err, "could not parse timestamp")
-	}
-	if time.Now().Sub(parsedTime).Minutes() < r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlTaskSpec.GetDurationMinutes() {
-		err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, r.KafkaCluster,
-			v1beta1.GracefulActionState{TaskStarted: brokerState.GracefulActionState.TaskStarted,
-				CruiseControlTaskId: brokerState.GracefulActionState.CruiseControlTaskId,
-				CruiseControlState:  brokerState.GracefulActionState.CruiseControlState,
-			}, log)
-		if err != nil {
-			return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokerIds, ","))
-		}
-		log.Info(fmt.Sprintf("Cruise control task: %s is still running", brokerState.GracefulActionState.CruiseControlTaskId))
-		return errorfactory.New(errorfactory.CruiseControlTaskRunning{}, errors.New("cc task is still running"), fmt.Sprintf("cc task id: %s", brokerState.GracefulActionState.CruiseControlTaskId))
-	}
-	// task timed out
-	log.Info(fmt.Sprintf("Killing Cruise control task: %s", brokerState.GracefulActionState.CruiseControlTaskId))
-	err = scale.KillCCTask(r.KafkaCluster.Namespace, r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
-	if err != nil {
-		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
-	}
-	err = k8sutil.UpdateBrokerStatus(r.Client, brokerIds, r.KafkaCluster,
-		v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateFailed,
-			CruiseControlTaskId: brokerState.GracefulActionState.CruiseControlTaskId,
-			ErrorMessage:        "Timed out waiting for the task to complete",
-			TaskStarted:         brokerState.GracefulActionState.TaskStarted,
-		}, log)
-	if err != nil {
-		return errors.WrapIfWithDetails(err, "could not update status for broker(s)", "id(s)", strings.Join(brokerIds, ","))
-	}
-	return errorfactory.New(errorfactory.CruiseControlTaskTimeout{}, errors.New("cc task timed out"), fmt.Sprintf("cc task id: %s", brokerState.GracefulActionState.CruiseControlTaskId))
-}
+	for brokerId, desiredPvcs := range brokersDesiredPvcs {
+		brokerVolumesState := make(map[string]v1beta1.VolumeState)
 
-func (r *Reconciler) reconcileKafkaPvc(log logr.Logger, desiredPvc *corev1.PersistentVolumeClaim) error {
-	var currentPvc = desiredPvc.DeepCopy()
-	desiredType := reflect.TypeOf(desiredPvc)
-	log = log.WithValues("kind", desiredType)
-	log.V(1).Info("searching with label because name is empty")
+		pvcList := &corev1.PersistentVolumeClaimList{}
 
-	pvcList := &corev1.PersistentVolumeClaimList{}
+		matchingLabels := client.MatchingLabels(
+			util.MergeLabels(
+				LabelsForKafka(r.KafkaCluster.Name),
+				map[string]string{"brokerId": brokerId},
+			),
+		)
 
-	matchingLabels := client.MatchingLabels(
-		util.MergeLabels(
-			labelsForKafka(r.KafkaCluster.Name),
-			map[string]string{"brokerId": desiredPvc.Labels["brokerId"]},
-		),
-	)
-	err := r.Client.List(context.TODO(), pvcList,
-		client.InNamespace(currentPvc.Namespace), matchingLabels)
-	if err != nil && len(pvcList.Items) == 0 {
-		return errorfactory.New(errorfactory.APIFailure{}, err, "getting resource failed", "kind", desiredType)
-	}
-	mountPath := currentPvc.Annotations["mountPath"]
+		for _, desiredPvc := range desiredPvcs {
+			currentPvc := desiredPvc.DeepCopy()
+			desiredType := reflect.TypeOf(desiredPvc)
+			log = log.WithValues("kind", desiredType)
+			log.V(1).Info("searching with label because name is empty")
 
-	// Creating the first PersistentVolume For Pod
-	if len(pvcList.Items) == 0 {
-		if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPvc); err != nil {
-			return errors.WrapIf(err, "could not apply last state to annotation")
-		}
-		if err := r.Client.Create(context.TODO(), desiredPvc); err != nil {
-			return errorfactory.New(errorfactory.APIFailure{}, err, "creating resource failed", "kind", desiredType)
-		}
-		log.Info("resource created")
-		return nil
-	}
-	alreadyCreated := false
-	for _, pvc := range pvcList.Items {
-		if mountPath == pvc.Annotations["mountPath"] {
-			currentPvc = pvc.DeepCopy()
-			alreadyCreated = true
-			break
-		}
-	}
-	if !alreadyCreated {
-		// Creating the 2+ PersistentVolumes for Pod
-		if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPvc); err != nil {
-			return errors.WrapIf(err, "could not apply last state to annotation")
-		}
-		if err := r.Client.Create(context.TODO(), desiredPvc); err != nil {
-			return errorfactory.New(errorfactory.APIFailure{}, err, "creating resource failed", "kind", desiredType)
-		}
-		return nil
-	}
-	if err == nil {
-		if k8sutil.CheckIfObjectUpdated(log, desiredType, currentPvc, desiredPvc) {
+			err := r.Client.List(context.TODO(), pvcList,
+				client.InNamespace(currentPvc.Namespace), matchingLabels)
+			if err != nil && len(pvcList.Items) == 0 {
+				return errorfactory.New(errorfactory.APIFailure{}, err, "getting resource failed", "kind", desiredType)
+			}
+
+			mountPath := currentPvc.Annotations["mountPath"]
+			// Creating the first PersistentVolume For Pod
+			if len(pvcList.Items) == 0 {
+				if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPvc); err != nil {
+					return errors.WrapIf(err, "could not apply last state to annotation")
+				}
+				if err := r.Client.Create(context.TODO(), desiredPvc); err != nil {
+					return errorfactory.New(errorfactory.APIFailure{}, err, "creating resource failed", "kind", desiredType)
+				}
+				log.Info("resource created")
+				continue
+			}
+
+			alreadyCreated := false
+			for _, pvc := range pvcList.Items {
+				if mountPath == pvc.Annotations["mountPath"] {
+					currentPvc = pvc.DeepCopy()
+					alreadyCreated = true
+					break
+				}
+			}
 
-			if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPvc); err != nil {
-				return errors.WrapIf(err, "could not apply last state to annotation")
+			if !alreadyCreated {
+				// Creating the 2+ PersistentVolumes for Pod
+				if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPvc); err != nil {
+					return errors.WrapIf(err, "could not apply last state to annotation")
+				}
+				if err := r.Client.Create(context.TODO(), desiredPvc); err != nil {
+					return errorfactory.New(errorfactory.APIFailure{}, err, "creating resource failed", "kind", desiredType)
+				}
+				brokerVolumesState[desiredPvc.Annotations["mountPath"]] = v1beta1.VolumeState{CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceRequired}
+				continue
 			}
-			desiredPvc = currentPvc
+			if err == nil {
+				if k8sutil.CheckIfObjectUpdated(log, desiredType, currentPvc, desiredPvc) {
 
-			if err := r.Client.Update(context.TODO(), desiredPvc); err != nil {
-				return errorfactory.New(errorfactory.APIFailure{}, err, "updating resource failed", "kind", desiredType)
+					if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPvc); err != nil {
+						return errors.WrapIf(err, "could not apply last state to annotation")
+					}
+					desiredPvc = currentPvc
+
+					if err := r.Client.Update(context.TODO(), desiredPvc); err != nil {
+						return errorfactory.New(errorfactory.APIFailure{}, err, "updating resource failed", "kind", desiredType)
+					}
+					log.Info("resource updated")
+				}
 			}
-			log.Info("resource updated")
+		}
+
+		if len(brokerVolumesState) > 0 {
+			brokerIds = append(brokerIds, brokerId)
+			brokersVolumesState[brokerId] = brokerVolumesState
+		}
+
+	}
+
+	if len(brokersVolumesState) > 0 {
+		err := k8sutil.UpdateBrokerStatus(r.Client, brokerIds, r.KafkaCluster, brokersVolumesState, log)
+		if err != nil {
+			return err
 		}
 	}
+
 	return nil
 }
 
@@ -827,8 +807,17 @@ func GetBrokersWithPendingOrRunningCCTask(kafkaCluster *v1beta1.KafkaCluster) []
 	var brokerIDs []int32
 	for i := range kafkaCluster.Spec.Brokers {
 		if state, ok := kafkaCluster.Status.BrokersState[strconv.Itoa(int(kafkaCluster.Spec.Brokers[i].Id))]; ok {
-			if state.GracefulActionState.CruiseControlState == v1beta1.GracefulUpdateRequired || (state.GracefulActionState.CruiseControlTaskId != "" && state.GracefulActionState.CruiseControlState == v1beta1.GracefulUpdateRunning) {
+			if state.GracefulActionState.CruiseControlState.IsRequiredState() ||
+				(state.GracefulActionState.CruiseControlTaskId != "" && state.GracefulActionState.CruiseControlState.IsRunningState()) {
 				brokerIDs = append(brokerIDs, kafkaCluster.Spec.Brokers[i].Id)
+			} else {
+				// Check if the volumes are rebalancing
+				for _, volumeState := range state.GracefulActionState.VolumeStates {
+					if volumeState.CruiseControlVolumeState == v1beta1.GracefulDiskRebalanceRequired ||
+						(volumeState.CruiseControlTaskId != "" && volumeState.CruiseControlVolumeState == v1beta1.GracefulDiskRebalanceRunning) {
+						brokerIDs = append(brokerIDs, kafkaCluster.Spec.Brokers[i].Id)
+					}
+				}
 			}
 		}
 	}
diff --git a/pkg/resources/kafka/pod.go b/pkg/resources/kafka/pod.go
index 5ab12e41..85d5022e 100644
--- a/pkg/resources/kafka/pod.go
+++ b/pkg/resources/kafka/pod.go
@@ -86,7 +86,7 @@ fi
 		ObjectMeta: templates.ObjectMetaWithGeneratedNameAndAnnotations(
 			fmt.Sprintf("%s-%d-", r.KafkaCluster.Name, id),
 			util.MergeLabels(
-				labelsForKafka(r.KafkaCluster.Name),
+				LabelsForKafka(r.KafkaCluster.Name),
 				map[string]string{"brokerId": fmt.Sprintf("%d", id)},
 			),
 			util.MergeAnnotations(
@@ -247,7 +247,7 @@ func generatePodAntiAffinity(clusterName string, hardRuleEnabled bool) *corev1.P
 			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
 				{
 					LabelSelector: &metav1.LabelSelector{
-						MatchLabels: labelsForKafka(clusterName),
+						MatchLabels: LabelsForKafka(clusterName),
 					},
 					TopologyKey: "kubernetes.io/hostname",
 				},
@@ -260,7 +260,7 @@ func generatePodAntiAffinity(clusterName string, hardRuleEnabled bool) *corev1.P
 					Weight: int32(100),
 					PodAffinityTerm: corev1.PodAffinityTerm{
 						LabelSelector: &metav1.LabelSelector{
-							MatchLabels: labelsForKafka(clusterName),
+							MatchLabels: LabelsForKafka(clusterName),
 						},
 						TopologyKey: "kubernetes.io/hostname",
 					},
diff --git a/pkg/resources/kafka/pvc.go b/pkg/resources/kafka/pvc.go
index 256fcfe8..f4982d19 100644
--- a/pkg/resources/kafka/pvc.go
+++ b/pkg/resources/kafka/pvc.go
@@ -30,7 +30,7 @@ func (r *Reconciler) pvc(id int32, storage v1beta1.StorageConfig, log logr.Logge
 		ObjectMeta: templates.ObjectMetaWithGeneratedNameAndAnnotations(
 			fmt.Sprintf(brokerStorageTemplate, r.KafkaCluster.Name, id),
 			util.MergeLabels(
-				labelsForKafka(r.KafkaCluster.Name),
+				LabelsForKafka(r.KafkaCluster.Name),
 				map[string]string{"brokerId": fmt.Sprintf("%d", id)},
 			),
 			map[string]string{"mountPath": storage.MountPath}, r.KafkaCluster),
diff --git a/pkg/resources/kafka/service.go b/pkg/resources/kafka/service.go
index a29728c2..96b2bd23 100644
--- a/pkg/resources/kafka/service.go
+++ b/pkg/resources/kafka/service.go
@@ -60,14 +60,14 @@ func (r *Reconciler) service(id int32, log logr.Logger) runtime.Object {
 	return &corev1.Service{
 		ObjectMeta: templates.ObjectMeta(fmt.Sprintf("%s-%d", r.KafkaCluster.Name, id),
 			util.MergeLabels(
-				labelsForKafka(r.KafkaCluster.Name),
+				LabelsForKafka(r.KafkaCluster.Name),
 				map[string]string{"brokerId": fmt.Sprintf("%d", id)},
 			),
 			r.KafkaCluster),
 		Spec: corev1.ServiceSpec{
 			Type:            corev1.ServiceTypeClusterIP,
 			SessionAffinity: corev1.ServiceAffinityNone,
-			Selector:        util.MergeLabels(labelsForKafka(r.KafkaCluster.Name), map[string]string{"brokerId": fmt.Sprintf("%d", id)}),
+			Selector:        util.MergeLabels(LabelsForKafka(r.KafkaCluster.Name), map[string]string{"brokerId": fmt.Sprintf("%d", id)}),
 			Ports:           usedPorts,
 		},
 	}
diff --git a/pkg/scale/scale.go b/pkg/scale/scale.go
index 0f502dc9..aba0016a 100644
--- a/pkg/scale/scale.go
+++ b/pkg/scale/scale.go
@@ -21,12 +21,12 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
-	"strconv"
 	"strings"
 
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+
 	banzaicloudv1beta1 "github.com/banzaicloud/kafka-operator/api/v1beta1"
 	bcutil "github.com/banzaicloud/kafka-operator/pkg/util"
-	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
 )
 
 const (
@@ -35,7 +35,7 @@ const (
 	addBrokerAction         = "add_broker"
 	getTaskListAction       = "user_tasks"
 	kafkaClusterStateAction = "kafka_cluster_state"
-	clusterLoad             = "load"
+	clusterLoadAction       = "load"
 	rebalanceAction         = "rebalance"
 	killProposalAction      = "stop_proposal_execution"
 	serviceNameTemplate     = "%s-cruisecontrol-svc"
@@ -108,53 +108,65 @@ func parseCCErrorFromResp(input io.Reader) (string, error) {
 	return errorFromResponse.ErrorMessage, err
 }
 
-func isKafkaBrokerReady(brokerId, namespace, ccEndpoint, clusterName string) (bool, error) {
-
-	running := false
-
+func isKafkaBrokerDiskReady(brokerIdsWithMountPath map[string][]string, namespace, ccEndpoint, clusterName string) (bool, error) {
 	options := map[string]string{
 		"json": "true",
 	}
 
-	rsp, err := getCruiseControl(clusterLoad, namespace, options, ccEndpoint, clusterName)
+	rsp, err := getCruiseControl(kafkaClusterStateAction, namespace, options, ccEndpoint, clusterName)
 	if err != nil {
-		log.Error(err, "can't work with cruise-control because it is not ready")
-		return running, err
+		keyVals := []interface{}{
+			"namespace", namespace,
+			"clusterName", clusterName,
+			//"brokerId", brokerIds,
+			//"path", mountPath,
+		}
+		log.Error(err, "can't check if broker disk is ready as Cruise Control not ready", keyVals...)
+		return false, err
 	}
 
 	body, err := ioutil.ReadAll(rsp.Body)
 	if err != nil {
-		return running, err
+		return false, err
 	}
 
 	err = rsp.Body.Close()
 	if err != nil {
-		return running, err
+		return false, err
 	}
 
 	var response struct {
-		Brokers []struct {
-			Broker      float64
-			BrokerState string
+		KafkaBrokerState struct {
+			OnlineLogDirsByBrokerId map[string][]string
 		}
 	}
-
 	err = json.Unmarshal(body, &response)
 	if err != nil {
-		return running, err
-	}
-
-	bIdToFloat, _ := strconv.ParseFloat(brokerId, 32)
-
-	for _, broker := range response.Brokers {
-		if broker.Broker == bIdToFloat &&
-			broker.BrokerState == brokerAlive {
-			log.Info("broker is available in cruise-control", "brokerId", brokerId)
-			running = true
-			break
+		return false, err
+	}
+
+	for brokerId, volumeMounts := range brokerIdsWithMountPath {
+		if ccOnlineLogDirs, ok := response.KafkaBrokerState.OnlineLogDirsByBrokerId[brokerId]; ok {
+			for _, volumeMount := range volumeMounts {
+				match := false
+				for _, ccOnlineLogDir := range ccOnlineLogDirs {
+					if strings.HasPrefix(strings.TrimSpace(ccOnlineLogDir), strings.TrimSpace(volumeMount)) {
+						match = true
+						break
+					}
+				}
+
+				if !match {
+					return false, nil
+				}
+			}
+
+		} else {
+			return false, nil
 		}
 	}
-	return running, nil
+
+	return true, nil
 }
 
 // Get brokers status from CC from a provided list of broker ids
@@ -164,7 +176,7 @@ func GetLiveKafkaBrokersFromCruiseControl(brokerIds []string, namespace, ccEndpo
 		"json": "true",
 	}
 
-	rsp, err := getCruiseControl(clusterLoad, namespace, options, ccEndpoint, clusterName)
+	rsp, err := getCruiseControl(clusterLoadAction, namespace, options, ccEndpoint, clusterName)
 	if err != nil {
 		log.Error(err, "can't work with cruise-control because it is not ready")
 		return nil, err
@@ -254,12 +266,14 @@ func GetBrokerIDWithLeastPartition(namespace, ccEndpoint, clusterName string) (s
 }
 
 // UpScaleCluster upscales Kafka cluster
-func UpScaleCluster(brokerId, namespace, ccEndpoint, clusterName string) (string, string, error) {
+func UpScaleCluster(brokerIds []string, namespace, ccEndpoint, clusterName string) (string, string, error) {
 
-	ready, err := isKafkaBrokerReady(brokerId, namespace, ccEndpoint, clusterName)
+	liveBrokers, err := GetLiveKafkaBrokersFromCruiseControl(brokerIds, namespace, ccEndpoint, clusterName)
 	if err != nil {
 		return "", "", err
 	}
+	ready := bcutil.AreStringSlicesIdentical(liveBrokers, brokerIds)
+
 	if !ready {
 		return "", "", errors.New("broker is not ready yet")
 	}
@@ -267,7 +281,7 @@ func UpScaleCluster(brokerId, namespace, ccEndpoint, clusterName string) (string
 	options := map[string]string{
 		"json":     "true",
 		"dryrun":   "false",
-		"brokerid": brokerId,
+		"brokerid": strings.Join(brokerIds, ","),
 	}
 
 	uResp, err := postCruiseControl(addBrokerAction, namespace, options, ccEndpoint, clusterName)
@@ -331,7 +345,15 @@ func DownsizeCluster(brokerIds []string, namespace, ccEndpoint, clusterName stri
 }
 
 // RebalanceDisks rebalances Kafka broker replicas between disks using CC
-func RebalanceDisks(namespace, ccEndpoint, clusterName string) (string, string, error) {
+func RebalanceDisks(brokerIdsWithMountPath map[string][]string, namespace, ccEndpoint, clusterName string) (string, string, error) {
+
+	ready, err := isKafkaBrokerDiskReady(brokerIdsWithMountPath, namespace, ccEndpoint, clusterName)
+	if err != nil {
+		return "", "", err
+	}
+	if !ready {
+		return "", "", errors.New("broker disk is not ready yet")
+	}
 
 	options := map[string]string{
 		"dryrun":         "false",
diff --git a/pkg/util/util.go b/pkg/util/util.go
index 5c88f7a9..61fdaae7 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -16,17 +16,24 @@ package util
 
 import (
 	"math/rand"
+	"reflect"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
 
 	"emperror.dev/errors"
 	"github.com/imdario/mergo"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/intstr"
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 )
 
+const (
+	symbolSet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
+)
+
 // IntstrPointer generate IntOrString pointer from int
 func IntstrPointer(i int) *intstr.IntOrString {
 	is := intstr.FromInt(i)
@@ -58,6 +65,11 @@ func StringPointer(s string) *string {
 	return &s
 }
 
+// QuantityPointer generates Quantity pointer from Quantity
+func QuantityPointer(q resource.Quantity) *resource.Quantity {
+	return &q
+}
+
 // MapStringStringPointer generates a map[string]*string
 func MapStringStringPointer(in map[string]string) (out map[string]*string) {
 	out = make(map[string]*string, 0)
@@ -170,6 +182,12 @@ func ParsePropertiesFormat(properties string) map[string]string {
 	return config
 }
 
+func AreStringSlicesIdentical(a, b []string) bool {
+	sort.Strings(a)
+	sort.Strings(b)
+	return reflect.DeepEqual(a, b)
+}
+
 // GetBrokerConfig compose the brokerConfig for a given broker
 func GetBrokerConfig(broker v1beta1.Broker, clusterSpec v1beta1.KafkaClusterSpec) (*v1beta1.BrokerConfig, error) {
 
@@ -199,9 +217,7 @@ func GetBrokerImage(brokerConfig *v1beta1.BrokerConfig, clusterImage string) str
 func GetRandomString(length int) (string, error) {
 	rand.Seed(time.Now().UnixNano())
 
-	chars := []rune("ABCDEFGHIJKLMNOPQRSTUVWXYZ" +
-		"abcdefghijklmnopqrstuvwxyz" +
-		"0123456789")
+	chars := []rune(symbolSet)
 
 	var b strings.Builder
 	for i := 0; i < length; i++ {


Test patch:
diff --git a/controllers/suite_test.go b/controllers/suite_test.go
index e5dff132..4b59cfcc 100644
--- a/controllers/suite_test.go
+++ b/controllers/suite_test.go
@@ -36,14 +36,15 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
-	banzaicloudv1alpha1 "github.com/banzaicloud/kafka-operator/api/v1alpha1"
-	banzaicloudv1beta1 "github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	banzaicloudv1alpha1 "github.com/banzaicloud/kafka-operator/api/v1alpha1"
+	banzaicloudv1beta1 "github.com/banzaicloud/kafka-operator/api/v1beta1"
 	// +kubebuilder:scaffold:imports
 )
 
diff --git a/internal/alertmanager/currentalert/process_test.go b/internal/alertmanager/currentalert/process_test.go
index 2a6215bd..44ceee0b 100644
--- a/internal/alertmanager/currentalert/process_test.go
+++ b/internal/alertmanager/currentalert/process_test.go
@@ -249,7 +249,8 @@ func setupEnvironment(t *testing.T, testClient client.Client) {
 
 	err = testClient.Create(context.Background(), &v1beta1.KafkaCluster{
 		ObjectMeta: v1.ObjectMeta{
-			Name: "kafka",
+			Name:      "kafka",
+			Namespace: "kafka",
 		},
 		Spec: v1beta1.KafkaClusterSpec{
 			Brokers: []v1beta1.Broker{
diff --git a/pkg/resources/cruisecontrol/configmap_test.go b/pkg/resources/cruisecontrol/configmap_test.go
new file mode 100644
index 00000000..84689d17
--- /dev/null
+++ b/pkg/resources/cruisecontrol/configmap_test.go
@@ -0,0 +1,371 @@
+// Copyright © 2020 Banzai Cloud
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cruisecontrol
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+func TestGenerateCapacityConfig(t *testing.T) {
+	quantity, _ := resource.ParseQuantity("10Gi")
+
+	testCases := []struct {
+		testName              string
+		kafkaCluster          v1beta1.KafkaCluster
+		expectedConfiguration string
+	}{
+		{
+			testName: "if config is set manually then use that one",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					Brokers: []v1beta1.Broker{
+						{
+							Id: 0,
+						},
+						{
+							Id: 1,
+						},
+						{
+							Id: 2,
+						},
+						{
+							Id: 4,
+						},
+					},
+					CruiseControlConfig: v1beta1.CruiseControlConfig{
+						CapacityConfig: `
+                  {
+				  "brokerCapacities":[
+					{
+					  "brokerId": "-1",
+					  "capacity": {
+						"DISK": {"/tmp/kafka-logs-1": "100000", "/tmp/kafka-logs-2": "100000", "/tmp/kafka-logs-3": "50000",
+						  "/tmp/kafka-logs-4": "50000", "/tmp/kafka-logs-5": "150000", "/tmp/kafka-logs-6": "50000"},
+						"CPU": "100",
+						"NW_IN": "10000",
+						"NW_OUT": "10000"
+					  },
+					  "doc": "The default capacity for a broker with multiple logDirs each on a separate heterogeneous disk."
+					},
+					{
+					  "brokerId": "0",
+					  "capacity": {
+						"DISK": {"/tmp/kafka-logs": "500000"},
+						"CPU": "100",
+						"NW_IN": "50000",
+						"NW_OUT": "50000"
+					  },
+					  "doc": "This overrides the capacity for broker 0. This broker is not a JBOD broker."
+					},
+					{
+					  "brokerId": "1",
+					  "capacity": {
+						"DISK": {"/tmp/kafka-logs-1": "250000", "/tmp/kafka-logs-2": "250000"},
+						"CPU": "100",
+						"NW_IN": "50000",
+						"NW_OUT": "50000"
+					  },
+					  "doc": "This overrides the capacity for broker 1. This broker is a JBOD broker."
+					}
+				  ]
+				}`,
+					},
+				},
+			},
+			expectedConfiguration: `
+                  {
+				  "brokerCapacities":[
+					{
+					  "brokerId": "-1",
+					  "capacity": {
+						"DISK": {"/tmp/kafka-logs-1": "100000", "/tmp/kafka-logs-2": "100000", "/tmp/kafka-logs-3": "50000",
+						  "/tmp/kafka-logs-4": "50000", "/tmp/kafka-logs-5": "150000", "/tmp/kafka-logs-6": "50000"},
+						"CPU": "100",
+						"NW_IN": "10000",
+						"NW_OUT": "10000"
+					  },
+					  "doc": "The default capacity for a broker with multiple logDirs each on a separate heterogeneous disk."
+					},
+					{
+					  "brokerId": "0",
+					  "capacity": {
+						"DISK": {"/tmp/kafka-logs": "500000"},
+						"CPU": "100",
+						"NW_IN": "50000",
+						"NW_OUT": "50000"
+					  },
+					  "doc": "This overrides the capacity for broker 0. This broker is not a JBOD broker."
+					},
+					{
+					  "brokerId": "1",
+					  "capacity": {
+						"DISK": {"/tmp/kafka-logs-1": "250000", "/tmp/kafka-logs-2": "250000"},
+						"CPU": "100",
+						"NW_IN": "50000",
+						"NW_OUT": "50000"
+					  },
+					  "doc": "This overrides the capacity for broker 1. This broker is a JBOD broker."
+					}
+				  ]
+				}`,
+		},
+		{
+			testName: "generate correct capacity config when there is a broker config group",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
+						"default": {
+							StorageConfigs: []v1beta1.StorageConfig{
+								{
+									MountPath: "/path-from-default",
+									PvcSpec: &v1.PersistentVolumeClaimSpec{
+										Resources: v1.ResourceRequirements{
+											Requests: v1.ResourceList{
+												"storage": quantity,
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+					Brokers: []v1beta1.Broker{
+						{
+							Id:                0,
+							BrokerConfigGroup: "default",
+						},
+						{
+							Id:                1,
+							BrokerConfigGroup: "default",
+						},
+						{
+							Id:                2,
+							BrokerConfigGroup: "default",
+						},
+						{
+							Id:                3,
+							BrokerConfigGroup: "default",
+							BrokerConfig: &v1beta1.BrokerConfig{
+								StorageConfigs: []v1beta1.StorageConfig{
+									{
+										MountPath: "/path1",
+										PvcSpec: &v1.PersistentVolumeClaimSpec{
+											Resources: v1.ResourceRequirements{
+												Requests: v1.ResourceList{
+													"storage": quantity,
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedConfiguration: `
+				  {
+					"brokerCapacities": [
+                      {
+					  "brokerId": "0",
+					  "capacity": {
+					   "DISK": {
+						"/path-from-default/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 },
+                     {
+					  "brokerId": "1",
+					  "capacity": {
+					   "DISK": {
+						"/path-from-default/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 },
+                      {
+					  "brokerId": "2",
+					  "capacity": {
+					   "DISK": {
+						"/path-from-default/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 },
+					 {
+					  "brokerId": "3",
+					  "capacity": {
+					   "DISK": {
+						"/path1/kafka": "10737418240",
+						"/path-from-default/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 }
+					]
+                  }`,
+		},
+		{
+			testName: "generate correct capacity config when there is no broker config group on last broker",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
+						"default": {
+							StorageConfigs: []v1beta1.StorageConfig{
+								{
+									MountPath: "/path-from-default",
+									PvcSpec: &v1.PersistentVolumeClaimSpec{
+										Resources: v1.ResourceRequirements{
+											Requests: v1.ResourceList{
+												"storage": quantity,
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+					Brokers: []v1beta1.Broker{
+						{
+							Id:                0,
+							BrokerConfigGroup: "default",
+						},
+						{
+							Id:                1,
+							BrokerConfigGroup: "default",
+						},
+						{
+							Id:                2,
+							BrokerConfigGroup: "default",
+						},
+						{
+							Id: 3,
+							BrokerConfig: &v1beta1.BrokerConfig{
+								StorageConfigs: []v1beta1.StorageConfig{
+									{
+										MountPath: "/path1",
+										PvcSpec: &v1.PersistentVolumeClaimSpec{
+											Resources: v1.ResourceRequirements{
+												Requests: v1.ResourceList{
+													"storage": quantity,
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedConfiguration: `{
+					"brokerCapacities": [
+                      {
+					  "brokerId": "0",
+					  "capacity": {
+					   "DISK": {
+						"/path-from-default/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 },
+                     {
+					  "brokerId": "1",
+					  "capacity": {
+					   "DISK": {
+						"/path-from-default/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 },
+                      {
+					  "brokerId": "2",
+					  "capacity": {
+					   "DISK": {
+						"/path-from-default/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 },
+					 {
+					  "brokerId": "3",
+					  "capacity": {
+					   "DISK": {
+						"/path1/kafka": "10737418240"
+					   },
+					   "CPU": "100",
+					   "NW_IN": "10000",
+					   "NW_OUT": "10000"
+					  },
+					  "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+					 }
+					]
+                  }`,
+		},
+	}
+
+	t.Parallel()
+
+	for _, test := range testCases {
+		test := test
+
+		t.Run(test.testName, func(t *testing.T) {
+			var actual CruiseControlCapacityConfig
+			err := json.Unmarshal([]byte(GenerateCapacityConfig(&test.kafkaCluster, log.NullLogger{})), &actual)
+			if err != nil {
+				t.Error(err, "could not actual unmarshal json")
+			}
+
+			var expected CruiseControlCapacityConfig
+			err = json.Unmarshal([]byte(test.expectedConfiguration), &expected)
+			if err != nil {
+				t.Error(err, "could not unmarshal expected json")
+			}
+
+			if !reflect.DeepEqual(actual, expected) {
+				t.Error("Expected:", expected, ", got:", actual)
+			}
+		})
+	}
+}
diff --git a/pkg/resources/kafka/kafka_test.go b/pkg/resources/kafka/kafka_test.go
index 8cf491ff..85236584 100644
--- a/pkg/resources/kafka/kafka_test.go
+++ b/pkg/resources/kafka/kafka_test.go
@@ -28,7 +28,60 @@ func TestGetBrokersWithPendingOrRunningCCTask(t *testing.T) {
 		expectedIDs  []int32
 	}{
 		{
-			testName: "no pending or running CC tasks",
+			testName: "pending and running CC rebalance tasks",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					Brokers: []v1beta1.Broker{
+						{
+							Id: 0,
+						},
+						{
+							Id: 1,
+						},
+						{
+							Id: 2,
+						},
+						{
+							Id: 4,
+						},
+					},
+				},
+				Status: v1beta1.KafkaClusterStatus{
+					BrokersState: map[string]v1beta1.BrokerState{
+						"0": {
+							GracefulActionState: v1beta1.GracefulActionState{
+								CruiseControlTaskId: "cc-task-id-1",
+								CruiseControlState:  v1beta1.GracefulDownscaleRunning,
+							},
+						},
+						"1": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleSucceeded,
+								VolumeStates: map[string]v1beta1.VolumeState{
+									"/path1": {
+										CruiseControlTaskId:      "1",
+										CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceRunning,
+									}}},
+						},
+						"2": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleSucceeded,
+								VolumeStates: map[string]v1beta1.VolumeState{
+									"/path1": {
+										CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceRequired,
+									}}},
+						},
+						"3": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRequired},
+						},
+						"4": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRunning},
+						},
+					},
+				},
+			},
+			expectedIDs: []int32{0, 1, 2},
+		},
+		{
+			testName: "no pending or running CC upscale tasks",
 			kafkaCluster: v1beta1.KafkaCluster{
 				Spec: v1beta1.KafkaClusterSpec{
 					Brokers: []v1beta1.Broker{
@@ -49,20 +102,20 @@ func TestGetBrokersWithPendingOrRunningCCTask(t *testing.T) {
 							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleSucceeded},
 						},
 						"1": {
-							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateFailed},
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleSucceeded},
 						},
 						"2": {
 							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleSucceeded},
 						},
 						"3": {
-							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRequired},
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleRequired},
 						},
 					},
 				},
 			},
 		},
 		{
-			testName: "pending and running CC tasks",
+			testName: "pending and running CC upscale tasks",
 			kafkaCluster: v1beta1.KafkaCluster{
 				Spec: v1beta1.KafkaClusterSpec{
 					Brokers: []v1beta1.Broker{
@@ -85,26 +138,142 @@ func TestGetBrokersWithPendingOrRunningCCTask(t *testing.T) {
 						"0": {
 							GracefulActionState: v1beta1.GracefulActionState{
 								CruiseControlTaskId: "cc-task-id-1",
-								CruiseControlState:  v1beta1.GracefulUpdateRunning,
+								CruiseControlState:  v1beta1.GracefulUpscaleRunning,
 							},
 						},
 						"1": {
-							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateFailed},
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleRunning},
 						},
 						"2": {
-							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRequired},
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleRequired},
 						},
 						"3": {
-							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRequired},
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleRequired},
 						},
 						"4": {
-							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRunning},
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleRunning},
 						},
 					},
 				},
 			},
 			expectedIDs: []int32{0, 2},
 		},
+		{
+			testName: "no pending or running CC downscale tasks",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					Brokers: []v1beta1.Broker{
+						{
+							Id: 0,
+						},
+						{
+							Id: 1,
+						},
+						{
+							Id: 2,
+						},
+					},
+				},
+				Status: v1beta1.KafkaClusterStatus{
+					BrokersState: map[string]v1beta1.BrokerState{
+						"0": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleSucceeded},
+						},
+						"1": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleSucceeded},
+						},
+						"2": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleSucceeded},
+						},
+						"3": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRequired},
+						},
+					},
+				},
+			},
+		},
+		{
+			testName: "pending and running CC downscale tasks",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					Brokers: []v1beta1.Broker{
+						{
+							Id: 0,
+						},
+						{
+							Id: 1,
+						},
+						{
+							Id: 2,
+						},
+						{
+							Id: 4,
+						},
+					},
+				},
+				Status: v1beta1.KafkaClusterStatus{
+					BrokersState: map[string]v1beta1.BrokerState{
+						"0": {
+							GracefulActionState: v1beta1.GracefulActionState{
+								CruiseControlTaskId: "cc-task-id-1",
+								CruiseControlState:  v1beta1.GracefulDownscaleRunning,
+							},
+						},
+						"1": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRunning},
+						},
+						"2": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRequired},
+						},
+						"3": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRequired},
+						},
+						"4": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRunning},
+						},
+					},
+				},
+			},
+			expectedIDs: []int32{0, 2},
+		},
+		{
+			testName: "no pending or running CC rebalance tasks",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					Brokers: []v1beta1.Broker{
+						{
+							Id: 0,
+						},
+						{
+							Id: 1,
+						},
+						{
+							Id: 2,
+						},
+					},
+				},
+				Status: v1beta1.KafkaClusterStatus{
+					BrokersState: map[string]v1beta1.BrokerState{
+						"0": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleSucceeded,
+								VolumeStates: map[string]v1beta1.VolumeState{
+									"/path1": {
+										CruiseControlVolumeState: v1beta1.GracefulDiskRebalanceSucceeded,
+									}}},
+						},
+						"1": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpscaleSucceeded},
+						},
+						"2": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleSucceeded},
+						},
+						"3": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRequired},
+						},
+					},
+				},
+			},
+		},
 	}
 
 	t.Parallel()