You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ab...@apache.org on 2025/05/19 08:01:48 UTC
(hive) branch master updated: HIVE-28789: Session-level metrics from Tez sessions (#5813)
This is an automated email from the ASF dual-hosted git repository.
abstractdog pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new feccb31a7ef HIVE-28789: Session-level metrics from Tez sessions (#5813)
feccb31a7ef is described below
commit feccb31a7ef1ebada17cdea23a97a205818aae9a
Author: Bodor Laszlo <bo...@gmail.com>
AuthorDate: Mon May 19 10:01:42 2025 +0200
HIVE-28789: Session-level metrics from Tez sessions (#5813)
---
.../java/org/apache/hadoop/hive/conf/HiveConf.java | 4 +
.../hive/ql/exec/tez/TezSessionPoolManager.java | 11 +-
.../ql/exec/tez/TezSessionPoolManagerMetrics.java | 260 +++++++++++++++++++++
.../hadoop/hive/ql/exec/tez/TezSessionState.java | 34 +++
.../apache/hadoop/hive/ql/exec/tez/TezTask.java | 2 +-
.../hive/ql/exec/tez/monitoring/TezJobMonitor.java | 8 +-
.../exec/tez/TestTezSessionPoolManagerMetrics.java | 110 +++++++++
7 files changed, 425 insertions(+), 4 deletions(-)
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index a7352a826be..492637e7c94 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -4093,6 +4093,10 @@ public static enum ConfVars {
"This flag is used in HiveServer2 to enable a user to use HiveServer2 without\n" +
"turning on Tez for HiveServer2. The user could potentially want to run queries\n" +
"over Tez without the pool of sessions."),
+ HIVE_SERVER2_TEZ_SESSIONS_METRICS_COLLECTION_INTERVAL(
+ "hive.server2.tez.sessions.metrics.collection.interval", "10s",
+ new TimeValidator(TimeUnit.SECONDS),
+ "Interval for collecting metrics from Tez sessions."),
HIVE_SERVER2_TEZ_QUEUE_ACCESS_CHECK("hive.server2.tez.queue.access.check", false,
"Whether to check user access to explicitly specified YARN queues. " +
"yarn.resourcemanager.webapp.address must be configured to use this."),
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java
index c87c968b5c8..844eebdbe1b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java
@@ -89,6 +89,8 @@ private enum CustomQueueAllowed {
private TriggerValidatorRunnable triggerValidatorRunnable;
private YarnQueueHelper yarnQueueChecker;
+ private TezSessionPoolManagerMetrics metrics = null;
+
/** Note: this is not thread-safe. */
public static TezSessionPoolManager getInstance() {
TezSessionPoolManager local = instance;
@@ -100,6 +102,7 @@ public static TezSessionPoolManager getInstance() {
}
protected TezSessionPoolManager() {
+ metrics = new TezSessionPoolManagerMetrics(this);
}
public void startPool(HiveConf conf, final WMFullResourcePlan resourcePlan) throws Exception {
@@ -118,6 +121,7 @@ public void startPool(HiveConf conf, final WMFullResourcePlan resourcePlan) thro
LOG.info("Updated tez session pool manager with triggers {} from active resource plan: {}",
appliedTriggers, resourcePlan.getPlan().getName());
}
+ metrics.start(conf);
}
public void setupPool(HiveConf conf) throws Exception {
@@ -382,6 +386,7 @@ public void stop() throws Exception {
stopTriggerValidator();
}
+ metrics.stop();
instance = null;
}
@@ -536,7 +541,7 @@ public void registerOpenSession(TezSessionPoolSession session) {
private void updateSessions() {
if (sessionTriggerProvider != null) {
- sessionTriggerProvider.setSessions(new LinkedList<>(openSessions));
+ sessionTriggerProvider.setSessions(getSessions());
}
}
@@ -588,4 +593,8 @@ List<String> getTriggerCounterNames() {
}
return counterNames;
}
+
+ public List<TezSessionState> getSessions() {
+ return new LinkedList<>(openSessions);
+ }
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManagerMetrics.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManagerMetrics.java
new file mode 100644
index 00000000000..6e153b83ea8
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManagerMetrics.java
@@ -0,0 +1,260 @@
+/*
+ * 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.hadoop.hive.ql.exec.tez;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.AtomicDouble;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.hive.common.metrics.common.Metrics;
+import org.apache.hadoop.hive.common.metrics.common.MetricsFactory;
+import org.apache.hadoop.hive.common.metrics.common.MetricsVariable;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+/**
+ * TezSessionPoolManagerMetrics collects, aggregates and exposes data from Tez sessions to Hive's own MetricsFactory.
+ * This is independent of the Tez DagClient which is active only while running DAGs.
+ */
+public class TezSessionPoolManagerMetrics {
+ private static final Logger LOG = LoggerFactory.getLogger(TezSessionPoolManagerMetrics.class);
+
+ // metric names to be used by Tez sessions when returning data
+ public static final String TEZ_SESSION_METRIC_RUNNING_TASKS = "tez_session_running_tasks";
+ public static final String TEZ_SESSION_METRIC_PENDING_TASKS = "tez_session_pending_tasks";
+
+ // derived metrics (not set by tez sessions, calculated here)
+ public static final String TEZ_SESSION_METRIC_TASK_BACKLOG_RATIO = "tez_session_task_backlog_ratio";
+ public static final String TEZ_SESSION_METRIC_TASK_BACKLOG_RATIO_PRESENT_SINCE =
+ "tez_session_task_backlog_ratio_%d_present_since";
+ // For scaling decisions, TezSessionPoolManagerMetrics tracks the duration
+ // since the task backlog ratio exceeded a certain threshold.
+ private static final Integer[] TRACKED_TASK_BACKLOG_RATIOS = {1, 2, 5, 10, 20, 50, 100};
+
+ private final TezSessionPoolManager poolManager;
+ private final Metrics metrics;
+ private ScheduledExecutorService mainExecutor;
+
+ // session pool metrics
+ @VisibleForTesting
+ final TezSessionMetric<Integer> runningTasksCount = new TezSessionMetric<>(TEZ_SESSION_METRIC_RUNNING_TASKS, 0);
+ @VisibleForTesting
+ final TezSessionMetric<Integer> pendingTasksCount = new TezSessionMetric<>(TEZ_SESSION_METRIC_PENDING_TASKS, 0);
+ @VisibleForTesting
+ final TezSessionMetric<Double> taskBacklogRatio = new TezSessionMetric<>(TEZ_SESSION_METRIC_TASK_BACKLOG_RATIO, 0.0);
+ private final TaskBacklogRatioMetric[] taskBacklogRatioDurations = Arrays.stream(TRACKED_TASK_BACKLOG_RATIOS)
+ .map(
+ ratio -> new TaskBacklogRatioMetric(String.format(TEZ_SESSION_METRIC_TASK_BACKLOG_RATIO_PRESENT_SINCE, ratio),
+ ratio))
+ .toArray(TaskBacklogRatioMetric[]::new);
+
+
+ public TezSessionPoolManagerMetrics(TezSessionPoolManager poolManager) {
+ this.poolManager = poolManager;
+ this.metrics = MetricsFactory.getInstance();
+ }
+
+ private static void waitForTasksToFinish(List<Future<?>> collectTasks, ExecutorService collectorExecutor) {
+ try {
+ CompletableFuture.allOf(collectTasks.toArray(new CompletableFuture[0])).get();
+ } catch (InterruptedException e) {
+ LOG.info("Interrupted while collecting session metrics", e);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) { // ExecutionException wraps the original exception
+ LOG.error("Exception while collecting session metrics", e.getCause());
+ throw new RuntimeException(e.getCause());
+ } finally {
+ collectorExecutor.shutdown();
+ }
+ }
+
+ public TezSessionPoolManagerMetrics start(HiveConf conf) {
+ if (metrics == null) {
+ LOG.warn(
+ "Metrics are not enabled, cannot start TezSessionPoolManagerMetrics (check {})",
+ HiveConf.ConfVars.HIVE_SERVER2_METRICS_ENABLED.varname);
+ return this;
+ }
+ this.mainExecutor = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder().setNameFormat("TezSessionPoolManagerMetrics worker thread").setDaemon(true).build());
+ long collectIntervalSeconds = HiveConf.getTimeVar(conf,
+ HiveConf.ConfVars.HIVE_SERVER2_TEZ_SESSIONS_METRICS_COLLECTION_INTERVAL, TimeUnit.SECONDS);
+ mainExecutor.scheduleAtFixedRate(this::collectMetrics, 0, collectIntervalSeconds, TimeUnit.SECONDS);
+ LOG.info("Starting TezSessionPoolManagerMetrics, collection interval: {}s", collectIntervalSeconds);
+
+ runningTasksCount.createGauge(metrics);
+ pendingTasksCount.createGauge(metrics);
+ taskBacklogRatio.createGauge(metrics);
+ for (TaskBacklogRatioMetric m : taskBacklogRatioDurations) {
+ m.createGauge(metrics);
+ }
+ clearMetrics(); //initial empty values for gauges
+ return this;
+ }
+
+ public void stop() {
+ if (mainExecutor != null) {
+ mainExecutor.shutdown();
+ mainExecutor = null;
+ }
+ }
+
+ void collectMetrics() {
+ List<TezSessionState> sessions = poolManager.getSessions();
+ LOG.debug("Updating metrics, session count: {}", sessions.size());
+
+ if (sessions.isEmpty()) {
+ // no active session, don't mess with the rest
+ clearMetrics();
+ return;
+ }
+
+ AtomicDouble currentRunningTasksCount = new AtomicDouble();
+ AtomicDouble currentPendingTasksCount = new AtomicDouble();
+ List<Future<?>> collectTasks = new ArrayList<>();
+
+ // run on maximum 10 threads
+ ExecutorService collectorExecutor = Executors.newFixedThreadPool(Math.min(sessions.size(), 10),
+ new ThreadFactoryBuilder().setNameFormat("TezSessionPoolManagerMetrics collector thread - #%d").build());
+ long start = Time.monotonicNow();
+
+ for (TezSessionState session : sessions) {
+ collectTasks.add(CompletableFuture.runAsync(() -> {
+ Map<String, Double> metrics = session.getMetrics();
+ LOG.debug("Achieved metrics from Tez session ({}): {}", session.getSessionId(), metrics);
+
+ currentRunningTasksCount.addAndGet(metrics.getOrDefault(runningTasksCount.name, 0.0));
+ currentPendingTasksCount.addAndGet(metrics.getOrDefault(pendingTasksCount.name, 0.0));
+ }, collectorExecutor));
+ }
+
+ waitForTasksToFinish(collectTasks, collectorExecutor);
+ long elapsed = Time.monotonicNow() - start;
+
+ refreshMetrics(currentRunningTasksCount, currentPendingTasksCount);
+
+ LOG.debug("Collected metrics from {} sessions (in {}ms): {}", sessions.size(), elapsed, toMetricsString());
+ }
+
+ @VisibleForTesting
+ void refreshMetrics(AtomicDouble currentRunningTasksCount, AtomicDouble currentPendingTasksCount) {
+ this.runningTasksCount.setValue((int) currentRunningTasksCount.get());
+ this.pendingTasksCount.setValue((int) currentPendingTasksCount.get());
+ double currentTaskBacklogRatio = (double) this.pendingTasksCount.value / (this.runningTasksCount.value + 1);
+ this.taskBacklogRatio.setValue(currentTaskBacklogRatio);
+
+ for (TaskBacklogRatioMetric m : taskBacklogRatioDurations) {
+ m.refresh(currentTaskBacklogRatio);
+ }
+ }
+
+ private void clearMetrics() {
+ runningTasksCount.setValue(0);
+ pendingTasksCount.setValue(0);
+ taskBacklogRatio.setValue(0.0);
+ Arrays.stream(taskBacklogRatioDurations).forEach(m -> m.timeSet = 0);
+ }
+
+ private String toMetricsString() {
+ return String.format("[running tasks: %d, pending tasks: %d, task backlog ratio: %.2f, task backlog ratios seen: " +
+ "{%s}]", this.runningTasksCount.value, this.pendingTasksCount.value, this.taskBacklogRatio.value,
+ getBackLogRatioDurationsString());
+ }
+
+ private String getBackLogRatioDurationsString() {
+ long now = Time.monotonicNow();
+ return Arrays.stream(taskBacklogRatioDurations).filter(m -> m.timeSet > 0)
+ .map(m -> m.name + ": " + ((now - m.timeSet) / 1000) + "s")
+ .collect(Collectors.joining(", "));
+ }
+
+ /**
+ * TezSessionMetric is a base adapter class for metrics to be exposed from simple primitive types to Hive metrics.
+ */
+ @VisibleForTesting
+ static class TezSessionMetric<T> {
+ String name;
+ T value;
+ long timeSet = 0L;
+
+ TezSessionMetric(String name, T value) {
+ this.name = name;
+ this.value = value;
+ }
+
+ public void createGauge(Metrics metrics) {
+ metrics.addGauge(name, (MetricsVariable<T>) () -> this.value);
+ }
+
+ public void setValue(T value) {
+ this.value = value;
+ this.timeSet = Time.monotonicNow();
+ }
+ }
+
+ /**
+ * TaskBacklogRatioMetric is a special type of metric where the key information is the time elapsed
+ * since the initial timeSet value was recorded (rather than a static value).
+ * Here's how it works:
+ * 1. TaskBacklogRatioMetric is instantiated with a specific ratio to monitor.
+ * 2. On every updateMetrics call, the refresh() method is invoked with the current task backlog ratio.
+ * 3. If currentTaskBacklogRatio is greater than or equal to the monitored ratio, it begins tracking time
+ * by setting timeSet to now()—but only if it hasn't been set already.
+ * 4. If currentTaskBacklogRatio drops below the monitored ratio, timeSet is reset to 0.
+ */
+ private static class TaskBacklogRatioMetric extends TezSessionMetric<Long> {
+ private final Integer ratio;
+
+ public TaskBacklogRatioMetric(String name, Integer ratio) {
+ super(name, 0L);
+ this.ratio = ratio;
+ }
+
+ public void createGauge(Metrics metrics) {
+ // The gauge returns timeSet - now() if timeSet > 0.
+ metrics.addGauge(name, (MetricsVariable<Long>) () -> (timeSet == 0) ? 0 :
+ ((Time.monotonicNow() - timeSet) / 1000));
+ }
+
+ public void refresh(double currentTaskBacklogRatio) {
+ if (currentTaskBacklogRatio >= this.ratio) {
+ if (timeSet == 0) { // this is the first time we see this backlog ratio
+ timeSet = Time.monotonicNow();
+ }
+ } else {
+ // current backlog ratio is lower than this threshold, let's clear
+ timeSet = 0;
+ }
+ }
+ }
+}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
index 1f56d081e1e..8b66454c51e 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
@@ -82,6 +82,8 @@
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezException;
import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.Progress;
import org.apache.tez.mapreduce.hadoop.DeprecatedKeys;
import org.apache.tez.mapreduce.hadoop.MRHelpers;
import org.apache.tez.mapreduce.hadoop.MRJobConfig;
@@ -156,6 +158,9 @@ public String toString() {
private KillQuery killQuery;
private static final Cache<String, String> shaCache = CacheBuilder.newBuilder().maximumSize(100).build();
+
+ private DAGStatus dagStatus;
+
/**
* Constructor. We do not automatically connect, because we only want to
* load tez classes when the user has tez installed.
@@ -1014,4 +1019,33 @@ public String getAppMasterUri() {
tezClient -> tezClient.getAmHost() + ":" + tezClient.getAmPort())
.get();
}
+
+ /**
+ * This method makes a best-effort attempt to retrieve data from the most recent dagStatus.
+ * An alternative approach would be for Tez to expose a dedicated metrics endpoint for this purpose.
+ * However, as long as TezJobMonitor continues polling the DAG states from the Tez AM, this method
+ * will remain a reliable solution.
+ *
+ * @return A map containing metrics for TezSessionPoolManagerMetrics.
+ */
+ public Map<String, Double> getMetrics() {
+ Map<String, Double> metrics = new HashMap<>();
+ if (dagStatus == null) {
+ return metrics;
+ }
+ Progress progress = dagStatus.getDAGProgress();
+ metrics.put(TezSessionPoolManagerMetrics.TEZ_SESSION_METRIC_RUNNING_TASKS, (double) progress.getRunningTaskCount());
+ // this logic for calculating pending tasks is inline with the one in TezProgressMonitor
+ int pendingTasks = progress.getTotalTaskCount() - progress.getSucceededTaskCount() - progress.getRunningTaskCount();
+ metrics.put(TezSessionPoolManagerMetrics.TEZ_SESSION_METRIC_PENDING_TASKS, (double) pendingTasks);
+ return metrics;
+ }
+
+ /**
+ * TezSessionState receives periodic updates from the current DAG's status.
+ * @param dagStatus status of the current DAG
+ */
+ public void updateDagStatus(DAGStatus dagStatus) {
+ this.dagStatus = dagStatus;
+ }
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
index 83bd35fbb40..662f774b7e1 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
@@ -277,7 +277,7 @@ public int execute() {
runtimeContext.setApplicationId(appId);
// finally monitor will print progress until the job is done
- monitor = new TezJobMonitor(work.getAllWork(), dagClient, conf, dag, ctx, runtimeContext.counters,
+ monitor = new TezJobMonitor(session, work.getAllWork(), dagClient, conf, dag, ctx, runtimeContext.counters,
perfLogger);
runtimeContext.setMonitor(monitor);
rc = monitor.monitorExecution();
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/TezJobMonitor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/TezJobMonitor.java
index db83a5c4e1f..0b99b0142d7 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/TezJobMonitor.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/TezJobMonitor.java
@@ -40,6 +40,7 @@ Licensed to the Apache Software Foundation (ASF) under one
import org.apache.hadoop.hive.ql.Context;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.tez.TezSessionPoolManager;
+import org.apache.hadoop.hive.ql.exec.tez.TezSessionState;
import org.apache.hadoop.hive.ql.exec.tez.Utils;
import org.apache.hadoop.hive.ql.log.PerfLogger;
import org.apache.hadoop.hive.ql.plan.BaseWork;
@@ -77,6 +78,7 @@ public class TezJobMonitor {
private static final int MAX_RETRY_INTERVAL = 2500;
private static final int MAX_RETRY_FAILURES = (MAX_RETRY_INTERVAL / MAX_CHECK_INTERVAL) + 1;
+ private final TezSessionState session;
private final PerfLogger perfLogger;
private static final List<DAGClient> shutdownList;
private final List<BaseWork> topSortedWorks;
@@ -118,8 +120,9 @@ public static void initShutdownHook() {
// compile time tez counters
private final TezCounters counters;
- public TezJobMonitor(List<BaseWork> topSortedWorks, final DAGClient dagClient, HiveConf conf, DAG dag,
- Context ctx, final TezCounters counters, PerfLogger perfLogger) {
+ public TezJobMonitor(TezSessionState session, List<BaseWork> topSortedWorks, final DAGClient dagClient, HiveConf conf,
+ DAG dag, Context ctx, final TezCounters counters, PerfLogger perfLogger) {
+ this.session = session;
this.topSortedWorks = topSortedWorks;
this.dagClient = dagClient;
this.hiveConf = conf;
@@ -185,6 +188,7 @@ public int monitorExecution() {
}
status = dagClient.getDAGStatus(opts, checkInterval);
+ session.updateDagStatus(status);
vertexProgressMap = status.getVertexProgress();
List<String> vertexNames = vertexProgressMap.keySet()
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPoolManagerMetrics.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPoolManagerMetrics.java
new file mode 100644
index 00000000000..3c79a34bee0
--- /dev/null
+++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPoolManagerMetrics.java
@@ -0,0 +1,110 @@
+/*
+ * 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.hadoop.hive.ql.exec.tez;
+
+import com.google.common.util.concurrent.AtomicDouble;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.IntStream;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestTezSessionPoolManagerMetrics {
+ private static final int NO_OF_SESSIONS = 2;
+ private static final double NO_OF_RUNNING_TASKS_PER_SESSION = 10.0;
+ private static final double NO_OF_PENDING_TASKS_PER_SESSION = 100.0;
+
+ /**
+ * This method tests the core metric calculation logic (refreshMetrics)
+ * without involving the actual metric collection process.
+ */
+ @Test
+ public void testRefreshMetrics(){
+ TezSessionPoolManager poolManager = mock(TezSessionPoolManager.class);
+ List<TezSessionState> sessions = testSessions();
+ when(poolManager.getSessions()).thenReturn(sessions);
+
+ TezSessionPoolManagerMetrics metrics = new TezSessionPoolManagerMetrics(poolManager);
+ metrics.refreshMetrics(new AtomicDouble(9.0), new AtomicDouble(1.0));
+
+ Assert.assertEquals(9.0, metrics.runningTasksCount.value, 0.0001);
+ Assert.assertEquals(1.0, metrics.pendingTasksCount.value, 0.0001);
+ Assert.assertEquals(0.1, metrics.taskBacklogRatio.value, 0.0001);
+ }
+
+ @Test
+ public void testBasicMetricsUpdate() {
+ TezSessionPoolManager poolManager = mock(TezSessionPoolManager.class);
+ List<TezSessionState> sessions = testSessions();
+ when(poolManager.getSessions()).thenReturn(sessions);
+
+ TezSessionPoolManagerMetrics metrics = new TezSessionPoolManagerMetrics(poolManager);
+ metrics.collectMetrics();
+
+ double expectedRunningTasks = NO_OF_SESSIONS * NO_OF_RUNNING_TASKS_PER_SESSION;
+ double expectedPendingTasks = NO_OF_SESSIONS * NO_OF_PENDING_TASKS_PER_SESSION;
+ double expectedBacklog = expectedPendingTasks / (expectedRunningTasks + 1);
+
+ Assert.assertEquals(expectedRunningTasks, metrics.runningTasksCount.value, 0.0001);
+ Assert.assertEquals(expectedPendingTasks, metrics.pendingTasksCount.value, 0.0001);
+ Assert.assertEquals(expectedBacklog, metrics.taskBacklogRatio.value, 0.0001);
+ }
+
+ @Test
+ public void testEmptyMetrics(){
+ TezSessionPoolManager poolManager = mock(TezSessionPoolManager.class);
+ when(poolManager.getSessions()).thenReturn(new ArrayList<>()); // no open sessions
+ TezSessionPoolManagerMetrics metrics = new TezSessionPoolManagerMetrics(poolManager);
+ metrics.collectMetrics();
+
+ Assert.assertEquals(0.0, metrics.runningTasksCount.value, 0.0001);
+ Assert.assertEquals(0.0, metrics.pendingTasksCount.value, 0.0001);
+ Assert.assertEquals(0.0, metrics.taskBacklogRatio.value, 0.0001);
+ }
+
+ @Test
+ public void testMetricsBeforeFirstUpdateAndStart(){
+ TezSessionPoolManager poolManager = mock(TezSessionPoolManager.class);
+ TezSessionPoolManagerMetrics metrics = new TezSessionPoolManagerMetrics(poolManager);
+
+ Assert.assertEquals(0.0, metrics.runningTasksCount.value, 0.0001);
+ Assert.assertEquals(0.0, metrics.pendingTasksCount.value, 0.0001);
+ Assert.assertEquals(0.0, metrics.taskBacklogRatio.value, 0.0001);
+ }
+
+ private List<TezSessionState> testSessions() {
+ List<TezSessionState> sessions = new ArrayList<>();
+ TezSessionState session = mock(TezSessionState.class);
+ when(session.getMetrics()).thenReturn(testMetrics());
+ IntStream.range(0, NO_OF_SESSIONS).forEach(i -> sessions.add(session));
+ return sessions;
+ }
+
+ private Map<String, Double> testMetrics() {
+ Map<String, Double> metrics = new HashMap<>();
+ metrics.put(TezSessionPoolManagerMetrics.TEZ_SESSION_METRIC_RUNNING_TASKS, NO_OF_RUNNING_TASKS_PER_SESSION);
+ metrics.put(TezSessionPoolManagerMetrics.TEZ_SESSION_METRIC_PENDING_TASKS, NO_OF_PENDING_TASKS_PER_SESSION);
+ return metrics;
+ }
+}