Repository: linkedin/cruise-control (java)

Issue:
Maximum replication factor is not considered when assessing whether the cluster is over-provisioned
While goals such as `RackAwareDistributionGoal` and `RackAwareGoal` take the maximum replication factor into account when assessing over-provisioning, other goals do not. This can lead to provision responses that suggest removing brokers, potentially violating the replication factor of certain topics.

Pull request description:
Consider max RF when assessing whether the cluster is over-provisioned
This PR resolves #2170.


Patch:
diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/AbstractGoal.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/AbstractGoal.java
index ccf1f896..6a2b9b63 100644
--- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/AbstractGoal.java
+++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/AbstractGoal.java
@@ -117,10 +117,10 @@ public abstract class AbstractGoal implements Goal {
                                                         + " Reason: %s.", name(), comparator.explainLastComparison()));
         }
       }
-      // Ensure that a cluster is not identified as over provisioned unless it has the minimum required number of alive brokers.
-      if (_provisionResponse.status() == OVER_PROVISIONED && clusterModel.aliveBrokers().size() < _balancingConstraint.overprovisionedMinBrokers()) {
-        _provisionResponse = new ProvisionResponse(RIGHT_SIZED);
-      }
+      // Ensure that a cluster is not identified as over provisioned unless it has the minimum required number of alive brokers and
+      // expected number of brokers after the provisioning will still be larger than or equal to the max RF
+      _provisionResponse =
+          GoalUtils.validateProvisionResponse(_provisionResponse, clusterModel, _balancingConstraint.overprovisionedMinBrokers());
       return _succeeded;
     } catch (OptimizationFailureException ofe) {
       _provisionResponse = new ProvisionResponse(UNDER_PROVISIONED, ofe.provisionRecommendation(), name());
diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtils.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtils.java
index 2d946e1e..4137b1b7 100644
--- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtils.java
+++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtils.java
@@ -9,6 +9,7 @@ import com.linkedin.kafka.cruisecontrol.analyzer.ActionType;
 import com.linkedin.kafka.cruisecontrol.analyzer.BalancingConstraint;
 import com.linkedin.kafka.cruisecontrol.analyzer.OptimizationOptions;
 import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionRecommendation;
+import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionResponse;
 import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionStatus;
 import com.linkedin.kafka.cruisecontrol.analyzer.goals.rackaware.RackAwareGoalRackIdMapper;
 import com.linkedin.kafka.cruisecontrol.common.Resource;
@@ -599,4 +600,52 @@ public final class GoalUtils {
     }
     return (balancePercentage - 1) * balanceMargin;
   }
