Addressing comments - pp, tc

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

Author: kyguy

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/cruisecontrol/ExecutorStateProcessor.java

@@ -7,7 +7,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 
 import java.time.Instant;
-import java.util.Arrays;
 import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -90,11 +89,15 @@ public static void verifyRebalancingState(JsonNode executorJson) {
         }
 
         private static ExecutorState fromString(String state) {
-            String normalized = state.trim();
-            return Arrays.stream(ExecutorState.values())
-                    .filter(e -> e.name().equals(normalized))
-                    .findFirst()
-                    .orElseThrow(() -> new IllegalArgumentException("Unknown ExecutorState: " + state));
+            if (state == null) {
+                throw new IllegalArgumentException("ExecutorState cannot be null");
+            }
+
+            try {
+                return ExecutorState.valueOf(state);
+            } catch (IllegalArgumentException e) {
+                throw new IllegalArgumentException("Unknown ExecutorState: " + state, e);
+            }
         }
     }
 

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

@@ -358,6 +358,7 @@ private static Future<MapAndStatus<ConfigMap, KafkaRebalanceStatus>> updateProgr
                     desiredStatus.setProgress(Map.of(REBALANCE_PROGRESS_CONFIG_MAP_KEY, configMapName));
                     return KafkaRebalanceConfigMapUtils.updateRebalanceConfigMap(reconciliation, desiredStatus, host, cruiseControlPort, apiClient, desiredConfigMap)
                             .recover(exception -> {
+                                LOGGER.infoCr(reconciliation, "Progress information of rebalance not updated due to the following error: {}", exception);
                                 KafkaRebalanceUtils.addWarningCondition(desiredStatus, exception);
                                 return Future.succeededFuture();
                             })

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceConfigMapUtils.java

