@Test public void testNewWithMissingHardGoalsAndRefresh(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceBadGoalsError(ccServer); KafkaRebalanceSpec kafkaRebalanceSpec = new KafkaRebalanceSpecBuilder() .withGoals("DiskCapacityGoal", "CpuCapacityGoal") .build(); KafkaRebalance kr = createKafkaRebalance(CLUSTER_NAMESPACE, CLUSTER_NAME, RESOURCE_NAME, kafkaRebalanceSpec); Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).create(kr); when(mockKafkaOps.getAsync(CLUSTER_NAMESPACE, CLUSTER_NAME)).thenReturn(Future.succeededFuture(kafka)); mockRebalanceOperator(mockRebalanceOps, CLUSTER_NAMESPACE, RESOURCE_NAME, kubernetesClient); Checkpoint checkpoint = context.checkpoint(); kcrao.reconcileRebalance(new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), kr) .onComplete(context.succeeding(v -> context.verify(() -> { KafkaRebalance kr1 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); assertState(kr1, KafkaRebalanceState.NotReady, CruiseControlRestException.class, "Error processing POST request '/rebalance' due to: " + "'java.lang.IllegalArgumentException: Missing hard goals [NetworkInboundCapacityGoal, DiskCapacityGoal, RackAwareGoal, NetworkOutboundCapacityGoal, CpuCapacityGoal, ReplicaCapacityGoal] " + "in the provided goals: [RackAwareGoal, ReplicaCapacityGoal]. " + "Add skip_hard_goal_check=true parameter to ignore this sanity check.'."); }))) .compose(v -> { ccServer.reset(); try { MockCruiseControl.setupCCRebalanceResponse(ccServer, 0); } catch (IOException | URISyntaxException e) { context.failNow(e); } KafkaRebalance kr2 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); KafkaRebalance patchedKr = new KafkaRebalanceBuilder(kr2) .editSpec() .withSkipHardGoalCheck(true) .endSpec() .build(); Crds.kafkaRebalanceOperation(kubernetesClient) .inNamespace(CLUSTER_NAMESPACE) .withName(RESOURCE_NAME) .patch(patchedKr); patchedKr = annotate(kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceAnnotation.refresh); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), patchedKr); }) .onComplete(context.succeeding(v -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.ProposalReady); checkpoint.flag(); })); }
Future<Void> reconcileRebalance(Reconciliation reconciliation, KafkaRebalance kafkaRebalance) { if (kafkaRebalance == null) { log.info("{}: Rebalance resource deleted", reconciliation); return Future.succeededFuture(); } String clusterName = kafkaRebalance.getMetadata().getLabels() == null ? null : kafkaRebalance.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL); String clusterNamespace = kafkaRebalance.getMetadata().getNamespace(); if (clusterName == null) { log.warn("{}: Resource lacks label '{}': No cluster related to a possible rebalance.", reconciliation, Labels.STRIMZI_CLUSTER_LABEL); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Resource lacks label '" + Labels.STRIMZI_CLUSTER_LABEL + "': No cluster related to a possible rebalance.")).mapEmpty(); } return kafkaOperator.getAsync(clusterNamespace, clusterName) .compose(kafka -> { if (kafka == null) { log.warn("{}: Kafka resource '{}' identified by label '{}' does not exist in namespace {}.", reconciliation, clusterName, Labels.STRIMZI_CLUSTER_LABEL, clusterNamespace); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new NoSuchResourceException("Kafka resource '" + clusterName + "' identified by label '" + Labels.STRIMZI_CLUSTER_LABEL + "' does not exist in namespace " + clusterNamespace + ".")).mapEmpty(); } else if (kafka.getSpec().getCruiseControl() == null) { log.warn("{}: Kafka resource lacks 'cruiseControl' declaration : No deployed Cruise Control for doing a rebalance.", reconciliation); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Kafka resource lacks 'cruiseControl' declaration " + ": No deployed Cruise Control for doing a rebalance.")).mapEmpty(); } CruiseControlApi apiClient = cruiseControlClientProvider(); return kafkaRebalanceOperator.getAsync(kafkaRebalance.getMetadata().getNamespace(), kafkaRebalance.getMetadata().getName()) .compose(currentKafkaRebalance -> { KafkaRebalanceStatus kafkaRebalanceStatus = currentKafkaRebalance.getStatus(); KafkaRebalanceState currentState; if (kafkaRebalanceStatus == null) { currentState = KafkaRebalanceState.New; } else { String rebalanceStateType = rebalanceStateConditionType(kafkaRebalanceStatus); if (rebalanceStateType == null) { throw new RuntimeException("Unable to find KafkaRebalance State in current KafkaRebalance status"); } currentState = KafkaRebalanceState.valueOf(rebalanceStateType); } KafkaRebalanceAnnotation rebalanceAnnotation = rebalanceAnnotation(currentKafkaRebalance); return reconcile(reconciliation, cruiseControlHost(clusterName, clusterNamespace), apiClient, currentKafkaRebalance, currentState, rebalanceAnnotation).mapEmpty(); }, exception -> Future.failedFuture(exception).mapEmpty()); }, exception -> updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), exception).mapEmpty()); }
KafkaRebalanceAssemblyOperator extends AbstractOperator<KafkaRebalance, KafkaRebalanceSpec, KafkaRebalanceStatus, AbstractWatchableStatusedResourceOperator<KubernetesClient, KafkaRebalance, KafkaRebalanceList, DoneableKafkaRebalance, Resource<KafkaRebalance, DoneableKafkaRebalance>>> { Future<Void> reconcileRebalance(Reconciliation reconciliation, KafkaRebalance kafkaRebalance) { if (kafkaRebalance == null) { log.info("{}: Rebalance resource deleted", reconciliation); return Future.succeededFuture(); } String clusterName = kafkaRebalance.getMetadata().getLabels() == null ? null : kafkaRebalance.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL); String clusterNamespace = kafkaRebalance.getMetadata().getNamespace(); if (clusterName == null) { log.warn("{}: Resource lacks label '{}': No cluster related to a possible rebalance.", reconciliation, Labels.STRIMZI_CLUSTER_LABEL); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Resource lacks label '" + Labels.STRIMZI_CLUSTER_LABEL + "': No cluster related to a possible rebalance.")).mapEmpty(); } return kafkaOperator.getAsync(clusterNamespace, clusterName) .compose(kafka -> { if (kafka == null) { log.warn("{}: Kafka resource '{}' identified by label '{}' does not exist in namespace {}.", reconciliation, clusterName, Labels.STRIMZI_CLUSTER_LABEL, clusterNamespace); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new NoSuchResourceException("Kafka resource '" + clusterName + "' identified by label '" + Labels.STRIMZI_CLUSTER_LABEL + "' does not exist in namespace " + clusterNamespace + ".")).mapEmpty(); } else if (kafka.getSpec().getCruiseControl() == null) { log.warn("{}: Kafka resource lacks 'cruiseControl' declaration : No deployed Cruise Control for doing a rebalance.", reconciliation); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Kafka resource lacks 'cruiseControl' declaration " + ": No deployed Cruise Control for doing a rebalance.")).mapEmpty(); } CruiseControlApi apiClient = cruiseControlClientProvider(); return kafkaRebalanceOperator.getAsync(kafkaRebalance.getMetadata().getNamespace(), kafkaRebalance.getMetadata().getName()) .compose(currentKafkaRebalance -> { KafkaRebalanceStatus kafkaRebalanceStatus = currentKafkaRebalance.getStatus(); KafkaRebalanceState currentState; if (kafkaRebalanceStatus == null) { currentState = KafkaRebalanceState.New; } else { String rebalanceStateType = rebalanceStateConditionType(kafkaRebalanceStatus); if (rebalanceStateType == null) { throw new RuntimeException("Unable to find KafkaRebalance State in current KafkaRebalance status"); } currentState = KafkaRebalanceState.valueOf(rebalanceStateType); } KafkaRebalanceAnnotation rebalanceAnnotation = rebalanceAnnotation(currentKafkaRebalance); return reconcile(reconciliation, cruiseControlHost(clusterName, clusterNamespace), apiClient, currentKafkaRebalance, currentState, rebalanceAnnotation).mapEmpty(); }, exception -> Future.failedFuture(exception).mapEmpty()); }, exception -> updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), exception).mapEmpty()); } }
KafkaRebalanceAssemblyOperator extends AbstractOperator<KafkaRebalance, KafkaRebalanceSpec, KafkaRebalanceStatus, AbstractWatchableStatusedResourceOperator<KubernetesClient, KafkaRebalance, KafkaRebalanceList, DoneableKafkaRebalance, Resource<KafkaRebalance, DoneableKafkaRebalance>>> { Future<Void> reconcileRebalance(Reconciliation reconciliation, KafkaRebalance kafkaRebalance) { if (kafkaRebalance == null) { log.info("{}: Rebalance resource deleted", reconciliation); return Future.succeededFuture(); } String clusterName = kafkaRebalance.getMetadata().getLabels() == null ? null : kafkaRebalance.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL); String clusterNamespace = kafkaRebalance.getMetadata().getNamespace(); if (clusterName == null) { log.warn("{}: Resource lacks label '{}': No cluster related to a possible rebalance.", reconciliation, Labels.STRIMZI_CLUSTER_LABEL); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Resource lacks label '" + Labels.STRIMZI_CLUSTER_LABEL + "': No cluster related to a possible rebalance.")).mapEmpty(); } return kafkaOperator.getAsync(clusterNamespace, clusterName) .compose(kafka -> { if (kafka == null) { log.warn("{}: Kafka resource '{}' identified by label '{}' does not exist in namespace {}.", reconciliation, clusterName, Labels.STRIMZI_CLUSTER_LABEL, clusterNamespace); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new NoSuchResourceException("Kafka resource '" + clusterName + "' identified by label '" + Labels.STRIMZI_CLUSTER_LABEL + "' does not exist in namespace " + clusterNamespace + ".")).mapEmpty(); } else if (kafka.getSpec().getCruiseControl() == null) { log.warn("{}: Kafka resource lacks 'cruiseControl' declaration : No deployed Cruise Control for doing a rebalance.", reconciliation); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Kafka resource lacks 'cruiseControl' declaration " + ": No deployed Cruise Control for doing a rebalance.")).mapEmpty(); } CruiseControlApi apiClient = cruiseControlClientProvider(); return kafkaRebalanceOperator.getAsync(kafkaRebalance.getMetadata().getNamespace(), kafkaRebalance.getMetadata().getName()) .compose(currentKafkaRebalance -> { KafkaRebalanceStatus kafkaRebalanceStatus = currentKafkaRebalance.getStatus(); KafkaRebalanceState currentState; if (kafkaRebalanceStatus == null) { currentState = KafkaRebalanceState.New; } else { String rebalanceStateType = rebalanceStateConditionType(kafkaRebalanceStatus); if (rebalanceStateType == null) { throw new RuntimeException("Unable to find KafkaRebalance State in current KafkaRebalance status"); } currentState = KafkaRebalanceState.valueOf(rebalanceStateType); } KafkaRebalanceAnnotation rebalanceAnnotation = rebalanceAnnotation(currentKafkaRebalance); return reconcile(reconciliation, cruiseControlHost(clusterName, clusterNamespace), apiClient, currentKafkaRebalance, currentState, rebalanceAnnotation).mapEmpty(); }, exception -> Future.failedFuture(exception).mapEmpty()); }, exception -> updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), exception).mapEmpty()); } KafkaRebalanceAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                          ResourceOperatorSupplier supplier); }
KafkaRebalanceAssemblyOperator extends AbstractOperator<KafkaRebalance, KafkaRebalanceSpec, KafkaRebalanceStatus, AbstractWatchableStatusedResourceOperator<KubernetesClient, KafkaRebalance, KafkaRebalanceList, DoneableKafkaRebalance, Resource<KafkaRebalance, DoneableKafkaRebalance>>> { Future<Void> reconcileRebalance(Reconciliation reconciliation, KafkaRebalance kafkaRebalance) { if (kafkaRebalance == null) { log.info("{}: Rebalance resource deleted", reconciliation); return Future.succeededFuture(); } String clusterName = kafkaRebalance.getMetadata().getLabels() == null ? null : kafkaRebalance.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL); String clusterNamespace = kafkaRebalance.getMetadata().getNamespace(); if (clusterName == null) { log.warn("{}: Resource lacks label '{}': No cluster related to a possible rebalance.", reconciliation, Labels.STRIMZI_CLUSTER_LABEL); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Resource lacks label '" + Labels.STRIMZI_CLUSTER_LABEL + "': No cluster related to a possible rebalance.")).mapEmpty(); } return kafkaOperator.getAsync(clusterNamespace, clusterName) .compose(kafka -> { if (kafka == null) { log.warn("{}: Kafka resource '{}' identified by label '{}' does not exist in namespace {}.", reconciliation, clusterName, Labels.STRIMZI_CLUSTER_LABEL, clusterNamespace); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new NoSuchResourceException("Kafka resource '" + clusterName + "' identified by label '" + Labels.STRIMZI_CLUSTER_LABEL + "' does not exist in namespace " + clusterNamespace + ".")).mapEmpty(); } else if (kafka.getSpec().getCruiseControl() == null) { log.warn("{}: Kafka resource lacks 'cruiseControl' declaration : No deployed Cruise Control for doing a rebalance.", reconciliation); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Kafka resource lacks 'cruiseControl' declaration " + ": No deployed Cruise Control for doing a rebalance.")).mapEmpty(); } CruiseControlApi apiClient = cruiseControlClientProvider(); return kafkaRebalanceOperator.getAsync(kafkaRebalance.getMetadata().getNamespace(), kafkaRebalance.getMetadata().getName()) .compose(currentKafkaRebalance -> { KafkaRebalanceStatus kafkaRebalanceStatus = currentKafkaRebalance.getStatus(); KafkaRebalanceState currentState; if (kafkaRebalanceStatus == null) { currentState = KafkaRebalanceState.New; } else { String rebalanceStateType = rebalanceStateConditionType(kafkaRebalanceStatus); if (rebalanceStateType == null) { throw new RuntimeException("Unable to find KafkaRebalance State in current KafkaRebalance status"); } currentState = KafkaRebalanceState.valueOf(rebalanceStateType); } KafkaRebalanceAnnotation rebalanceAnnotation = rebalanceAnnotation(currentKafkaRebalance); return reconcile(reconciliation, cruiseControlHost(clusterName, clusterNamespace), apiClient, currentKafkaRebalance, currentState, rebalanceAnnotation).mapEmpty(); }, exception -> Future.failedFuture(exception).mapEmpty()); }, exception -> updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), exception).mapEmpty()); } KafkaRebalanceAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                          ResourceOperatorSupplier supplier); Future<Void> createRebalanceWatch(String watchNamespaceOrWildcard); }
KafkaRebalanceAssemblyOperator extends AbstractOperator<KafkaRebalance, KafkaRebalanceSpec, KafkaRebalanceStatus, AbstractWatchableStatusedResourceOperator<KubernetesClient, KafkaRebalance, KafkaRebalanceList, DoneableKafkaRebalance, Resource<KafkaRebalance, DoneableKafkaRebalance>>> { Future<Void> reconcileRebalance(Reconciliation reconciliation, KafkaRebalance kafkaRebalance) { if (kafkaRebalance == null) { log.info("{}: Rebalance resource deleted", reconciliation); return Future.succeededFuture(); } String clusterName = kafkaRebalance.getMetadata().getLabels() == null ? null : kafkaRebalance.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL); String clusterNamespace = kafkaRebalance.getMetadata().getNamespace(); if (clusterName == null) { log.warn("{}: Resource lacks label '{}': No cluster related to a possible rebalance.", reconciliation, Labels.STRIMZI_CLUSTER_LABEL); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Resource lacks label '" + Labels.STRIMZI_CLUSTER_LABEL + "': No cluster related to a possible rebalance.")).mapEmpty(); } return kafkaOperator.getAsync(clusterNamespace, clusterName) .compose(kafka -> { if (kafka == null) { log.warn("{}: Kafka resource '{}' identified by label '{}' does not exist in namespace {}.", reconciliation, clusterName, Labels.STRIMZI_CLUSTER_LABEL, clusterNamespace); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new NoSuchResourceException("Kafka resource '" + clusterName + "' identified by label '" + Labels.STRIMZI_CLUSTER_LABEL + "' does not exist in namespace " + clusterNamespace + ".")).mapEmpty(); } else if (kafka.getSpec().getCruiseControl() == null) { log.warn("{}: Kafka resource lacks 'cruiseControl' declaration : No deployed Cruise Control for doing a rebalance.", reconciliation); return updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException("Kafka resource lacks 'cruiseControl' declaration " + ": No deployed Cruise Control for doing a rebalance.")).mapEmpty(); } CruiseControlApi apiClient = cruiseControlClientProvider(); return kafkaRebalanceOperator.getAsync(kafkaRebalance.getMetadata().getNamespace(), kafkaRebalance.getMetadata().getName()) .compose(currentKafkaRebalance -> { KafkaRebalanceStatus kafkaRebalanceStatus = currentKafkaRebalance.getStatus(); KafkaRebalanceState currentState; if (kafkaRebalanceStatus == null) { currentState = KafkaRebalanceState.New; } else { String rebalanceStateType = rebalanceStateConditionType(kafkaRebalanceStatus); if (rebalanceStateType == null) { throw new RuntimeException("Unable to find KafkaRebalance State in current KafkaRebalance status"); } currentState = KafkaRebalanceState.valueOf(rebalanceStateType); } KafkaRebalanceAnnotation rebalanceAnnotation = rebalanceAnnotation(currentKafkaRebalance); return reconcile(reconciliation, cruiseControlHost(clusterName, clusterNamespace), apiClient, currentKafkaRebalance, currentState, rebalanceAnnotation).mapEmpty(); }, exception -> Future.failedFuture(exception).mapEmpty()); }, exception -> updateStatus(kafkaRebalance, new KafkaRebalanceStatus(), exception).mapEmpty()); } KafkaRebalanceAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                          ResourceOperatorSupplier supplier); Future<Void> createRebalanceWatch(String watchNamespaceOrWildcard);  }