+
+  /**
+   * This method validates that provision response is {@link ProvisionStatus#OVER_PROVISIONED} only if:
+   * <ul>
+   *   <li>Number of alive brokers is larger than or equal to the value of {@link AnalyzerConfig#OVERPROVISIONED_MIN_BROKERS_CONFIG}.</li>
+   *   <li>Expected number of alive brokers after the provisioning will still be larger than or equal to the maximum replication factor.</li>
+   * </ul>
+   * if any of the above conditions is violated then provision response is set to {@link ProvisionStatus#RIGHT_SIZED}
+   *
+   * @param provisionResponse Provision response after goal is optimization
+   * @param clusterModel Cluster usage model
+   * @param overprovisionedMinBrokers value of the {@link AnalyzerConfig#OVERPROVISIONED_MIN_BROKERS_CONFIG}
+   * @return Validated provision response
+   * @throws IllegalArgumentException when provision status is {@link ProvisionStatus#OVER_PROVISIONED}
+   * and goal doesn't have exactly one recommendation
+   */
+  public static ProvisionResponse validateProvisionResponse(ProvisionResponse provisionResponse, ClusterModel clusterModel,
+                                                            int overprovisionedMinBrokers) {
+    if (provisionResponse.status() != ProvisionStatus.OVER_PROVISIONED) {
+      return provisionResponse;
+    }
+    // ensure that a cluster is not identified as over provisioned unless it has the minimum required number of alive brokers
+    if (clusterModel.aliveBrokers().size() < overprovisionedMinBrokers) {
+      return new ProvisionResponse(ProvisionStatus.RIGHT_SIZED);
+    }
+    // when status is OVER_PROVISIONED goal is expected to have exactly 1 recommendation
+    if (provisionResponse.recommendationByRecommender().size() != 1) {
+      throw new IllegalArgumentException(String.format("Expected to have exactly 1 provision recommendation, but got: %d",
+                                                       provisionResponse.recommendationByRecommender().size()));
+    }
+    String goalName = provisionResponse.recommendationByRecommender().keySet().iterator().next();
+    int numBrokersToDrop = provisionResponse.recommendationByRecommender().values().iterator().next().numBrokers();
+    // this variable indicates the maximum number of brokers that can be safely dropped.
+    int maxAllowedNumBrokersToDrop = clusterModel.aliveBrokers().size() - clusterModel.maxReplicationFactor();
+    if (numBrokersToDrop <= maxAllowedNumBrokersToDrop) {
+      // return provision response as it is if the recommended number of brokers to drop is smaller or equal to the max allowed number of
+      // brokers that can be safely dropped
+      return provisionResponse;
+    } else if (maxAllowedNumBrokersToDrop > 0) {
+      // change the recommended number of brokers to drop if the original recommendation can not be safely satisfied
+      ProvisionRecommendation recommendation =
+          new ProvisionRecommendation.Builder(ProvisionStatus.OVER_PROVISIONED).numBrokers(maxAllowedNumBrokersToDrop).build();
+      return new ProvisionResponse(ProvisionStatus.OVER_PROVISIONED, recommendation, goalName);
+    } else {
+      // change provision response status to RIGHT_SIZED if none of the brokers can be safely dropped
+      return new ProvisionResponse(ProvisionStatus.RIGHT_SIZED);
+    }
+  }
 }
diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/UserTaskManagerConfig.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/UserTaskManagerConfig.java
index 3dd23943..55ef8155 100644
--- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/UserTaskManagerConfig.java
+++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/UserTaskManagerConfig.java
@@ -9,7 +9,6 @@ import org.apache.kafka.common.config.ConfigDef;
 
 import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
 
