Fix negative reload estimated time and year 2088 rebalance timestamp (#17833)

Two bugs are fixed:

1. Reload status: estimatedTimeRemainingInMinutes goes negative when
   successCount exceeds totalSegmentCount (due to segments being
   added/replaced after the reload job started). Fixed by clamping
   remainingSegments to Math.max(0, ...). Also added a derived "status"
   field (COMPLETED / COMPLETED_WITH_ERRORS / IN_PROGRESS) to the reload
   status response so consumers don't have to infer completion from raw
   counts.

2. Rebalance: timeToFinishInSeconds becomes a Unix timestamp (~1.7B)
   instead of a duration when startTimeMs is never initialized (stays at
   default 0). This happens for NO_OP or early-failure rebalances that
   skip START_TRIGGER. The UI then computes
   submissionTimeMs + (timestamp * 1000) ≈ year 2088.
   Fixed by guarding against startTimeMs <= 0 and returning 0 seconds.

Made-with: Cursor

Author: jayeshchoudhary

pinot-controller/src/main/java/org/apache/pinot/controller/api/dto/PinotTableReloadStatusResponse.java

@@ -25,6 +25,7 @@
 
 @InterfaceStability.Evolving
 public class PinotTableReloadStatusResponse {
+  private String _status;
   private double _timeElapsedInMinutes;
   private double _estimatedTimeRemainingInMinutes;
   private int _totalSegmentCount;
@@ -35,6 +36,15 @@ public class PinotTableReloadStatusResponse {
   private PinotControllerJobMetadataDto _metadata;
   private List<SegmentReloadFailureResponse> _segmentReloadFailures;
 
+  public String getStatus() {
+    return _status;
+  }
+
+  public PinotTableReloadStatusResponse setStatus(String status) {
+    _status = status;
+    return this;
+  }
+
   public int getTotalSegmentCount() {
     return _totalSegmentCount;
   }

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java

@@ -196,9 +196,8 @@ private void updateOnStart(Map<String, Map<String, String>> currentState,
   @Override
   public void onNoop(String msg) {
     _controllerMetrics.setValueOfTableGauge(_tableNameWithType, ControllerGauge.TABLE_REBALANCE_IN_PROGRESS, 0);
-    long timeToFinishInSeconds = (System.currentTimeMillis() - _tableRebalanceProgressStats.getStartTimeMs()) / 1000L;
     _tableRebalanceProgressStats.setCompletionStatusMsg(msg);
-    _tableRebalanceProgressStats.setTimeToFinishInSeconds(timeToFinishInSeconds);
+    _tableRebalanceProgressStats.setTimeToFinishInSeconds(computeElapsedTimeInSeconds());
     _tableRebalanceProgressStats.setStatus(RebalanceResult.Status.NO_OP);
     trackStatsInZk();
   }
@@ -208,9 +207,8 @@ public void onSuccess(String msg) {
     Preconditions.checkState(RebalanceResult.Status.DONE != _tableRebalanceProgressStats.getStatus(),
         "Table Rebalance already completed");
     _controllerMetrics.setValueOfTableGauge(_tableNameWithType, ControllerGauge.TABLE_REBALANCE_IN_PROGRESS, 0);
-    long timeToFinishInSeconds = (System.currentTimeMillis() - _tableRebalanceProgressStats.getStartTimeMs()) / 1000L;
     _tableRebalanceProgressStats.setCompletionStatusMsg(msg);
-    _tableRebalanceProgressStats.setTimeToFinishInSeconds(timeToFinishInSeconds);
+    _tableRebalanceProgressStats.setTimeToFinishInSeconds(computeElapsedTimeInSeconds());
     _tableRebalanceProgressStats.setStatus(RebalanceResult.Status.DONE);
     // Zero out the in_progress convergence stats
     TableRebalanceProgressStats.RebalanceStateStats stats = new TableRebalanceProgressStats.RebalanceStateStats();
@@ -226,13 +224,25 @@ public void onSuccess(String msg) {
   @Override
   public void onError(String errorMsg) {
     _controllerMetrics.setValueOfTableGauge(_tableNameWithType, ControllerGauge.TABLE_REBALANCE_IN_PROGRESS, 0);
-    long timeToFinishInSeconds = (System.currentTimeMillis() - _tableRebalanceProgressStats.getStartTimeMs()) / 1000L;
-    _tableRebalanceProgressStats.setTimeToFinishInSeconds(timeToFinishInSeconds);
+    _tableRebalanceProgressStats.setTimeToFinishInSeconds(computeElapsedTimeInSeconds());
     _tableRebalanceProgressStats.setStatus(RebalanceResult.Status.FAILED);
     _tableRebalanceProgressStats.setCompletionStatusMsg(errorMsg);
     trackStatsInZk();
   }
 
+  /**
+   * Safely computes elapsed time in seconds since rebalance started.
+   * Returns 0 if startTimeMs was never set (i.e. still at default 0), which happens
+   * when the rebalance completes as NO_OP or fails before START_TRIGGER fires.
+   */
+  private long computeElapsedTimeInSeconds() {
+    long startTimeMs = _tableRebalanceProgressStats.getStartTimeMs();
+    if (startTimeMs <= 0) {
+      return 0L;
+    }
+    return (System.currentTimeMillis() - startTimeMs) / 1000L;
+  }
+
   @Override
   public void onRollback() {
     _tableRebalanceProgressStats.setRebalanceProgressStatsCurrentStep(

pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadStatusReporter.java

@@ -70,7 +70,8 @@ public PinotTableReloadStatusReporter(PinotHelixResourceManager pinotHelixResour
 
   private static double computeEstimatedRemainingTimeInMinutes(PinotTableReloadStatusResponse finalResponse,
       double timeElapsedInMinutes) {
-    int remainingSegments = finalResponse.getTotalSegmentCount() - finalResponse.getSuccessCount();
+    // Clamp to 0 to handle cases where successCount > totalSegmentCount (e.g. segments added after job started)
+    int remainingSegments = Math.max(0, finalResponse.getTotalSegmentCount() - finalResponse.getSuccessCount());
 
     double estimatedRemainingTimeInMinutes = -1;
     if (finalResponse.getSuccessCount() > 0) {
@@ -80,6 +81,26 @@ private static double computeEstimatedRemainingTimeInMinutes(PinotTableReloadSta
     return estimatedRemainingTimeInMinutes;
   }
 
+  /**
+   * Derives the overall reload job status from aggregated counts.
+   * - COMPLETED: all segments reloaded successfully, no server call failures
+   * - COMPLETED_WITH_ERRORS: reload finished but some segments failed
+   * - IN_PROGRESS: reload is still running
+   */
+  private static String deriveReloadStatus(PinotTableReloadStatusResponse response) {
+    int processed = response.getSuccessCount() + (response.getFailureCount() != null
+        ? response.getFailureCount().intValue() : 0);
+    boolean allProcessed = processed >= response.getTotalSegmentCount();
+
+    if (allProcessed && response.getTotalServerCallsFailed() == 0) {
+      if (response.getFailureCount() != null && response.getFailureCount() > 0) {
+        return "COMPLETED_WITH_ERRORS";
+      }
+      return "COMPLETED";
+    }
+    return "IN_PROGRESS";
+  }
+
   private static double computeTimeElapsedInMinutes(double submissionTime) {
     return ((double) System.currentTimeMillis() - submissionTime) / (1000.0 * 60.0);
   }
@@ -210,7 +231,8 @@ public PinotTableReloadStatusResponse getReloadJobStatus(String reloadJobId)
 
     return response.setMetadata(reloadJobMetadata)
         .setTimeElapsedInMinutes(timeElapsedInMinutes)
-        .setEstimatedTimeRemainingInMinutes(estimatedRemainingTimeInMinutes);
+        .setEstimatedTimeRemainingInMinutes(estimatedRemainingTimeInMinutes)
+        .setStatus(deriveReloadStatus(response));
   }
 
   private PinotControllerJobMetadataDto getControllerJobMetadataFromZk(String reloadJobId) {