HIVE-29679: Update Tez AM K8s Operator Auto-Scaling to scale down idle AMs (#6561)

Author: tanishq-chugh

llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java

@@ -1080,6 +1080,9 @@ public void dagComplete() {
     writeLock.lock();
     try {
       dagRunning = false;
+      if (metrics != null) {
+        metrics.setDagRunning(false);
+      }
       dagStats = new StatsPerDag();
       int pendingCount = 0;
       for (Entry<Priority, List<TaskInfo>> entry : pendingTasks.entrySet()) {
@@ -1173,6 +1176,9 @@ public void allocateTask(Object task, Resource capability, String[] hosts, Strin
         metrics.setDagId(id.getDAGID().toString());
       }
       dagRunning = true;
+      if (metrics != null) {
+        metrics.setDagRunning(true);
+      }
     }
     dagStats.registerTaskRequest(hosts, racks);
     addPendingTask(taskInfo);
@@ -1194,6 +1200,9 @@ public void allocateTask(Object task, Resource capability, ContainerId container
         metrics.setDagId(id.getDAGID().toString());
       }
       dagRunning = true;
+      if (metrics != null) {
+        metrics.setDagRunning(true);
+      }
     }
     dagStats.registerTaskRequest(null, null);
     addPendingTask(taskInfo);

llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerInfo.java

@@ -38,7 +38,8 @@ public enum LlapTaskSchedulerInfo implements MetricsInfo {
   SchedulerRunningTaskCount("Total number of running tasks"),
   SchedulerPendingPreemptionTaskCount("Total number of tasks pending for pre-emption"),
   SchedulerPreemptedTaskCount("Total number of tasks pre-empted"),
-  SchedulerCompletedDagCount("Number of DAGs completed");
+  SchedulerCompletedDagCount("Number of DAGs completed"),
+  SchedulerDagStatus("Current AM operational DAG status: 0 for IDLE, 1 for RUNNING");
 
   private final String desc;
 

llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerMetrics.java

@@ -25,6 +25,7 @@
 import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerPendingPreemptionTaskCount;
 import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerPendingTaskCount;
 import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerPreemptedTaskCount;
+import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerDagStatus;
 import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerRunningTaskCount;
 import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerSchedulableTaskCount;
 import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerSuccessfulTaskCount;
@@ -84,6 +85,8 @@ public class LlapTaskSchedulerMetrics implements MetricsSource {
   @Metric
   MutableCounterInt completedDagcount;
   @Metric
+  MutableGaugeInt dagRunning;
+  @Metric
   MutableCounterInt pendingPreemptionTasksCount;
   @Metric
   MutableGaugeInt wmUnusedGuaranteedCount;
@@ -276,6 +279,7 @@ private void getTaskSchedulerStats(MetricsRecordBuilder rb) {
         .addGauge(SchedulerMemoryPerInstance, memoryPerInstance.value())
         .addGauge(SchedulerCpuCoresPerInstance, cpuCoresPerInstance.value())
         .addGauge(SchedulerDisabledNodeCount, disabledNodeCount.value())
+        .addGauge(SchedulerDagStatus, dagRunning.value())
         .addCounter(SchedulerPendingTaskCount, pendingTasksCount.value())
         .addCounter(SchedulerSchedulableTaskCount, schedulableTasksCount.value())
         .addCounter(SchedulerRunningTaskCount, runningTasksCount.value())
@@ -285,6 +289,10 @@ private void getTaskSchedulerStats(MetricsRecordBuilder rb) {
         .addCounter(SchedulerCompletedDagCount, completedDagcount.value());
   }
 
+  public void setDagRunning(boolean running) {
+    dagRunning.set(running ? 1 : 0);
+  }
+
   public JvmMetrics getJvmMetrics() {
     return jvmMetrics;
   }

packaging/src/kubernetes/README.md

@@ -520,7 +520,7 @@ HS2 routes sessions to clusters server-side based on admin-defined user/group ru
 
 Each LLAP cluster is fully isolated:
 - **Separate LLAP daemon StatefulSet** with independent executor count, memory, and replicas
-- **Separate TezAM StatefulSet** (one per LLAP cluster) with its own ZooKeeper registration
+- **Separate TezAM Deployment** (one per LLAP cluster) with its own ZooKeeper registration
 - **Separate autoscaling** — each cluster scales independently based on its own metrics
 - **Shared scratch PVC** (ReadWriteMany) for HS2 ↔ TezAM coordination files
 
@@ -618,11 +618,11 @@ For the above configuration, the operator creates:
 | Resource | Name | Purpose |
 |----------|------|---------|
 | StatefulSet | `hive-production` | LLAP daemons for production cluster |
-| StatefulSet | `hive-tezam-production` | TezAM for production cluster |
+| Deployment | `hive-tezam-production` | TezAM for production cluster |
 | StatefulSet | `hive-analytics` | LLAP daemons for analytics cluster |
-| StatefulSet | `hive-tezam-analytics` | TezAM for analytics cluster |
+| Deployment | `hive-tezam-analytics` | TezAM for analytics cluster |
 | StatefulSet | `hive-dev` | LLAP daemons for dev cluster |
-| StatefulSet | `hive-tezam-dev` | TezAM for dev cluster |
+| Deployment | `hive-tezam-dev` | TezAM for dev cluster |
 | Service (headless) | `hive-production`, `hive-analytics`, `hive-dev` | LLAP daemon discovery |
 | Service (headless) | `hive-tezam-production`, `hive-tezam-analytics`, `hive-tezam-dev` | TezAM discovery |
 | ConfigMap | `hive-production-config`, etc. | `llap-daemon-site.xml` per cluster |
@@ -1391,7 +1391,7 @@ setup is needed — simply connect to HS2 and the operator wakes LLAP/TezAM as n
 
 LLAP is configured as an array (`llapClusters`) to support multi-tenant deployments with
 independent scaling. Each entry creates a separate LLAP StatefulSet, Service, ConfigMap,
-and a paired TezAM StatefulSet (when `tezAm.enabled: true`).
+and a paired TezAM Deployment (when `tezAm.enabled: true`).
 
 | Value | Default | Description |
 |-------|---------|-------------|
@@ -1419,7 +1419,7 @@ Clients connect with just their identity — no cluster-specific JDBC URL params
 
 ### Tez AM
 
-TezAM is deployed as one StatefulSet per LLAP cluster. The global `tezAm` section
+TezAM is deployed as one Deployment per LLAP cluster. The global `tezAm` section
 controls shared settings (enabled flag, scratch PVC). Per-LLAP TezAM settings
 (replicas, autoscaling) can be overridden in each `llapClusters[].tezAm` entry.
 
@@ -1587,14 +1587,14 @@ HiveClusterReconciler
   |
   +-- [Imperative] Per-LLAP-Cluster Resources (for each llapClusters[] entry):
         +-- LLAP StatefulSet + headless Service + ConfigMap + PDB
-        +-- TezAM StatefulSet + headless Service + ConfigMap (one TezAM per LLAP cluster)
+        +-- TezAM Deployment + headless Service + ConfigMap (one TezAM per LLAP cluster)
 ```
 
 LLAP clusters and their paired TezAM instances are managed imperatively by the reconciler
 (not via JOSDK workflow dependents) because the number of clusters is dynamic — determined
 at runtime from the CR spec. Each `llapClusters[]` entry produces:
 - **LLAP**: StatefulSet (`{cluster}-{name}`), headless Service, ConfigMap (`llap-daemon-site.xml`), PDB
-- **TezAM**: StatefulSet (`{cluster}-tezam-{name}`), headless Service, ConfigMap (`tez-site.xml`)
+- **TezAM**: Deployment (`{cluster}-tezam-{name}`), headless Service, ConfigMap (`tez-site.xml`)
 
 All imperative resources are applied via `serverSideApply()`. Removed LLAP clusters (and
 their TezAMs) are garbage-collected automatically using label-based discovery.

packaging/src/kubernetes/helm/hive-operator/Chart.yaml

@@ -19,7 +19,7 @@ description: Apache Hive Kubernetes Operator - deploys and manages Hive clusters
 type: application
 version: "4.3.0-SNAPSHOT"
 appVersion: "4.3.0-SNAPSHOT"
-kubeVersion: ">=1.25.0"
+kubeVersion: ">=1.25.0-0"
 keywords:
   - hive
   - hadoop

packaging/src/kubernetes/helm/hive-operator/crds/hiveclusters.hive.apache.org-v1.yml

@@ -692,6 +692,10 @@ spec:
                     type: string
                 type: object
                 x-kubernetes-preserve-unknown-fields: true
+              serviceAccountName:
+                description: "Kubernetes ServiceAccount name for all component pods.\
+                  \ If not specified, pods use the namespace default service account."
+                type: string
               suspend:
                 description: "When true, the cluster is immediately suspended (all\
                   \ components scaled to 0). Set to false to wake a suspended cluster."

packaging/src/kubernetes/helm/hive-operator/templates/clusterrole.yaml

@@ -58,3 +58,7 @@ rules:
   - apiGroups: ["policy"]
     resources: ["poddisruptionbudgets"]
     verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
+  # EndpointSlices: operator manages a custom per-pod-hostname slice for TezAM DNS
+  - apiGroups: ["discovery.k8s.io"]
+    resources: ["endpointslices"]
+    verbs: ["get", "list", "create", "patch", "delete"]

packaging/src/kubernetes/helm/hive-operator/templates/hivecluster.yaml

@@ -26,6 +26,9 @@ metadata:
 spec:
   image: {{ .Values.cluster.image }}
   imagePullPolicy: {{ .Values.cluster.imagePullPolicy }}
+  {{- if .Values.cluster.serviceAccountName }}
+  serviceAccountName: {{ .Values.cluster.serviceAccountName }}
+  {{- end }}
 
   metastore:
     enabled: {{ .Values.cluster.metastore.enabled }}
@@ -178,7 +181,6 @@ spec:
   tezAm:
     enabled: {{ .Values.cluster.tezAm.enabled }}
     {{- if .Values.cluster.tezAm.enabled }}
-    replicas: {{ .Values.cluster.tezAm.replicas }}
     scratchStorageSize: {{ .Values.cluster.tezAm.scratchStorageSize | quote }}
     {{- if .Values.cluster.tezAm.scratchStorageClassName }}
     scratchStorageClassName: {{ .Values.cluster.tezAm.scratchStorageClassName | quote }}

packaging/src/kubernetes/helm/hive-operator/values.yaml

@@ -47,6 +47,10 @@ cluster:
   image: "apache/hive:4.3.0-SNAPSHOT"
   imagePullPolicy: IfNotPresent
 
+  # ServiceAccount name for all component pods (HS2, Metastore, LLAP, TezAM, schema-init).
+  # If empty, pods use the namespace default service account.
+  serviceAccountName: ""
+
   # ---------------------------------------------------------------------------
   # DATABASE (Required) — RDBMS for the Hive Metastore backend
   # ---------------------------------------------------------------------------
@@ -207,7 +211,6 @@ cluster:
   # ---------------------------------------------------------------------------
   tezAm:
     enabled: true
-    replicas: 2
     scratchStorageSize: "1Gi"
     scratchStorageClassName: ""
     resources: {}

packaging/src/kubernetes/pom.xml

@@ -52,10 +52,6 @@
       <artifactId>kubernetes-httpclient-vertx</artifactId>
       <version>${fabric8.version}</version>
     </dependency>
-    <dependency>
-      <groupId>org.apache.commons</groupId>
-      <artifactId>commons-lang3</artifactId>
-    </dependency>
     <dependency>
       <groupId>io.github.java-diff-utils</groupId>
       <artifactId>java-diff-utils</artifactId>
@@ -73,6 +69,11 @@
       <version>${fabric8.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.curator</groupId>
+      <artifactId>curator-framework</artifactId>
+      <version>${curator.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>