-
 /**
  * A class to keep Cruise Control User Task Manager Configs and defaults.
  * DO NOT CHANGE EXISTING CONFIG NAMES AS CHANGES WOULD BREAK USER CODE.
diff --git a/docs/wiki/User Guide/Configurations.md b/docs/wiki/User Guide/Configurations.md
index 5100b9cc..6e509dee 100644
--- a/docs/wiki/User Guide/Configurations.md
+++ b/docs/wiki/User Guide/Configurations.md
@@ -239,7 +239,7 @@ The following configurations are inherited from the open source Kafka client con
 ### UserTaskManager Configurations
 | Name                                                          | Type    | Required?   | Default Value | Description                                                                                                                                                                                                                               |
 |---------------------------------------------------------------|---------|-------------|---------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| completed.user.task.retention.time.ms                         | Integer | Y           |               | The maximum time in milliseconds to store the response and access details of a completed user task.                                                                                                                                       |
+| completed.user.task.retention.time.ms                         | Integer | Y           | 86400000      | The maximum time in milliseconds to store the response and access details of a completed user task.                                                                                                                                       |
 | max.cached.completed.user.tasks                               | Integer | N           | 100           | The maximum number of completed user tasks for which the response and access details will be cached.                                                                                                                                      |
 | max.active.user.tasks                                         | Integer | N           | 5             | The maximum number of user tasks for concurrently running in async endpoints across all users.                                                                                                                                            |
 | completed.kafka.monitor.user.task.retention.time.ms	        | Long	  | N	        | null	        | The maximum time in milliseconds to store the response and access details of a completed kafka monitoring user task. If this config is missing, the value set in config completed.user.task.retention.time.ms will be used.	            |
diff --git a/gradle.properties b/gradle.properties
index 47e57fd5..80315997 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -1,9 +1,9 @@
 org.gradle.daemon=false
 org.gradle.parallel=false
 org.gradle.jvmargs=-Xms512m -Xmx512m
-scalaVersion=2.13.10
+scalaVersion=2.13.13
 kafkaVersion=3.5.1
 zookeeperVersion=3.8.3
 nettyVersion=4.1.100.Final
 jettyVersion=9.4.53.v20231009
-vertxVersion=4.5.0
+vertxVersion=4.5.8


Test patch:
diff --git a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtilsTest.java b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtilsTest.java
index 1be52c05..d1e86c6d 100644
--- a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtilsTest.java
+++ b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/analyzer/goals/GoalUtilsTest.java
@@ -5,11 +5,21 @@
 package com.linkedin.kafka.cruisecontrol.analyzer.goals;
 
 import com.linkedin.kafka.cruisecontrol.analyzer.BalancingConstraint;
+import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionRecommendation;
+import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionResponse;
 import com.linkedin.kafka.cruisecontrol.common.Resource;
+import com.linkedin.kafka.cruisecontrol.model.Broker;
+import com.linkedin.kafka.cruisecontrol.model.ClusterModel;
+import java.util.Set;
 import org.easymock.EasyMock;
 import org.junit.Assert;
 import org.junit.Test;
 
+import static com.linkedin.kafka.cruisecontrol.analyzer.ProvisionStatus.OVER_PROVISIONED;
+import static com.linkedin.kafka.cruisecontrol.analyzer.ProvisionStatus.RIGHT_SIZED;
+import static org.junit.Assert.assertEquals;
+
+
 public class GoalUtilsTest {
 
   private static final double AVG_UTILIZATION_PERCENTAGE = 0.3;
@@ -60,4 +70,43 @@ public class GoalUtilsTest {
     EasyMock.verify(mockBalanceConstraint);
     Assert.assertEquals(expectedComputedBalanceThreshold, computedBalanceThreshold, 0.0);
   }
+
+  @Test
+  public void testValidateProvisionResponse() {
+    // min required num of brokers > num of brokers
+    ProvisionResponse response = GoalUtils.validateProvisionResponse(provisionResponse(1), clusterModel(3), 6);
+    assertEquals(RIGHT_SIZED, response.status());
+
+    // recommended num of brokers to drop is 1 and max allowed num of brokers to drop is 2, so 1 broker can be safely dropped
+    response = GoalUtils.validateProvisionResponse(provisionResponse(1), clusterModel(2), 4);
+    assertEquals(OVER_PROVISIONED, response.status());
+    assertEquals(1, response.recommendationByRecommender().values().iterator().next().numBrokers());
+
+    // recommended num of brokers to drop is 3 but max allowed num of brokers to drop is 1, so final provision response is to drop 1 broker
+    response = GoalUtils.validateProvisionResponse(provisionResponse(3), clusterModel(3), 4);
+    assertEquals(OVER_PROVISIONED, response.status());
+    assertEquals(1, response.recommendationByRecommender().values().iterator().next().numBrokers());
+
+    // recommended num of brokers to drop is 1 but max allowed num of brokers to drop is 0, so none of the brokers can be dropped
+    response = GoalUtils.validateProvisionResponse(provisionResponse(1), clusterModel(4), 4);
+    assertEquals(RIGHT_SIZED, response.status());
+  }
+
+  private ClusterModel clusterModel(int maxRF) {
+    ClusterModel mockClusterModel = EasyMock.mock(ClusterModel.class);
+    EasyMock.expect(mockClusterModel.aliveBrokers())
+            .andReturn(
+                Set.of(EasyMock.mock(Broker.class), EasyMock.mock(Broker.class), EasyMock.mock(Broker.class), EasyMock.mock(Broker.class)))
+            .times(2);
+    EasyMock.expect(mockClusterModel.maxReplicationFactor()).andReturn(maxRF);
+    EasyMock.replay(mockClusterModel);
+
+    return mockClusterModel;
+  }
+
+  private ProvisionResponse provisionResponse(int numBrokers) {
+    ProvisionRecommendation recommendation =
+        new ProvisionRecommendation.Builder(OVER_PROVISIONED).numBrokers(numBrokers).build();
+    return new ProvisionResponse(OVER_PROVISIONED, recommendation, "goal");
+  }
 }