improve messages when resource utilization exceed (#18993)

Author: J-HowHuang

pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java

@@ -77,7 +77,7 @@ public RealtimeSegmentValidationManager(ControllerConf config, PinotHelixResourc
       ValidationMetrics validationMetrics, ControllerMetrics controllerMetrics, StorageQuotaChecker quotaChecker,
       ResourceUtilizationManager resourceUtilizationManager) {
     super("RealtimeSegmentValidationManager", config.getRealtimeSegmentValidationFrequencyInSeconds(),
-            config.getRealtimeSegmentValidationManagerInitialDelaySeconds(),
+        config.getRealtimeSegmentValidationManagerInitialDelaySeconds(),
         config.getRealtimeSegmentValidationCronExpression(),
         pinotHelixResourceManager, leadControllerManager, controllerMetrics);
     _llcRealtimeSegmentManager = llcRealtimeSegmentManager;
@@ -174,7 +174,9 @@ boolean shouldEnsureConsuming(String tableNameWithType) {
       if (!isTablePaused || !pauseStatus.getReasonCode()
           .equals(PauseState.ReasonCode.RESOURCE_UTILIZATION_LIMIT_EXCEEDED)) {
         _llcRealtimeSegmentManager.pauseConsumption(tableNameWithType,
-            PauseState.ReasonCode.RESOURCE_UTILIZATION_LIMIT_EXCEEDED, "Resource utilization limit exceeded.");
+            PauseState.ReasonCode.RESOURCE_UTILIZATION_LIMIT_EXCEEDED,
+            "Resource utilization limit exceeded. Check ResourceUtilizationManager log on controllers for detail "
+                + "reasons");
       }
       return false; // if resource utilization check failed, then skip subsequent checks
     } else if ((isResourceUtilizationWithinLimits == UtilizationChecker.CheckResult.PASS) && isTablePaused

pinot-controller/src/main/java/org/apache/pinot/controller/validation/ResourceUtilizationManager.java

@@ -54,22 +54,28 @@ public UtilizationChecker.CheckResult isResourceUtilizationWithinLimits(String t
     if (StringUtils.isEmpty(tableNameWithType)) {
       throw new IllegalArgumentException("Table name found to be null or empty while checking resource utilization.");
     }
-    LOGGER.info("Checking resource utilization for table: {}", tableNameWithType);
     UtilizationChecker.CheckResult overallIsResourceUtilizationWithinLimits = UtilizationChecker.CheckResult.PASS;
     for (UtilizationChecker utilizationChecker : _utilizationCheckers) {
       UtilizationChecker.CheckResult isResourceUtilizationWithinLimits =
           utilizationChecker.isResourceUtilizationWithinLimits(tableNameWithType, purpose);
-      LOGGER.info("For utilization checker: {}, isResourceUtilizationWithinLimits: {}, purpose: {}",
-          utilizationChecker.getName(), isResourceUtilizationWithinLimits, purpose);
       if (isResourceUtilizationWithinLimits == UtilizationChecker.CheckResult.FAIL) {
         // If any UtilizationChecker returns FAIL, we should mark the overall as FAIL. FAIL should always have
         // priority over other results
+        LOGGER.error(
+            "The utilization checker: {} failed on table {}, isResourceUtilizationWithinLimits: {}, purpose: {}",
+            utilizationChecker.getName(), tableNameWithType, isResourceUtilizationWithinLimits, purpose);
         overallIsResourceUtilizationWithinLimits = UtilizationChecker.CheckResult.FAIL;
-      } else if ((overallIsResourceUtilizationWithinLimits == UtilizationChecker.CheckResult.PASS)
-          && (isResourceUtilizationWithinLimits == UtilizationChecker.CheckResult.UNDETERMINED)) {
+      } else if (isResourceUtilizationWithinLimits == UtilizationChecker.CheckResult.UNDETERMINED) {
         // If we haven't already updated the overall to a value other than PASS, and we get an UNDETERMINED result,
         // update the overall to UNDETERMINED. Should not update to UNDETERMINED if we have set the overall to FAIL
-        overallIsResourceUtilizationWithinLimits = UtilizationChecker.CheckResult.UNDETERMINED;
+        if (overallIsResourceUtilizationWithinLimits == UtilizationChecker.CheckResult.PASS) {
+          overallIsResourceUtilizationWithinLimits = UtilizationChecker.CheckResult.UNDETERMINED;
+        }
+        LOGGER.warn("The utilization checker: {} couldn't determine for table {}, purpose: {}",
+            utilizationChecker.getName(), tableNameWithType, purpose);
+      } else {
+        LOGGER.info("For utilization checker: {} pass on table {}, purpose: {}", utilizationChecker.getName(),
+            tableNameWithType, purpose);
       }
     }
     return overallIsResourceUtilizationWithinLimits;