@@ -24,8 +24,21 @@
 public class KafkaRebalanceConfigMapUtils {
 
     /* test */ static final String REBALANCE_PROGRESS_CONFIG_MAP_KEY = "rebalanceProgressConfigMap";
+    /**
+     * The estimated time it will take in minutes until the partition rebalance is complete rounded
+     * to the nearest minute.
+     */
     /* test */ static final String ESTIMATED_TIME_TO_COMPLETION_IN_MINUTES_KEY = "estimatedTimeToCompletionInMinutes";
+    /**
+     *  The percentage of the byte movement of the partition rebalance that is completed as a rounded down integer
+     *  value in the range [0-100].
+     */
     /* test */ static final String COMPLETED_BYTE_MOVEMENT_PERCENTAGE_KEY = "completedByteMovementPercentage";
+    /**
+     * The “non-verbose” JSON payload from the /kafkacruisecontrol/state?substates=executor endpoint,
+     * providing details about the executor's current status, including partition movement progress,
+     * concurrency limits, and total data to move.
+     */
     /* test */ static final String EXECUTOR_STATE_KEY = "executorState.json";
 
     /* test */ static final String TIME_COMPLETED = "0";
@@ -58,7 +71,6 @@ public class KafkaRebalanceConfigMapUtils {
 
                 int estimatedTimeToCompletion = KafkaRebalanceProgressUtils.estimateTimeToCompletionInMinutes(
                         taskStartTime,
-                        Instant.now(),
                         totalDataToMove,
                         finishedDataMovement);
 

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceProgressUtils.java

@@ -16,7 +16,6 @@ public class KafkaRebalanceProgressUtils {
      * Estimates the number of minutes it will take an ongoing partition rebalance to complete.
      *
      * @param taskStartTime The time when the task started.
-     * @param currentTime The time at the moment of the method call.
      * @param totalDataToMoveInMB The total amount of data that needs to be moved, in megabytes.
      * @param finishedDataMovementInMB The amount of data that has already been moved, in megabytes.
      * @return The estimated time to completion in minutes.
@@ -27,15 +26,27 @@ public class KafkaRebalanceProgressUtils {
      *     - The elapsed time between `taskStartTime` and `currentTime` is zero, making rate calculation impossible.
      *     - The data movement rate is zero, making the time to completion estimation impossible.
      */
+    protected static int estimateTimeToCompletionInMinutes(Instant taskStartTime,
+                                                 int totalDataToMoveInMB,
+                                                 int finishedDataMovementInMB)
+            throws IllegalArgumentException, ArithmeticException {
+        return estimateTimeToCompletionInMinutes(
+                taskStartTime,
+                Instant.now(),
+                totalDataToMoveInMB,
+                finishedDataMovementInMB
+        );
+    }
+
     /* test */ static int estimateTimeToCompletionInMinutes(Instant taskStartTime,
                                                             Instant currentTime,
                                                             int totalDataToMoveInMB,
                                                             int finishedDataMovementInMB)
             throws IllegalArgumentException, ArithmeticException {
-        if (taskStartTime == null || currentTime == null || totalDataToMoveInMB < 0 || finishedDataMovementInMB < 0) {
+        if (taskStartTime == null || totalDataToMoveInMB < 0 || finishedDataMovementInMB < 0) {
             throw new IllegalArgumentException(
-                    String.format("Invalid value(s) provided for one of the following arguments: taskStartTime %s, currentTime %s, totalDataToMoveInMB: %d, finishedDataMovementInMB: %d.",
-                            taskStartTime, currentTime, totalDataToMoveInMB, finishedDataMovementInMB)
+                    String.format("Invalid value(s) provided for one of the following arguments: taskStartTime %s, totalDataToMoveInMB: %d, finishedDataMovementInMB: %d.",
+                            taskStartTime, totalDataToMoveInMB, finishedDataMovementInMB)
             );
         }
 
@@ -52,9 +63,9 @@ public class KafkaRebalanceProgressUtils {
             throw new ArithmeticException("finishedDataMovementInMB is zero, cannot estimate time to completion.");
         }
 
-        float rateMBperMinute = ((float) finishedDataMovementInMB / timeElapsed.getSeconds()) * 60;
+        double rateMBperMinute = ((double) finishedDataMovementInMB / timeElapsed.getSeconds()) * 60;
         int dataLeftToMoveMB = totalDataToMoveInMB - finishedDataMovementInMB;
-        return Math.round(dataLeftToMoveMB / rateMBperMinute);
+        return (int) Math.round(dataLeftToMoveMB / rateMBperMinute);
     }
 
     /**

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceUtils.java

@@ -14,7 +14,6 @@
 import io.strimzi.api.kafka.model.kafka.cruisecontrol.KafkaAutoRebalanceStatusBuilder;
 import io.strimzi.api.kafka.model.rebalance.KafkaRebalanceState;
 import io.strimzi.api.kafka.model.rebalance.KafkaRebalanceStatus;
-import io.strimzi.operator.cluster.operator.resource.ConditionType;
 import io.strimzi.operator.common.model.StatusUtils;
 
 import java.util.ArrayList;
@@ -179,7 +178,7 @@ public static void addWarningCondition(KafkaRebalanceStatus status, Throwable ex
 
     /* test */ static Condition getWarningCondition(KafkaRebalanceStatus status) {
         return status.getConditions().stream()
-                .filter(ConditionType::isWarning)
+                .filter(condition -> condition.getType().equals("Warning"))
                 .findFirst()
                 .orElse(null);
     }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/ConditionType.java

@@ -50,7 +50,6 @@ public class CruiseControlApiImpl implements CruiseControlApi {
      */
     public static final int HTTP_DEFAULT_IDLE_TIMEOUT_SECONDS = -1;
     private static final String STATUS_KEY = "Status";
-    private static final String START_MS_KEY = "StartMs";
     private final long idleTimeout;
     private final boolean apiSslEnabled;
     private final HTTPHeader authHttpHeader;
@@ -364,7 +363,6 @@ public CompletableFuture<CruiseControlUserTasksResponse> getUserTaskStatus(Recon
                         } else {
                             JsonNode jsonUserTask = userTasks.get(0);
                             String taskStatusStr = jsonUserTask.get(STATUS_KEY).asText();
-                            String taskStartTime = jsonUserTask.get(START_MS_KEY).asText();
                             LOGGER.debugCr(reconciliation, "Got {} response to GET request to {} : userTaskID = {}, status = {}", response.statusCode(), path, userTaskID, taskStatusStr);
                             // This should not be an error with a 200 status but we play it safe
                             if (jsonUserTask.has(CC_REST_API_ERROR_KEY)) {
@@ -374,7 +372,6 @@ public CompletableFuture<CruiseControlUserTasksResponse> getUserTaskStatus(Recon
                             }
 
                             statusJson.put(STATUS_KEY, taskStatusStr);
-                            statusJson.put(START_MS_KEY, taskStartTime);
                             CruiseControlUserTaskStatus taskStatus = CruiseControlUserTaskStatus.lookup(taskStatusStr);
                             switch (taskStatus) {
                                 case ACTIVE:

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

@@ -5,20 +5,12 @@
 package io.strimzi.operator.cluster.operator.resource.cruisecontrol;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import io.strimzi.operator.common.model.cruisecontrol.CruiseControlUserTaskStatus;
-
-import java.time.Instant;
 
 /**
  * Response to user tasks request
  */
 public class CruiseControlUserTasksResponse extends CruiseControlResponse {
 
-    private static final String STATUS_KEY = "Status";
-    private static final String START_MS_KEY = "StartMs";
-
-    private CruiseControlUserTaskStatus status;
-    private Instant taskStartTime;
     private boolean isMaxActiveUserTasksReached;
 
     /**
@@ -32,8 +24,6 @@ public class CruiseControlUserTasksResponse extends CruiseControlResponse {
         // The maximum number of active user tasks that can run concurrently has reached
         // Sourced from the error message that contains "reached the servlet capacity" from the Cruise Control response
         this.isMaxActiveUserTasksReached = false;
-        setTaskStatus(json);
-        setTaskStartTime(json);
     }
 
     /**
@@ -47,41 +37,4 @@ protected void setMaxActiveUserTasksReached(boolean maxActiveUserTasksReached) {
         this.isMaxActiveUserTasksReached = maxActiveUserTasksReached;
     }
 
-    private void setTaskStatus(JsonNode userTaskJson) {
-        if (userTaskJson != null && userTaskJson.has(STATUS_KEY)) {
-            this.status = CruiseControlUserTaskStatus.lookup(userTaskJson.get("Status").asText());
-        }
-    }
-
-    /**
-     * @return status of user task.
-     */
-    public CruiseControlUserTaskStatus getTaskStatus() {
-        return this.status;
-    }
-
-    /**
-     * Extracts the task start time from the provided `userTaskJson` JSON object.
-     * The task start time is extracted from the "StartMs" field, which contains a
-     * timestamp in of Unix epoch in milliseconds. The timestamp is then parsed into Instant object.
-     *
-     * @param userTaskJson The `JsonNode` object containing the user task json,
-     *                      from which the task start time will be extracted.
-     */
-    private void setTaskStartTime(JsonNode userTaskJson) {
-        if (userTaskJson != null && userTaskJson.has(START_MS_KEY)) {
-            // Extract the task start time as Unix epoch in milliseconds
-            long taskStartTimeInMilliseconds = userTaskJson.get(START_MS_KEY).asLong();
-
-            this.taskStartTime = Instant.ofEpochMilli(taskStartTimeInMilliseconds);
-        }
-    }
-
-    /**
-     * @return Start time of user task as an Instant object
-     */
-    public Instant getTaskStartTime() {
-        return this.taskStartTime;
-    }
-
 }

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

@@ -59,6 +59,14 @@ public void testProgressFieldsForProposalRebalancingState() throws Exception {
         ConfigMap cm = createKafkaRebalanceConfigMap(BROKER_LOAD_MAP);
         updateRebalanceConfigMapWithProgressFields(KafkaRebalanceState.Rebalancing, es0, cm);
 
+        /**
+         * Total Data to Move:          10,000 MB
+         * Data Moved:                  250 MB
+         * Time Since Task Start:       2 seconds
+         * Data Rate:                   125 MB/s
+         * Remaining Data:              9,750 MB
+         * Estimated Time to Complete:  78 seconds
+         */
         Map<String, String> m =  cm.getData();
         assertThat(m.get(ESTIMATED_TIME_TO_COMPLETION_IN_MINUTES_KEY), is("1"));
         assertThat(m.get(COMPLETED_BYTE_MOVEMENT_PERCENTAGE_KEY), is("2"));

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceConfigMapUtilsTest.java

@@ -20,15 +20,86 @@ public class KafkaRebalanceProgressUtilsTest {
     public void testEstimateTimeToCompletionInMinutes() {
         Instant currentTime = Instant.now();
 
+        /**
+         * Total Data to Move:          1000 MB
+         * Data Moved:                  10 MB
+         * Time Since Task Start:       1 second
+         * Data Rate:                   600 MB/s
+         * Remaining Data:              990 MB
+         * Estimated Time to Complete:  1.5 seconds -> 2 minutes
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minusSeconds(1), currentTime, 1000, 10), is(2));
+        /**
+         * Total Data to Move:          1000 MB
+         * Data Moved:                  10 MB
+         * Time Since Task Start:       60 seconds
+         * Data Rate:                   0.167 MB/s
+         * Remaining Data:              990 MB
+         * Estimated Time to Complete:  5940 seconds -> 99 minutes
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minusSeconds(60), currentTime, 1000, 10), is(99));
+        /**
+         * Total Data to Move:          1000 MB
+         * Data Moved:                  500 MB
+         * Time Since Task Start:       60 seconds
+         * Data Rate:                   8.33 MB/s
+         * Remaining Data:              500 MB
+         * Estimated Time to Complete:  60 seconds -> 1 minute
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minusSeconds(60), currentTime, 1000, 500), is(1));
-
+        /**
+         * Total Data to Move:          1,000,000 MB
+         * Data Moved:                  100 MB
+         * Time Since Task Start:       60 seconds
+         * Data Rate:                   1.6666666666666667 MB/s
+         * Remaining Data:              999,900 MB
+         * Estimated Time to Complete:  599,940 seconds -> 9999 minutes
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minusSeconds(60), currentTime, 1000000, 100), is(9999));
+        /**
+         * Total Data to Move:          1000 MB
+         * Data Moved:                  990 MB
+         * Time Since Task Start:       60 seconds
+         * Data Rate:                   16.5 MB/s
+         * Remaining Data:              10 MB
+         * Estimated Time to Complete:  ~0.6 seconds -> 0 minutes
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minusSeconds(60), currentTime, 1000, 990), is(0));
+        /**
+         * Total Data to Move:          2,147,483,647 MB (Integer.MAX_VALUE)
+         * Data Moved:                  1,073,741,823.5 MB
+         * Time Since Task Start:       1000 seconds
+         * Data Rate:                   1073741.823 MB/s
+         * Remaining Data:              1,073,741,824.5 MB
+         * Estimated Time to Complete:  ~1000 seconds -> 17 minutes
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minusSeconds(1000), currentTime, Integer.MAX_VALUE, Integer.MAX_VALUE / 2), is(17));
+        /**
+         * Total Data to Move:          1000 MB
+         * Data Moved:                  10 MB
+         * Time Since Task Start:       1000 seconds
+         * Data Rate:                   0.01 MB/s
+         * Remaining Data:              990 MB
+         * Estimated Time to Complete:  99000 seconds -> 1650 minutes
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minusSeconds(1000), currentTime, 1000, 10), is(1650));
+        /**
+         * Total Data to Move:          1 MB
+         * Data Moved:                  1 MB
+         * Time Since Task Start:       1 hour (3600 seconds)
+         * Data Rate:                   0.000278 MB/s
+         * Remaining Data:              0 MB
+         * Estimated Time to Complete:  0 seconds -> 0 minutes
+         */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minus(1, ChronoUnit.HOURS), currentTime, 1, 1), is(0));
+        /**
+        * Total Data to Move:          1000 MB
+        * Data Moved:                  500 MB
+        * Time Since Task Start:       30 days (2,592,000 seconds)
+        * Data Rate:                   ~0.000193 MB/s
+        * Remaining Data:              500 MB
+        * Estimated Time to Complete:  2,592,000 seconds -> 43,200 minutes
+        */
         assertThat(estimateTimeToCompletionInMinutes(currentTime.minus(30, ChronoUnit.DAYS), currentTime, 1000, 500), is(43200));
 
         assertThrows(IllegalArgumentException.class, () -> estimateTimeToCompletionInMinutes(currentTime.plusSeconds(1), currentTime, 1000, 10));

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaRebalanceProgressUtilsTest.java

@@ -175,16 +175,16 @@ public void testUpdateStatusCondition() {
         // Test Warning condition can be added when there are no existing conditions
         KafkaRebalanceUtils.addWarningCondition(status, exception("Example error 0"));
         assertThat(status.getConditions().size(), is(1));
-        assertThat(KafkaRebalanceUtils.getWarningCondition(status).getType(), is("Warning"));
+        assertThat(KafkaRebalanceUtils.getWarningCondition(status), notNullValue());
 
         // Test Warning condition addition doesn't remove existing conditions.
         Condition c0 = new ConditionBuilder()
-                .withType("Rebalancing")
+                .withType(KafkaRebalanceState.Rebalancing.toString())
                 .build();
         status.setConditions(List.of(c0));
         KafkaRebalanceUtils.addWarningCondition(status, exception("Example error 0"));
         assertThat(status.getConditions().size(), is(2));
-        assertThat(KafkaRebalanceUtils.getWarningCondition(status).getType(), is("Warning"));
+        assertThat(KafkaRebalanceUtils.getWarningCondition(status), notNullValue());
 
         // Test original Warning is kept when a new error with same reason and message is thrown.
         String timestamp = "2024-11-05T15:28:23.995129903Z";