Add more logging to help debugging the time spent on goal based operation

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/async/progress/OperationProgress.java

@@ -100,6 +100,26 @@ public synchronized String toString() {
     return sb.toString();
   }
 
+  /**
+   * @return the up-to-date total execution time on the progress
+   */
+  public synchronized long getCurrentTotalExecutionTimeMs() {
+    if (_startTimes.isEmpty()) {
+      return 0;
+    } else {
+      long currentTime = System.currentTimeMillis();
+      long startTime = _startTimes.get(0);
+      return currentTime - startTime;
+    }
+  }
+
+  /**
+   * @return the name of the operation
+   */
+  public String getOperation() {
+    return _operation;
+  }
+
   /**
    * @return The map describing the progress of the operation.
    */

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/Executor.java

@@ -832,6 +832,9 @@ public synchronized void executeProposals(Collection<ExecutionProposal> proposal
                             isTriggeredByUserRequest, loadMonitor);
       startExecution(loadMonitor, null, removedBrokers, replicationThrottle, isTriggeredByUserRequest);
     } catch (Exception e) {
+      if (e instanceof OngoingExecutionException) {
+        LOG.info("Broker removal operation with uuid {} aborted due to ongoing execution", uuid);
+      }
       processExecuteProposalsFailure();
       throw e;
     }

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/servlet/handler/async/runnable/GoalBasedOperationRunnable.java

@@ -17,6 +17,8 @@
 import java.util.concurrent.TimeoutException;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import static com.linkedin.kafka.cruisecontrol.KafkaCruiseControlUtils.goalsByPriority;
 import static com.linkedin.kafka.cruisecontrol.KafkaCruiseControlUtils.sanityCheckGoals;
@@ -34,6 +36,7 @@
  * An abstract class to extract the common logic of goal based operation runnables.
  */
 public abstract class GoalBasedOperationRunnable extends OperationRunnable {
+  private static final Logger LOG = LoggerFactory.getLogger(GoalBasedOperationRunnable.class);
   protected final List<String> _goals;
   protected final ModelCompletenessRequirements _modelCompletenessRequirements;
   protected final boolean _dryRun;
@@ -207,6 +210,11 @@ protected abstract OptimizerResult workWithClusterModel()
    * Perform the memory clean up after {@link #computeResult()}.
    */
   protected void finish() {
+    if (_operationProgress != null) {
+      long totalTimeMs = _operationProgress.getCurrentTotalExecutionTimeMs();
+      LOG.info("Operation {} finished with uuid {}; total time: {}ms; steps: {}",
+          _operationProgress.getOperation(), _uuid, totalTimeMs, _operationProgress);
+    }
     _operationProgress = null;
     _combinedCompletenessRequirements = null;
     if (_goalsByPriority != null) {