[SPARK-49404] Adjust ERROR-level log messages

### What changes were proposed in this pull request?

This PR aims to audit and adjust `ERROR`-level log messages.

### Why are the changes needed?

To be consistent on the log-level and avoid misleading error messages.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Manual review.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#111 from dongjoon-hyun/SPARK-49404.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java

@@ -100,9 +100,10 @@ private boolean checkInformersHealth(RuntimeInfo operatorRuntimeInfo) {
           } else {
             if (log.isErrorEnabled()) {
               log.error(
-                  "Controller: {}, Event Source: {}, Informer: {} is not in a healthy state",
+                  "Controller: {}, Event Source: {}, Informer: {} is in {}, not a healthy state",
                   controllerEntry.getKey(),
                   eventSourceEntry.getKey(),
+                  informerEntry.getValue().getStatus(),
                   informerEntry.getKey());
             }
             informersHealthList.add(false);

spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java

@@ -93,7 +93,7 @@ public UpdateControl<SparkApplication> reconcile(
           sparkApplication, context.getClient())) {
         return UpdateControl.noUpdate();
       }
-      log.debug("Start reconciliation.");
+      log.debug("Start application reconciliation.");
       sparkAppStatusRecorder.updateStatusFromCache(sparkApplication);
       SparkAppContext ctx = new SparkAppContext(sparkApplication, context, submissionWorker);
       List<AppReconcileStep> reconcileSteps = getReconcileSteps(sparkApplication);

spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java

@@ -113,7 +113,7 @@ private static void deleteLocalFileFromPathKey(
         log.warn("Local temp file not found at {}", pathKey);
       }
     } catch (Throwable t) {
-      log.error("Failed to delete temp file. Attempting delete upon exit.", t);
+      log.warn("Failed to delete temp file. Attempting delete upon exit.", t);
     } finally {
       if (!deleted && localFile.isPresent() && localFile.get().exists()) {
         localFile.get().deleteOnExit();

spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java

@@ -78,7 +78,7 @@ public UpdateControl<SparkCluster> reconcile(
       if (sentinelManager.handleSentinelResourceReconciliation(sparkCluster, context.getClient())) {
         return UpdateControl.noUpdate();
       }
-      log.error("Start SparkClusterReconciler.");
+      log.debug("Start cluster reconciliation.");
       sparkClusterStatusRecorder.updateStatusFromCache(sparkCluster);
       SparkClusterContext ctx = new SparkClusterContext(sparkCluster, context, submissionWorker);
       List<ClusterReconcileStep> reconcileSteps = getReconcileSteps(sparkCluster);
@@ -90,7 +90,7 @@ public UpdateControl<SparkCluster> reconcile(
       }
       return ReconcilerUtils.toUpdateControl(sparkCluster, completeAndDefaultRequeue());
     } finally {
-      log.error("Reconciliation completed.");
+      log.debug("Reconciliation completed.");
       trackedMDC.reset();
     }
   }

spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java

@@ -97,13 +97,14 @@ private void patchAndStatusWithVersionLocked(CR resource, KubernetesClient clien
         resource.getMetadata().setResourceVersion(updated.getMetadata().getResourceVersion());
         err = null;
       } catch (KubernetesClientException e) {
-        log.error("Error while patching status, retrying {}/{}...", i + 1, maxRetry, e);
+        log.warn("Error while patching status, retrying {}/{}...", i + 1, maxRetry, e);
         Thread.sleep(TimeUnit.SECONDS.toMillis(API_RETRY_ATTEMPT_AFTER_SECONDS.getValue()));
         err = e;
       }
     }
 
     if (err != null) {
+      log.error("Fail to patch status.", err);
       throw err;
     }