Skip to content

Commit a816fdf

Browse files
ppatiernoscholzj
ppatierno
authored and
scholzj
committed
Fix logging errors and updating KafkaRebalance status (#6830)
* Fixed reporting error when brokers missing on add/remove brokers rebalance Signed-off-by: Paolo Patierno <ppatierno@live.com> * Changed Cruise Control API related to not existing brokers on add/remove mode Fixed error condition reverted back when in NotReady Signed-off-by: Paolo Patierno <ppatierno@live.com> * Fixed updating with current condition Signed-off-by: Paolo Patierno <ppatierno@live.com>
1 parent 0c54f69 commit a816fdf

File tree

5 files changed

+80
-86
lines changed

5 files changed

+80
-86
lines changed

‎cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceAssemblyOperator.java

Lines changed: 63 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -302,14 +302,13 @@ private Future<KafkaRebalance> updateStatus(Reconciliation reconciliation,
302302
if (desiredStatus.getConditions() != null) {
303303
previous = desiredStatus.getConditions().stream().filter(condition -> condition != cond).collect(Collectors.toList());
304304
}
305-
String rebalanceType = rebalanceStateConditionType(desiredStatus);
306305

307306
// If a throwable is supplied, it is set in the status with priority
308307
if (e != null) {
309308
StatusUtils.setStatusConditionAndObservedGeneration(kafkaRebalance, desiredStatus, KafkaRebalanceState.NotReady.toString(), e);
310309
desiredStatus.setConditions(Stream.concat(desiredStatus.getConditions().stream(), previous.stream()).collect(Collectors.toList()));
311-
} else if (rebalanceType != null) {
312-
StatusUtils.setStatusConditionAndObservedGeneration(kafkaRebalance, desiredStatus, rebalanceType);
310+
} else if (cond != null) {
311+
StatusUtils.setStatusConditionAndObservedGeneration(kafkaRebalance, desiredStatus, cond);
313312
desiredStatus.setConditions(Stream.concat(desiredStatus.getConditions().stream(), previous.stream()).collect(Collectors.toList()));
314313
} else {
315314
throw new IllegalArgumentException("Status related exception and the Status condition's type cannot both be null");
@@ -416,67 +415,72 @@ private Future<Void> reconcile(Reconciliation reconciliation, String host,
416415
return updateStatus(reconciliation, kafkaRebalance, new KafkaRebalanceStatus(), new InvalidResourceException(error)).mapEmpty();
417416
}
418417

419-
AbstractRebalanceOptions.AbstractRebalanceOptionsBuilder<?, ?> rebalanceOptionsBuilder = convertRebalanceSpecToRebalanceOptions(kafkaRebalance.getSpec());
420-
421-
return computeNextStatus(reconciliation, host, apiClient, kafkaRebalance, currentState, rebalanceAnnotation, rebalanceOptionsBuilder)
422-
.compose(desiredStatusAndMap -> {
423-
// More events related to resource modification might be queued with a stale state. (potentially updated by the rebalance holding the lock)
424-
// Due to possible long rebalancing operations that take the lock for the entire period,
425-
// do a new get to retrieve the current resource state.
426-
return kafkaRebalanceOperator.getAsync(reconciliation.namespace(), reconciliation.name())
427-
.compose(currentKafkaRebalance -> {
428-
if (currentKafkaRebalance != null) {
429-
return configMapOperator.reconcile(reconciliation, kafkaRebalance.getMetadata().getNamespace(),
430-
kafkaRebalance.getMetadata().getName(), desiredStatusAndMap.getLoadMap())
431-
.compose(i -> updateStatus(reconciliation, currentKafkaRebalance, desiredStatusAndMap.getStatus(), null))
432-
.compose(updatedKafkaRebalance -> {
433-
String message = "State updated to [{}] ";
434-
if (rawRebalanceAnnotation(updatedKafkaRebalance) == null) {
435-
LOGGER.infoCr(reconciliation, message + "and annotation {} is not set ",
436-
rebalanceStateConditionType(updatedKafkaRebalance.getStatus()),
437-
ANNO_STRIMZI_IO_REBALANCE);
438-
} else {
439-
LOGGER.infoCr(reconciliation, message + "with annotation {}={} ",
440-
rebalanceStateConditionType(updatedKafkaRebalance.getStatus()),
441-
ANNO_STRIMZI_IO_REBALANCE,
442-
rawRebalanceAnnotation(updatedKafkaRebalance)
443-
);
444-
}
445-
if (hasRebalanceAnnotation(updatedKafkaRebalance)) {
446-
if (currentState != KafkaRebalanceState.ReconciliationPaused && rebalanceAnnotation != KafkaRebalanceAnnotation.none && !currentState.isValidateAnnotation(rebalanceAnnotation)) {
447-
return Future.succeededFuture();
418+
try {
419+
AbstractRebalanceOptions.AbstractRebalanceOptionsBuilder<?, ?> rebalanceOptionsBuilder = convertRebalanceSpecToRebalanceOptions(kafkaRebalance.getSpec());
420+
421+
return computeNextStatus(reconciliation, host, apiClient, kafkaRebalance, currentState, rebalanceAnnotation, rebalanceOptionsBuilder)
422+
.compose(desiredStatusAndMap -> {
423+
// More events related to resource modification might be queued with a stale state. (potentially updated by the rebalance holding the lock)
424+
// Due to possible long rebalancing operations that take the lock for the entire period,
425+
// do a new get to retrieve the current resource state.
426+
return kafkaRebalanceOperator.getAsync(reconciliation.namespace(), reconciliation.name())
427+
.compose(currentKafkaRebalance -> {
428+
if (currentKafkaRebalance != null) {
429+
return configMapOperator.reconcile(reconciliation, kafkaRebalance.getMetadata().getNamespace(),
430+
kafkaRebalance.getMetadata().getName(), desiredStatusAndMap.getLoadMap())
431+
.compose(i -> updateStatus(reconciliation, currentKafkaRebalance, desiredStatusAndMap.getStatus(), null))
432+
.compose(updatedKafkaRebalance -> {
433+
String message = "State updated to [{}] ";
434+
if (rawRebalanceAnnotation(updatedKafkaRebalance) == null) {
435+
LOGGER.infoCr(reconciliation, message + "and annotation {} is not set ",
436+
rebalanceStateConditionType(updatedKafkaRebalance.getStatus()),
437+
ANNO_STRIMZI_IO_REBALANCE);
448438
} else {
449-
LOGGER.infoCr(reconciliation, "Removing annotation {}={}",
439+
LOGGER.infoCr(reconciliation, message + "with annotation {}={} ",
440+
rebalanceStateConditionType(updatedKafkaRebalance.getStatus()),
450441
ANNO_STRIMZI_IO_REBALANCE,
451-
rawRebalanceAnnotation(updatedKafkaRebalance));
452-
// Updated KafkaRebalance has rebalance annotation removed as
453-
// action specified by user has been completed.
454-
KafkaRebalance patchedKafkaRebalance = new KafkaRebalanceBuilder(updatedKafkaRebalance)
455-
.editMetadata()
442+
rawRebalanceAnnotation(updatedKafkaRebalance)
443+
);
444+
}
445+
if (hasRebalanceAnnotation(updatedKafkaRebalance)) {
446+
if (currentState != KafkaRebalanceState.ReconciliationPaused && rebalanceAnnotation != KafkaRebalanceAnnotation.none && !currentState.isValidateAnnotation(rebalanceAnnotation)) {
447+
return Future.succeededFuture();
448+
} else {
449+
LOGGER.infoCr(reconciliation, "Removing annotation {}={}",
450+
ANNO_STRIMZI_IO_REBALANCE,
451+
rawRebalanceAnnotation(updatedKafkaRebalance));
452+
// Updated KafkaRebalance has rebalance annotation removed as
453+
// action specified by user has been completed.
454+
KafkaRebalance patchedKafkaRebalance = new KafkaRebalanceBuilder(updatedKafkaRebalance)
455+
.editMetadata()
456456
.removeFromAnnotations(ANNO_STRIMZI_IO_REBALANCE)
457-
.endMetadata()
458-
.build();
459-
return kafkaRebalanceOperator.patchAsync(reconciliation, patchedKafkaRebalance);
457+
.endMetadata()
458+
.build();
459+
return kafkaRebalanceOperator.patchAsync(reconciliation, patchedKafkaRebalance);
460+
}
461+
} else {
462+
LOGGER.debugCr(reconciliation, "No annotation {}", ANNO_STRIMZI_IO_REBALANCE);
463+
return Future.succeededFuture();
460464
}
461-
} else {
462-
LOGGER.debugCr(reconciliation, "No annotation {}", ANNO_STRIMZI_IO_REBALANCE);
463-
return Future.succeededFuture();
464-
}
465-
})
466-
.mapEmpty();
467-
} else {
468-
return Future.succeededFuture();
469-
}
470-
}, exception -> {
465+
})
466+
.mapEmpty();
467+
} else {
468+
return Future.succeededFuture();
469+
}
470+
}, exception -> {
471471
LOGGER.errorCr(reconciliation, "Status updated to [NotReady] due to error: {}", exception.getMessage());
472472
return updateStatus(reconciliation, kafkaRebalance, new KafkaRebalanceStatus(), exception)
473473
.mapEmpty();
474-
}); },
475-
exception -> {
476-
LOGGER.errorCr(reconciliation, "Status updated to [NotReady] due to error: {}", exception.getMessage());
477-
return updateStatus(reconciliation, kafkaRebalance, new KafkaRebalanceStatus(), exception)
478-
.mapEmpty();
479-
});
474+
});
475+
}, exception -> {
476+
LOGGER.errorCr(reconciliation, "Status updated to [NotReady] due to error: {}", exception.getMessage());
477+
return updateStatus(reconciliation, kafkaRebalance, new KafkaRebalanceStatus(), exception)
478+
.mapEmpty();
479+
});
480+
} catch (IllegalArgumentException e) {
481+
LOGGER.errorCr(reconciliation, "Status updated to [NotReady] due to error: {}", e.getMessage());
482+
return updateStatus(reconciliation, kafkaRebalance, new KafkaRebalanceStatus(), e).mapEmpty();
483+
}
480484
}
481485

482486
/**
@@ -776,10 +780,10 @@ private Future<MapAndStatus<ConfigMap, KafkaRebalanceStatus>> onNotReady(Reconci
776780
// CC is activated from the disabled state or the user has fixed the error on the resource and want to 'refresh'
777781
return onNew(reconciliation, host, apiClient, kafkaRebalance, rebalanceOptionsBuilder);
778782
} else {
779-
// Stay in the current NotReady state, returning null as next state
783+
// Stay in the current NotReady state
780784
Set<Condition> conditions = validate(reconciliation, kafkaRebalance);
781785
validateAnnotation(conditions, KafkaRebalanceState.NotReady, rebalanceAnnotation);
782-
return Future.succeededFuture(buildRebalanceStatus(null, KafkaRebalanceState.NotReady, conditions));
786+
return Future.succeededFuture(new MapAndStatus<>(null, buildRebalanceStatusFromPreviousStatus(kafkaRebalance.getStatus(), conditions)));
783787
}
784788
}
785789

@@ -1261,10 +1265,6 @@ private MapAndStatus<ConfigMap, KafkaRebalanceStatus> handleRebalanceResponse(Re
12611265
// If rebalance proposal is still being processed, we need to re-request the proposal at a later time
12621266
// with the corresponding session-id so we move to the PendingProposal State.
12631267
return buildRebalanceStatus(response.getUserTaskId(), KafkaRebalanceState.PendingProposal, validate(reconciliation, kafkaRebalance));
1264-
} else if (response.isBrokersNotExist()) {
1265-
// If there are some specified brokers which don't exist, it's an error at the Cruise Control level
1266-
// Need to move to NotReady state, having the user fixing the brokers list and refresh
1267-
return buildRebalanceStatus(null, KafkaRebalanceState.NotReady, validate(reconciliation, kafkaRebalance));
12681268
}
12691269
} else {
12701270
if (response.isNotEnoughDataForProposal()) {
@@ -1276,10 +1276,6 @@ private MapAndStatus<ConfigMap, KafkaRebalanceStatus> handleRebalanceResponse(Re
12761276
// soon as it is ready, so set the state to rebalancing.
12771277
// In the onRebalancing method the optimization proposal will be added when it is ready.
12781278
return buildRebalanceStatus(response.getUserTaskId(), KafkaRebalanceState.Rebalancing, validate(reconciliation, kafkaRebalance));
1279-
} else if (response.isBrokersNotExist()) {
1280-
// If there are some specified brokers which don't exist, it's an error at the Cruise Control level
1281-
// Need to move to NotReady state, having the user fixing the brokers list and refresh
1282-
return buildRebalanceStatus(null, KafkaRebalanceState.NotReady, validate(reconciliation, kafkaRebalance));
12831279
}
12841280
}
12851281

‎cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlApiImpl.java

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -191,9 +191,7 @@ private void internalRebalance(String host, int port, String path, String userTa
191191
// ... or one or more brokers doesn't exist on a add/remove brokers rebalance request
192192
} else if (json.getString(CC_REST_API_ERROR_KEY).contains("IllegalArgumentException") &&
193193
json.getString(CC_REST_API_ERROR_KEY).contains("does not exist.")) {
194-
CruiseControlRebalanceResponse ccResponse = new CruiseControlRebalanceResponse(userTaskID, json);
195-
ccResponse.setBrokersNotExist(true);
196-
result.complete(ccResponse);
194+
result.fail(new IllegalArgumentException("Some/all brokers specified don't exist"));
197195
} else {
198196
// If there was any other kind of error propagate this to the operator
199197
result.fail(new CruiseControlRestException(

‎cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlRebalanceResponse.java

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@ public class CruiseControlRebalanceResponse extends CruiseControlResponse {
1010

1111
private boolean isNotEnoughDataForProposal;
1212
private boolean isProposalStillCalculating;
13-
private boolean isBrokersNotExist;
1413

1514
CruiseControlRebalanceResponse(String userTaskId, JsonObject json) {
1615
super(userTaskId, json);
@@ -20,8 +19,6 @@ public class CruiseControlRebalanceResponse extends CruiseControlResponse {
2019
// Proposal is not in progress unless response from Cruise Control says otherwise
2120
// Sourced from the "progress" field in the response with value "proposalStillCalaculating"
2221
this.isProposalStillCalculating = false;
23-
// One or more brokers provided during a rebalance with add/remove broker endpoints don't exist
24-
this.isBrokersNotExist = false;
2522
}
2623

2724
public boolean isNotEnoughDataForProposal() {
@@ -39,12 +36,4 @@ public boolean isProposalStillCalculating() {
3936
public void setProposalStillCalculating(boolean proposalStillCalculating) {
4037
this.isProposalStillCalculating = proposalStillCalculating;
4138
}
42-
43-
public boolean isBrokersNotExist() {
44-
return this.isBrokersNotExist;
45-
}
46-
47-
public void setBrokersNotExist(boolean brokersNotExist) {
48-
this.isBrokersNotExist = brokersNotExist;
49-
}
5039
}

0 commit comments

Comments
 (0)