Fix lint issues. refactor handleDiskRemoval into function to reduce complexity

Author: hvan

pkg/resources/kafka/kafka.go

@@ -1182,56 +1182,9 @@ func (r *Reconciler) reconcileKafkaPvc(ctx context.Context, log logr.Logger, bro
 
 		// Handle disk removal
 		if len(pvcList.Items) > len(desiredPvcs) {
-			for _, pvc := range pvcList.Items {
-				foundInDesired := false
-				existingMountPath := pvc.Annotations["mountPath"]
-
-				for _, desiredPvc := range desiredPvcs {
-					desiredMountPath := desiredPvc.Annotations["mountPath"]
-
-					if existingMountPath == desiredMountPath {
-						foundInDesired = true
-						break
-					}
-				}
-
-				if foundInDesired {
-					continue
-				}
-
-				mountPathToRemove := existingMountPath
-				if brokerState, ok := r.KafkaCluster.Status.BrokersState[brokerId]; ok {
-					volumeStateStatus, found := brokerState.GracefulActionState.VolumeStates[mountPathToRemove]
-					if !found {
-						// If the state is not found, it means that the disk removal was done according to the disk removal succeeded branch
-						log.Info("Disk removal was completed, waiting for Rolling Upgrade to remove PVC", "brokerId", brokerId, "mountPath", mountPathToRemove)
-						continue
-					}
-
-					// Check the volume state
-					ccVolumeState := volumeStateStatus.CruiseControlVolumeState
-					switch {
-					case ccVolumeState.IsDiskRemovalSucceeded():
-						if err := r.Client.Delete(ctx, &pvc); err != nil {
-							return errorfactory.New(errorfactory.APIFailure{}, err, "deleting resource failed", "kind", desiredType)
-						}
-						log.Info("resource deleted")
-						err = k8sutil.DeleteVolumeStatus(r.Client, brokerId, mountPathToRemove, r.KafkaCluster, log)
-						if err != nil {
-							return errors.WrapIfWithDetails(err, "could not delete volume status for broker volume", "brokerId", brokerId, "mountPath", mountPathToRemove)
-						}
-					case ccVolumeState.IsDiskRemoval():
-						log.Info("Graceful disk removal is in progress", "brokerId", brokerId, "mountPath", mountPathToRemove)
-						waitForDiskRemovalToFinish = true
-					case ccVolumeState.IsDiskRebalance():
-						log.Info("Graceful disk rebalance is in progress, waiting to mark disk for removal", "brokerId", brokerId, "mountPath", mountPathToRemove)
-						waitForDiskRemovalToFinish = true
-					default:
-						brokerVolumesState[mountPathToRemove] = v1beta1.VolumeState{CruiseControlVolumeState: v1beta1.GracefulDiskRemovalRequired}
-						log.Info("Marked the volume for removal", "brokerId", brokerId, "mountPath", mountPathToRemove)
-						waitForDiskRemovalToFinish = true
-					}
-				}
+			waitForDiskRemovalToFinish, err = handleDiskRemoval(ctx, pvcList, desiredPvcs, r, brokerId, log, desiredType, brokerVolumesState)
+			if err != nil {
+				return err
 			}
 		}
 
@@ -1330,6 +1283,63 @@ func (r *Reconciler) reconcileKafkaPvc(ctx context.Context, log logr.Logger, bro
 	return nil
 }
 
+func handleDiskRemoval(ctx context.Context, pvcList *corev1.PersistentVolumeClaimList, desiredPvcs []*corev1.PersistentVolumeClaim,
+	r *Reconciler, brokerId string, log logr.Logger, desiredType reflect.Type, brokerVolumesState map[string]v1beta1.VolumeState) (bool, error) {
+	waitForDiskRemovalToFinish := false
+	for _, pvc := range pvcList.Items {
+		foundInDesired := false
+		existingMountPath := pvc.Annotations["mountPath"]
+
+		for _, desiredPvc := range desiredPvcs {
+			desiredMountPath := desiredPvc.Annotations["mountPath"]
+
+			if existingMountPath == desiredMountPath {
+				foundInDesired = true
+				break
+			}
+		}
+
+		if foundInDesired {
+			continue
+		}
+
+		mountPathToRemove := existingMountPath
+		if brokerState, ok := r.KafkaCluster.Status.BrokersState[brokerId]; ok {
+			volumeStateStatus, found := brokerState.GracefulActionState.VolumeStates[mountPathToRemove]
+			if !found {
+				// If the state is not found, it means that the disk removal was done according to the disk removal succeeded branch
+				log.Info("Disk removal was completed, waiting for Rolling Upgrade to remove PVC", "brokerId", brokerId, "mountPath", mountPathToRemove)
+				continue
+			}
+
+			// Check the volume state
+			ccVolumeState := volumeStateStatus.CruiseControlVolumeState
+			switch {
+			case ccVolumeState.IsDiskRemovalSucceeded():
+				if err := r.Client.Delete(ctx, &pvc); err != nil {
+					return false, errorfactory.New(errorfactory.APIFailure{}, err, "deleting resource failed", "kind", desiredType)
+				}
+				log.Info("resource deleted")
+				err := k8sutil.DeleteVolumeStatus(r.Client, brokerId, mountPathToRemove, r.KafkaCluster, log)
+				if err != nil {
+					return false, errors.WrapIfWithDetails(err, "could not delete volume status for broker volume", "brokerId", brokerId, "mountPath", mountPathToRemove)
+				}
+			case ccVolumeState.IsDiskRemoval():
+				log.Info("Graceful disk removal is in progress", "brokerId", brokerId, "mountPath", mountPathToRemove)
+				waitForDiskRemovalToFinish = true
+			case ccVolumeState.IsDiskRebalance():
+				log.Info("Graceful disk rebalance is in progress, waiting to mark disk for removal", "brokerId", brokerId, "mountPath", mountPathToRemove)
+				waitForDiskRemovalToFinish = true
+			default:
+				brokerVolumesState[mountPathToRemove] = v1beta1.VolumeState{CruiseControlVolumeState: v1beta1.GracefulDiskRemovalRequired}
+				log.Info("Marked the volume for removal", "brokerId", brokerId, "mountPath", mountPathToRemove)
+				waitForDiskRemovalToFinish = true
+			}
+		}
+	}
+	return waitForDiskRemovalToFinish, nil
+}
+
 // GetBrokersWithPendingOrRunningCCTask returns list of brokers that are either waiting for CC
 // to start executing a broker task (add broker, remove broker, etc) or CC already running a task for it.
 func GetBrokersWithPendingOrRunningCCTask(kafkaCluster *v1beta1.KafkaCluster) []int32 {

pkg/resources/kafka/kafka_test.go

@@ -1233,10 +1233,9 @@ func execPvcTest(t *testing.T, testCases []PvcTestCase) {
 			},
 		},
 	}
-
 	mockCtrl := gomock.NewController(t)
-	for _, test := range testCases {
 
+	for _, test := range testCases {
 		if test.kafkaClusterSpec.Brokers != nil {
 			kafkaClusterSpec = test.kafkaClusterSpec
 		}