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);  }
@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(); })); }
@Test public void testNewToPendingProposalToProposalReadyRebalance(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 2); 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.PendingProposal); })) .compose(v -> { KafkaRebalance kr1 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); return kcrao.reconcileRebalance(new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), kr1); }) .onComplete(context.succeeding(v -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.ProposalReady); checkpoint.flag(); })); }
@Test public void testNewToPendingProposalToStoppedRebalance(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 5); 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, new Runnable() { int count = 0; @Override public void run() { if (++count == 4) { annotate(kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceAnnotation.stop); } return; } }); 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.PendingProposal); })) .compose(v -> { KafkaRebalance kr1 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); return kcrao.reconcileRebalance(new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), kr1); }) .onComplete(context.succeeding(v -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.Stopped); checkpoint.flag(); })); }
@Test public void testNewToPendingProposalToStoppedAndRefreshRebalance(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 2); 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, new Runnable() { int count = 0; @Override public void run() { if (++count == 4) { annotate(kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceAnnotation.stop); } return; } }); 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.PendingProposal))) .compose(v -> { KafkaRebalance kr1 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), kr1); }) .onComplete(context.succeeding(v -> assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.Stopped))) .compose(v -> { KafkaRebalance refreshedKr = annotate(kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceAnnotation.refresh); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), refreshedKr); }) .onComplete(context.succeeding(v -> assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.PendingProposal))) .compose(v -> { KafkaRebalance kr6 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), kr6); }) .onComplete(context.succeeding(v -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.ProposalReady); checkpoint.flag(); })); }
@Test public void testNewToProposalReadyToRebalancingToReadyRebalance(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 0); MockCruiseControl.setupCCUserTasksResponseNoGoals(ccServer, 0, 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))) .compose(v -> { KafkaRebalance approvedKr = annotate(kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceAnnotation.approve); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), approvedKr); }) .onComplete(context.succeeding(v -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.Rebalancing); })) .compose(v -> { KafkaRebalance kr4 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), kr4); }) .onComplete(context.succeeding(v -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.Ready); checkpoint.flag(); })); }
@Test public void testNewWithMissingHardGoals(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.'."); checkpoint.flag(); }))); }
@Test public void testNewToProposalReadySkipHardGoalsRebalance(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 0); KafkaRebalanceSpec kafkaRebalanceSpec = new KafkaRebalanceSpecBuilder() .withGoals("DiskCapacityGoal", "CpuCapacityGoal") .withSkipHardGoalCheck(true) .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 -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.ProposalReady); checkpoint.flag(); })); }
@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(); })); }
@Test public void testNewToProposalReadyToRebalancingToStoppedRebalance(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 0); MockCruiseControl.setupCCUserTasksResponseNoGoals(ccServer, 0, 2); MockCruiseControl.setupCCStopResponse(ccServer); 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, new Runnable() { int count = 0; @Override public void run() { if (++count == 6) { annotate(kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceAnnotation.stop); } return; } }); 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))) .compose(v -> { KafkaRebalance approvedKr = annotate(kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceAnnotation.approve); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), approvedKr); }) .onComplete(context.succeeding(v -> assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.Rebalancing))) .compose(v -> { KafkaRebalance kr5 = Crds.kafkaRebalanceOperation(kubernetesClient).inNamespace(CLUSTER_NAMESPACE).withName(RESOURCE_NAME).get(); return kcrao.reconcileRebalance( new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, CLUSTER_NAMESPACE, RESOURCE_NAME), kr5); }) .onComplete(context.succeeding(v -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.Stopped); checkpoint.flag(); })); }
@Test public void testNoKafkaCluster(VertxTestContext context) { 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(null)); 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(() -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.NotReady, NoSuchResourceException.class, "Kafka resource '" + CLUSTER_NAME + "' identified by label '" + Labels.STRIMZI_CLUSTER_LABEL + "' does not exist in namespace " + CLUSTER_NAMESPACE + "."); checkpoint.flag(); }))); }
@Test public void testCruiseControlDisabled(VertxTestContext context) { Kafka kafka = new KafkaBuilder(ResourceUtils.createKafka(CLUSTER_NAMESPACE, CLUSTER_NAME, replicas, image, healthDelay, healthTimeout)) .editSpec() .editKafka() .withVersion(version) .endKafka() .endSpec() .build(); 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 -> context.verify(() -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.NotReady, InvalidResourceException.class, "Kafka resource lacks 'cruiseControl' declaration : No deployed Cruise Control for doing a rebalance."); checkpoint.flag(); }))); }
@Test public void testNoKafkaClusterInKafkaRebalanceLabel(VertxTestContext context) { KafkaRebalance kr = createKafkaRebalance(CLUSTER_NAMESPACE, null, 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 -> context.verify(() -> { assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.NotReady, InvalidResourceException.class, "Resource lacks label '" + Labels.STRIMZI_CLUSTER_LABEL + "': No cluster related to a possible rebalance."); checkpoint.flag(); }))); }
@Test public void testCruiseControlTimingOut(VertxTestContext context) throws IOException, URISyntaxException { MockCruiseControl.setupCCRebalanceResponse(ccServer, 0, 10); 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.NotReady, TimeoutException.class, "The timeout period of 1000ms has been exceeded while executing POST"); checkpoint.flag(); })); }
@Test public void testCruiseControlNotReachable(VertxTestContext context) { ccServer.stop(); 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 -> { try { ccServer = MockCruiseControl.server(CruiseControl.REST_API_PORT); } catch (IOException e) { context.failNow(e); } assertState(context, kubernetesClient, CLUSTER_NAMESPACE, RESOURCE_NAME, KafkaRebalanceState.NotReady, ConnectException.class, "Connection refused"); checkpoint.flag(); })); }