@Test public void testNewToProposalReadyRebalance(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 0); KafkaRebalance kr = createKafkaRebalance(CLUSTER_NAMESPACE, CLUSTER_NAME, RESOURCE_NAME, new KafkaRebalanceSpecBuilder().build()); 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 -> { 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);  }