HIVE-30033: Hive K8s operator: log replica changes

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/dependent/HiveDependentResource.java

@@ -54,6 +54,7 @@
 import org.apache.hive.kubernetes.operator.model.spec.SecretKeyRef;
 import org.apache.hive.kubernetes.operator.model.spec.ProbeSpec;
 import org.apache.hive.kubernetes.operator.util.ConfigUtils;
+import org.apache.hive.kubernetes.operator.util.Workloads;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -164,6 +165,15 @@ protected R handleCreate(R desired, P primary, Context<P> context) {
    */
   protected Integer resolveReplicaCount(P primary, Context<P> context,
       AutoscalingSpec autoscaling, int staticReplicas, int initialReplicas) {
+    Optional<R> existing = getSecondaryResource(primary, context);
+    Integer resolved = computeReplicaCount(primary, existing, autoscaling,
+        staticReplicas, initialReplicas);
+    logReplicaChange(primary, existing, resolved);
+    return resolved;
+  }
+
+  private Integer computeReplicaCount(P primary, Optional<R> existing,
+      AutoscalingSpec autoscaling, int staticReplicas, int initialReplicas) {
     // Suspended cluster → 0 replicas (dependent resources natively respect suspend).
     // Exception: HMS stays running if includeMetastore=false in autoSuspend config.
     if (primary instanceof HiveCluster hc && hc.getSpec().suspend()) {
@@ -175,7 +185,6 @@ protected Integer resolveReplicaCount(P primary, Context<P> context,
     if (autoscaling == null || !autoscaling.isEnabled()) {
       return staticReplicas;
     }
-    Optional<R> existing = getSecondaryResource(primary, context);
     if (existing.isPresent()) {
       // Check if the autoscaler has made a decision during this operator's lifecycle
       Integer managed = HiveClusterAutoscaler.getManagedReplicas(
@@ -186,21 +195,37 @@ protected Integer resolveReplicaCount(P primary, Context<P> context,
         return managed;
       }
       // Fallback: operator restarted and MANAGED_REPLICAS is empty — read current value
-      R resource = existing.get();
-      if (resource instanceof io.fabric8.kubernetes.api.model.apps.Deployment d) {
-        return d.getSpec() != null && d.getSpec().getReplicas() != null
-            ? d.getSpec().getReplicas() : initialReplicas;
-      }
-      if (resource instanceof io.fabric8.kubernetes.api.model.apps.StatefulSet s) {
-        return s.getSpec() != null && s.getSpec().getReplicas() != null
-            ? s.getSpec().getReplicas() : initialReplicas;
-      }
-      return initialReplicas;
+      Integer current = Workloads.replicas(existing.get());
+      return current != null ? current : initialReplicas;
     }
     // First creation: start at minReplicas.
     return initialReplicas;
   }
 
+  /**
+   * Emits an INFO line when the SSA about to run will actually change the workload's replica
+   * count. Silence means the count already matches, so no scale is happening. Without this,
+   * every scale of an HS2/Metastore Deployment reached the cluster silently (only the imperative
+   * LLAP path and the autoscaler logged); a bare "Reconciled" line said nothing about the size.
+   * Uses the same "Scaling ... A -> B" shape the imperative LLAP path emits.
+   */
+  private void logReplicaChange(P primary, Optional<R> existing, Integer desired) {
+    String component = getComponentName();
+    if (component == null || desired == null) {
+      return;
+    }
+    String ns = primary.getMetadata().getNamespace();
+    String name = existing.map(r -> r.getMetadata().getName()).orElse(component);
+    if (existing.isEmpty()) {
+      LOG.info("Creating {} {}/{} with {} replicas", component, ns, name, desired);
+      return;
+    }
+    Integer current = Workloads.replicas(existing.get());
+    if (current != null && !current.equals(desired)) {
+      LOG.info("Scaling {} {}/{}: {} -> {} replicas", component, ns, name, current, desired);
+    }
+  }
+
 
   /**
    * Returns the component name for this dependent (used for autoscaler replica lookup).

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/reconciler/HiveClusterReconciler.java

@@ -59,6 +59,7 @@
 import org.apache.hive.kubernetes.operator.model.status.ComponentStatus;
 import org.apache.hive.kubernetes.operator.util.ConfigUtils;
 import org.apache.hive.kubernetes.operator.util.Labels;
+import org.apache.hive.kubernetes.operator.util.Workloads;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -561,7 +562,7 @@
       workloadName = resource.getMetadata().getName() + "-" + llapName;
     } else if (component.startsWith(ConfigUtils.COMPONENT_TEZAM + "-")) {
       String llapName = component.substring(ConfigUtils.COMPONENT_TEZAM.length() + 1);
       workloadName = resource.getMetadata().getName() + "-tezam-" + llapName;
     } else {
       workloadName = resource.getMetadata().getName() + "-" + component;
     }
@@ -577,6 +578,30 @@
     }
   }
 
+  /**
+   * Emits an INFO log when the reconciler's server-side apply is about to change the workload's
+   * replica count. Without this, an SSA-driven scale (a user editing spec.llapClusters[i].replicas,
+   * a helm upgrade rewriting it) reaches the StatefulSet/Deployment silently -- only the
+   * autoscaler path {@link #patchReplicas} logged its scales, so a plain scale looked like the
+   * operator was doing nothing. Read failures are swallowed at DEBUG: the SSA below runs either
+   * way, and a missing pre-scale line is not worth failing the reconcile over.
+   */
+  private void logReplicaChange(KubernetesClient client, String ns, String workloadName,
+      String kind, int desired, boolean isStatefulSet) {
+    try {
+      Integer current = Workloads.replicas(isStatefulSet
+          ? client.apps().statefulSets().inNamespace(ns).withName(workloadName).get()
+          : client.apps().deployments().inNamespace(ns).withName(workloadName).get());
+      if (current == null) {
+        LOG.info("Creating {} {}/{} with {} replicas", kind, ns, workloadName, desired);
+      } else if (current != desired) {
+        LOG.info("Scaling {} {}/{}: {} -> {} replicas", kind, ns, workloadName, current, desired);
+      }
+    } catch (Exception e) {
+      LOG.debug("Could not read current replicas for {}/{}: {}", ns, workloadName, e.getMessage());
+    }
+  }
+
   private void patchSuspendSpec(KubernetesClient client, HiveCluster resource, boolean suspend) {
     String ns = resource.getMetadata().getNamespace();
     String name = resource.getMetadata().getName();
@@ -627,6 +652,8 @@
       // brief scale-up-then-down on first create (K8s defaults to 1 if omitted).
       // resolveLlapReplicaCount already reads the autoscaler's managed value,
       // so this is always the correct replica count.
+      String llapWorkload = clusterName + "-" + llapSpec.name();
+      logReplicaChange(client, ns, llapWorkload, "llap", replicas, /*isStatefulSet=*/true);
       client.apps().statefulSets().inNamespace(ns)
           .resource(LlapResourceBuilder.buildStatefulSet(resource, llapSpec, replicas))
           .forceConflicts()
@@ -646,6 +673,8 @@
         client.services().inNamespace(ns)
             .resource(LlapResourceBuilder.buildTezAmService(resource, llapSpec))
             .serverSideApply();
+        String tezAmWorkload = clusterName + "-tezam-" + llapSpec.name();
+        logReplicaChange(client, ns, tezAmWorkload, "tezam", tezAmReplicas, /*isStatefulSet=*/false);
         client.apps().deployments().inNamespace(ns)
             .resource(LlapResourceBuilder.buildTezAmDeployment(resource, llapSpec, tezAmReplicas))
             .forceConflicts()

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/util/Workloads.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hive.kubernetes.operator.util;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+
+/**
+ * Small helpers over Deployment/StatefulSet that read fields without the
+ * null-guard boilerplate every caller would otherwise repeat.
+ */
+public final class Workloads {
+
+  private Workloads() {}
+
+  /**
+   * Returns spec.replicas from a Deployment or StatefulSet, or null when the resource is
+   * absent, has no spec, or the field is unset. A non-workload resource returns null too.
+   */
+  public static Integer replicas(HasMetadata resource) {
+    if (resource instanceof Deployment d) {
+      return d.getSpec() == null ? null : d.getSpec().getReplicas();
+    }
+    if (resource instanceof StatefulSet s) {
+      return s.getSpec() == null ? null : s.getSpec().getReplicas();
+    }
+    return null;
+  }
+}