Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: Correct the order of generating default and overridden broker capacities in Capacity.processCapacityEntries #10509

Merged
merged 2 commits into from
Aug 30, 2024
Merged

fix: Correct the order of generating default and overridden broker capacities in Capacity.processCapacityEntries #10509

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f1fe805
fix: Correct the order of generating default and overridden broker ca…
Xander-run Aug 27, 2024
857c9b9
chore: Rearrange the place for initialize overrideIds
Xander-run Aug 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 30 additions & 29 deletions cluster-operator/src/main/java/io/strimzi/operator/cluster/model/cruisecontrol/Capacity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,44 +365,45 @@ private void processCapacityEntries(CruiseControlSpec spec, Set<NodeRef> kafkaBr
String inboundNetwork = processInboundNetwork(brokerCapacity, null);
String outboundNetwork = processOutboundNetwork(brokerCapacity, null);

// We create a capacity for each broker node
// Initialize default capacities
for (NodeRef node : kafkaBrokerNodes) {
DiskCapacity disk = processDisk(kafkaStorage.get(node.poolName()), node.nodeId());
CpuCapacity cpu = processCpu(null, brokerCapacity, kafkaBrokerResources.get(node.poolName()));

BrokerCapacity broker = new BrokerCapacity(node.nodeId(), cpu, disk, inboundNetwork, outboundNetwork);
capacityEntries.put(node.nodeId(), broker);
}

if (brokerCapacity != null) {
// For checking for duplicate brokerIds
Set<Integer> overrideIds = new HashSet<>();
List<BrokerCapacityOverride> overrides = brokerCapacity.getOverrides();
// Override broker entries
if (overrides != null) {
if (overrides.isEmpty()) {
LOGGER.warnCr(reconciliation, "Ignoring empty overrides list");
} else {
for (BrokerCapacityOverride override : overrides) {
List<Integer> ids = override.getBrokers();
inboundNetwork = processInboundNetwork(brokerCapacity, override);
outboundNetwork = processOutboundNetwork(brokerCapacity, override);
for (int id : ids) {
if (id == BrokerCapacity.DEFAULT_BROKER_ID) {
LOGGER.warnCr(reconciliation, "Ignoring broker capacity override with illegal broker id -1.");
} else {
if (capacityEntries.containsKey(id)) {
if (overrideIds.add(id)) {
BrokerCapacity brokerCapacityEntry = capacityEntries.get(id);
brokerCapacityEntry.setCpu(processCpu(override, brokerCapacity, kafkaBrokerResources.get(Integer.toString(id))));
brokerCapacityEntry.setInboundNetwork(inboundNetwork);
brokerCapacityEntry.setOutboundNetwork(outboundNetwork);
} else {
LOGGER.warnCr(reconciliation, "Duplicate broker id {} found in overrides, using first occurrence.", id);
}
// Override default capacities
if (brokerCapacity != null) {
List<BrokerCapacityOverride> overrides = brokerCapacity.getOverrides();
// Override broker entries
if (overrides != null) {
if (overrides.isEmpty()) {
LOGGER.warnCr(reconciliation, "Ignoring empty overrides list");
} else {
// For checking for duplicate brokerIds
Set<Integer> overrideIds = new HashSet<>();
for (BrokerCapacityOverride override : overrides) {
List<Integer> ids = override.getBrokers();
inboundNetwork = processInboundNetwork(brokerCapacity, override);
outboundNetwork = processOutboundNetwork(brokerCapacity, override);
for (int id : ids) {
if (id == BrokerCapacity.DEFAULT_BROKER_ID) {
LOGGER.warnCr(reconciliation, "Ignoring broker capacity override with illegal broker id -1.");
} else {
if (capacityEntries.containsKey(id)) {
if (overrideIds.add(id)) {
BrokerCapacity brokerCapacityEntry = capacityEntries.get(id);
brokerCapacityEntry.setCpu(processCpu(override, brokerCapacity, kafkaBrokerResources.get(Integer.toString(id))));
brokerCapacityEntry.setInboundNetwork(inboundNetwork);
brokerCapacityEntry.setOutboundNetwork(outboundNetwork);
} else {
LOGGER.warnCr(reconciliation, "Ignoring broker capacity override for unknown node ID {}", id);
overrideIds.add(id);
LOGGER.warnCr(reconciliation, "Duplicate broker id {} found in overrides, using first occurrence.", id);
}
} else {
LOGGER.warnCr(reconciliation, "Ignoring broker capacity override for unknown node ID {}", id);
overrideIds.add(id);
}
}
}
Expand Down
Loading